382 lines
14 KiB
Python
382 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""YIERP CLI：通过 URL、apikey 和 ApiSecret 访问 YIERP 数据接口。
|
||
|
||
本脚本适配 ERPNext / Frappe 风格的 REST API，适用于 YIERP（源自 ERPNext v16）。
|
||
支持的核心能力包括：
|
||
- 查询列表
|
||
- 获取单条记录
|
||
- 创建文档
|
||
- 更新文档
|
||
- 删除文档
|
||
- 调用自定义方法
|
||
- 连接检测
|
||
|
||
使用方法：
|
||
python tools/yierp_cli.py --help
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any, Dict, Optional
|
||
|
||
import requests
|
||
|
||
|
||
class YIERPConfigError(RuntimeError):
|
||
"""配置错误。"""
|
||
|
||
|
||
class YIERPRequestError(RuntimeError):
|
||
"""API 请求错误。"""
|
||
|
||
|
||
def _safe_json_loads(raw: Optional[str], label: str) -> Any:
|
||
"""安全解析 JSON 字符串。"""
|
||
if raw is None:
|
||
return None
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"{label} 不是合法的 JSON: {raw}") from exc
|
||
|
||
|
||
def _mask_value(value: Optional[str]) -> str:
|
||
"""掩码保护敏感字段。"""
|
||
if value is None:
|
||
return "未配置"
|
||
if len(value) <= 6:
|
||
return "******"
|
||
return f"{value[:2]}******{value[-2:]}"
|
||
|
||
|
||
def _parse_env_value(value: str) -> str:
|
||
"""去除 .env 值两侧空白与包裹的引号。"""
|
||
value = value.strip()
|
||
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
|
||
value = value[1:-1]
|
||
return value
|
||
|
||
|
||
def _iter_dotenv_candidates() -> list:
|
||
"""从当前工作目录向上逐级查找 .env，遇到 .git 目录即停止上溯。"""
|
||
candidates = []
|
||
cwd = Path.cwd().resolve()
|
||
candidates.append(cwd / ".env")
|
||
for parent in cwd.parents:
|
||
candidates.append(parent / ".env")
|
||
if (parent / ".git").exists():
|
||
break
|
||
return candidates
|
||
|
||
|
||
def load_dotenv() -> None:
|
||
"""加载 .env 到环境变量（不覆盖已存在的环境变量）。"""
|
||
for env_path in _iter_dotenv_candidates():
|
||
if not env_path.is_file():
|
||
continue
|
||
try:
|
||
text = env_path.read_text(encoding="utf-8")
|
||
except (OSError, UnicodeDecodeError):
|
||
continue
|
||
for line in text.splitlines():
|
||
stripped = line.strip()
|
||
if not stripped or stripped.startswith("#") or "=" not in stripped:
|
||
continue
|
||
if stripped.startswith("export "):
|
||
stripped = stripped[len("export "):].strip()
|
||
key, _, raw_value = stripped.partition("=")
|
||
key = key.strip()
|
||
if not key or key in os.environ:
|
||
continue
|
||
os.environ[key] = _parse_env_value(raw_value)
|
||
break
|
||
|
||
|
||
def _load_json_file(file_path: str) -> Dict[str, Any]:
|
||
"""读取 JSON 配置文件。"""
|
||
path = Path(file_path)
|
||
if not path.exists():
|
||
return {}
|
||
try:
|
||
with path.open("r", encoding="utf-8") as handle:
|
||
data = json.load(handle)
|
||
if isinstance(data, dict):
|
||
return data
|
||
raise YIERPConfigError(f"配置文件不是有效的 JSON 对象：{file_path}")
|
||
except json.JSONDecodeError as exc:
|
||
raise YIERPConfigError(f"配置文件 JSON 格式错误：{file_path}") from exc
|
||
|
||
|
||
def resolve_config(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""按优先级解析配置：命令行参数 > 环境变量（含 .env 文件）> 配置文件。"""
|
||
load_dotenv()
|
||
config_path = args.config or os.getenv("YIERP_CONFIG_PATH", ".yierp/config.json")
|
||
config_data: Dict[str, Any] = _load_json_file(config_path)
|
||
|
||
data: Dict[str, Any] = dict(config_data)
|
||
|
||
data["url"] = args.url or os.getenv("YIERP_URL") or data.get("url")
|
||
data["apikey"] = (
|
||
args.apikey
|
||
or os.getenv("YIERP_APIKEY")
|
||
or data.get("apikey")
|
||
or data.get("api_key")
|
||
or data.get("apiKey")
|
||
)
|
||
data["ApiSecret"] = (
|
||
args.apisecret
|
||
or os.getenv("YIERP_APISECRET")
|
||
or data.get("ApiSecret")
|
||
or data.get("api_secret")
|
||
or data.get("apiSecret")
|
||
)
|
||
data["timeout"] = args.timeout or data.get("timeout", 30)
|
||
data["verify_ssl"] = data.get("verify_ssl", True)
|
||
|
||
if args.no_verify_ssl:
|
||
data["verify_ssl"] = False
|
||
|
||
if not data.get("url"):
|
||
raise YIERPConfigError("缺少配置：URL 未设置。请配置 url 或设置 YIERP_URL。")
|
||
if not data.get("apikey"):
|
||
raise YIERPConfigError("缺少配置：apikey 未设置。请配置 apikey 或设置 YIERP_APIKEY。")
|
||
if not data.get("ApiSecret"):
|
||
raise YIERPConfigError("缺少配置：ApiSecret 未设置。请配置 ApiSecret 或设置 YIERP_APISECRET。")
|
||
|
||
return data
|
||
|
||
|
||
def ensure_headers(config: Dict[str, Any]) -> Dict[str, str]:
|
||
"""构造 Frappe / ERPNext 的认证头。"""
|
||
auth_value = f"token {config['apikey']}:{config['ApiSecret']}"
|
||
return {
|
||
"Authorization": auth_value,
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
|
||
def request_json(
|
||
config: Dict[str, Any],
|
||
method: str,
|
||
path: str,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
payload: Optional[Dict[str, Any]] = None,
|
||
) -> Any:
|
||
"""统一请求 YIERP REST API。"""
|
||
url = f"{config['url'].rstrip('/')}{path}"
|
||
headers = ensure_headers(config)
|
||
|
||
try:
|
||
response = requests.request(
|
||
method=method.upper(),
|
||
url=url,
|
||
headers=headers,
|
||
params=params,
|
||
json=payload,
|
||
timeout=config.get("timeout", 30),
|
||
verify=config.get("verify_ssl", True),
|
||
)
|
||
except requests.RequestException as exc:
|
||
raise YIERPRequestError(f"请求失败：{exc}") from exc
|
||
|
||
text = response.text.strip()
|
||
try:
|
||
payload_json = response.json() if text else {}
|
||
except ValueError:
|
||
payload_json = {"raw": text}
|
||
|
||
if response.status_code >= 400:
|
||
message = payload_json.get("exc") or payload_json.get("exception") or payload_json.get("message") or text or "未知错误"
|
||
raise YIERPRequestError(f"HTTP {response.status_code}：{message}")
|
||
|
||
return payload_json
|
||
|
||
|
||
def parse_json_argument(value: Optional[str], label: str) -> Optional[Any]:
|
||
"""将命令行 JSON 字符串解析为 Python 对象。"""
|
||
if value is None:
|
||
return None
|
||
return _safe_json_loads(value, label)
|
||
|
||
|
||
def build_list_params(filters: Optional[Any], fields: Optional[Any], limit: int, start: int, order_by: Optional[str]) -> Dict[str, Any]:
|
||
"""构造列表查询参数。"""
|
||
params: Dict[str, Any] = {}
|
||
if filters is not None:
|
||
params["filters"] = json.dumps(filters)
|
||
if fields is not None:
|
||
params["fields"] = json.dumps(fields)
|
||
if order_by:
|
||
params["order_by"] = order_by
|
||
if limit is not None:
|
||
params["limit_page_length"] = limit
|
||
if start is not None:
|
||
params["limit_start"] = start
|
||
return params
|
||
|
||
|
||
def cmd_whoami(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""检查当前登录用户。"""
|
||
config = resolve_config(args)
|
||
return request_json(config, "GET", "/api/method/frappe.auth.get_logged_user")
|
||
|
||
|
||
def cmd_list(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""查询文档列表。"""
|
||
config = resolve_config(args)
|
||
doctype = args.doctype
|
||
filters = parse_json_argument(args.filters, "filters")
|
||
fields = parse_json_argument(args.fields, "fields")
|
||
params = build_list_params(filters, fields, args.limit, args.start, args.order_by)
|
||
result = request_json(config, "GET", f"/api/resource/{doctype}", params=params)
|
||
return result
|
||
|
||
|
||
def cmd_get(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""获取单条文档。"""
|
||
config = resolve_config(args)
|
||
doctype = args.doctype
|
||
name = args.name
|
||
return request_json(config, "GET", f"/api/resource/{doctype}/{name}")
|
||
|
||
|
||
def cmd_create(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""创建单条文档。"""
|
||
config = resolve_config(args)
|
||
doctype = args.doctype
|
||
payload = parse_json_argument(args.data, "data")
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("创建数据必须是 JSON 对象。")
|
||
return request_json(config, "POST", f"/api/resource/{doctype}", payload=payload)
|
||
|
||
|
||
def cmd_update(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""更新单条文档。"""
|
||
config = resolve_config(args)
|
||
doctype = args.doctype
|
||
name = args.name
|
||
payload = parse_json_argument(args.data, "data")
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("更新数据必须是 JSON 对象。")
|
||
return request_json(config, "PUT", f"/api/resource/{doctype}/{name}", payload=payload)
|
||
|
||
|
||
def cmd_delete(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""删除单条文档。"""
|
||
config = resolve_config(args)
|
||
doctype = args.doctype
|
||
name = args.name
|
||
if not args.yes:
|
||
raise YIERPRequestError("删除操作需要显式添加 --yes 参数确认。")
|
||
return request_json(config, "DELETE", f"/api/resource/{doctype}/{name}")
|
||
|
||
|
||
def cmd_method(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""调用自定义的挂钩方法或后端方法。"""
|
||
config = resolve_config(args)
|
||
method_name = args.method
|
||
payload = parse_json_argument(args.args, "args")
|
||
if payload is None:
|
||
payload = {}
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("方法参数必须是 JSON 对象。")
|
||
return request_json(config, "POST", f"/api/method/{method_name}", payload=payload)
|
||
|
||
|
||
def cmd_config_show(args: argparse.Namespace) -> Dict[str, Any]:
|
||
"""显示当前配置摘要，隐藏敏感字段。"""
|
||
config = resolve_config(args)
|
||
return {
|
||
"url": config.get("url"),
|
||
"apikey": _mask_value(config.get("apikey")),
|
||
"ApiSecret": _mask_value(config.get("ApiSecret")),
|
||
"timeout": config.get("timeout", 30),
|
||
"verify_ssl": config.get("verify_ssl", True),
|
||
}
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
"""构建命令行解释器。"""
|
||
parser = argparse.ArgumentParser(description="YIERP CLI：基于 ERPNext v16 REST API 的数据操作工具", add_help=True)
|
||
parser.add_argument("--config", default=None, help="配置文件路径，默认 .yierp/config.json")
|
||
parser.add_argument("--url", default=None, help="YIERP 站点地址，例如 https://yierp.example.com")
|
||
parser.add_argument("--apikey", default=None, help="YIERP API Key")
|
||
parser.add_argument("--apisecret", default=None, help="YIERP ApiSecret")
|
||
parser.add_argument("--timeout", type=int, default=None, help="请求超时时间，单位秒")
|
||
parser.add_argument("--no-verify-ssl", action="store_true", help="跳过 SSL 证书校验，适用于自签名证书场景")
|
||
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
|
||
whoami_parser = subparsers.add_parser("whoami", help="查看当前登录用户")
|
||
whoami_parser.set_defaults(func=cmd_whoami)
|
||
|
||
config_parser = subparsers.add_parser("config", help="显示当前配置摘要")
|
||
config_parser.set_defaults(func=cmd_config_show)
|
||
|
||
list_parser = subparsers.add_parser("list", help="查询数据列表")
|
||
list_parser.add_argument("--doctype", required=True, help="DocType，例如 ToDo、Sales Invoice、Customer")
|
||
list_parser.add_argument("--filters", default=None, help="筛选条件，JSON 字符串，例如 '{\"status\":\"Open\"}'")
|
||
list_parser.add_argument("--fields", default=None, help="查询字段，JSON 数组，例如 '[\"name\",\"status\"]'")
|
||
list_parser.add_argument("--limit", type=int, default=20, help="返回条数上限")
|
||
list_parser.add_argument("--start", type=int, default=0, help="分页起始位置")
|
||
list_parser.add_argument("--order-by", default=None, help="排序字段，例如 name desc")
|
||
list_parser.set_defaults(func=cmd_list)
|
||
|
||
get_parser = subparsers.add_parser("get", help="获取单条记录")
|
||
get_parser.add_argument("--doctype", required=True, help="DocType")
|
||
get_parser.add_argument("--name", required=True, help="文档名称或主键值")
|
||
get_parser.set_defaults(func=cmd_get)
|
||
|
||
create_parser = subparsers.add_parser("create", help="创建单条文档")
|
||
create_parser.add_argument("--doctype", required=True, help="DocType")
|
||
create_parser.add_argument("--data", required=True, help="创建数据，JSON 对象，例如 '{\"description\":\"新任务\"}'")
|
||
create_parser.set_defaults(func=cmd_create)
|
||
|
||
update_parser = subparsers.add_parser("update", help="更新单条文档")
|
||
update_parser.add_argument("--doctype", required=True, help="DocType")
|
||
update_parser.add_argument("--name", required=True, help="文档名称或主键值")
|
||
update_parser.add_argument("--data", required=True, help="更新数据，JSON 对象，例如 '{\"status\":\"Closed\"}'")
|
||
update_parser.set_defaults(func=cmd_update)
|
||
|
||
delete_parser = subparsers.add_parser("delete", help="删除单条文档")
|
||
delete_parser.add_argument("--doctype", required=True, help="DocType")
|
||
delete_parser.add_argument("--name", required=True, help="文档名称或主键值")
|
||
delete_parser.add_argument("--yes", action="store_true", help="确认执行删除操作")
|
||
delete_parser.set_defaults(func=cmd_delete)
|
||
|
||
method_parser = subparsers.add_parser("method", help="调用后端方法")
|
||
method_parser.add_argument("--method", required=True, help="方法路径，例如 your_app.api.test")
|
||
method_parser.add_argument("--args", default=None, help="方法参数，JSON 对象，例如 '{\"message\":\"hello\"}'")
|
||
method_parser.set_defaults(func=cmd_method)
|
||
|
||
return parser
|
||
|
||
|
||
def main() -> int:
|
||
"""主函数入口。"""
|
||
parser = build_parser()
|
||
args = parser.parse_args()
|
||
|
||
try:
|
||
result = args.func(args)
|
||
except (YIERPConfigError, YIERPRequestError, ValueError) as exc:
|
||
print(f"错误：{exc}", file=sys.stderr)
|
||
return 1
|
||
except Exception as exc: # pragma: no cover
|
||
print(f"未处理错误：{exc}", file=sys.stderr)
|
||
return 1
|
||
|
||
print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|