111 lines
4.8 KiB
Python
111 lines
4.8 KiB
Python
"""Standalone Kingdee reader restricted to material codes and names."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import getpass
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
BASE_URL = "https://kingdeeapi.zeroerr.com"
|
|
LOCAL_CREDENTIALS_PATH = Path(__file__).resolve().parents[2] / "configs/material_cache_erp.local"
|
|
LOGIN_PATH = "/Kingdee.BOS.WebApi.ServicesStub.AuthService.ValidateUser.common.kdsvc"
|
|
QUERY_PATH = "/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.ExecuteBillQuery.common.kdsvc"
|
|
|
|
|
|
class KingdeeMaterialError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def credentials() -> dict[str, str]:
|
|
result = {key: os.environ.get(env, "") for key, env in (
|
|
("acct_id", "KINGDEE_ACCT_ID"), ("username", "KINGDEE_USERNAME"), ("password", "KINGDEE_PASSWORD"),
|
|
)}
|
|
if any(result.values()):
|
|
if not all(result.values()):
|
|
raise KingdeeMaterialError("金蝶运行环境凭证配置不完整。")
|
|
return result
|
|
if LOCAL_CREDENTIALS_PATH.exists():
|
|
try:
|
|
data = json.loads(LOCAL_CREDENTIALS_PATH.read_text(encoding="utf-8"))
|
|
result = {key: data.get(key) for key in result}
|
|
if not all(isinstance(value, str) and value for value in result.values()):
|
|
raise ValueError
|
|
return result
|
|
except (OSError, ValueError, AttributeError) as exc:
|
|
raise KingdeeMaterialError("本机金蝶凭证文件配置无效。") from exc
|
|
account = getpass.getuser()
|
|
for key, service in (
|
|
("acct_id", "TH51_KINGDEE_ACCT_ID"), ("username", "TH51_KINGDEE_USERNAME"), ("password", "TH51_KINGDEE_PASSWORD"),
|
|
):
|
|
try:
|
|
completed = subprocess.run(
|
|
["security", "find-generic-password", "-a", account, "-s", service, "-w"],
|
|
capture_output=True, text=True, timeout=15,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise KingdeeMaterialError("本机金蝶只读凭证不可用。") from exc
|
|
if completed.returncode:
|
|
raise KingdeeMaterialError("本机未配置金蝶只读凭证。")
|
|
result[key] = completed.stdout.rstrip("\r\n")
|
|
# macOS may return non-ASCII Keychain text as UTF-8 hexadecimal bytes.
|
|
for key, value in result.items():
|
|
try:
|
|
decoded = bytes.fromhex(value).decode("utf-8")
|
|
except (ValueError, UnicodeError):
|
|
continue
|
|
if any(ord(char) > 127 for char in decoded):
|
|
result[key] = decoded
|
|
return result
|
|
|
|
|
|
def build_opener() -> urllib.request.OpenerDirector:
|
|
return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
|
|
|
|
|
def post(opener: Any, path: str, payload: Any) -> Any:
|
|
request = urllib.request.Request(
|
|
BASE_URL + path, data=json.dumps(payload, ensure_ascii=True).encode("utf-8"),
|
|
headers={"Content-Type": "application/json; charset=utf-8"}, method="POST",
|
|
)
|
|
try:
|
|
with opener.open(request, timeout=30) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as exc:
|
|
raise KingdeeMaterialError(f"金蝶只读接口返回 HTTP {exc.code}。") from exc
|
|
except (urllib.error.URLError, OSError) as exc:
|
|
raise KingdeeMaterialError("金蝶只读接口网络或 TLS 连接失败。") from exc
|
|
except (ValueError, UnicodeError) as exc:
|
|
raise KingdeeMaterialError("金蝶只读接口响应格式无效。") from exc
|
|
|
|
|
|
def login(opener: Any) -> None:
|
|
values = credentials()
|
|
result = post(opener, LOGIN_PATH, {
|
|
"acctID": values["acct_id"], "username": values["username"],
|
|
"password": values["password"], "lcid": 2052,
|
|
})
|
|
if not isinstance(result, dict) or not (result.get("IsSuccess") is True or result.get("LoginResultType") == 1):
|
|
raise KingdeeMaterialError("金蝶登录失败，请检查本机只读账号和凭证。")
|
|
|
|
|
|
def execute_list_query(opener: Any, args: Any) -> list[Any]:
|
|
if args.form_id != "BD_MATERIAL" or args.fields != "FNumber,FName" or args.filter:
|
|
raise KingdeeMaterialError("物料名称客户端只允许未筛选的 BD_MATERIAL 编码和名称查询。")
|
|
parameters = {
|
|
"FormId": "BD_MATERIAL", "FieldKeys": "FNumber,FName", "FilterString": "",
|
|
"OrderString": "FNumber ASC,FMaterialId ASC", "TopRowCount": 0,
|
|
"StartRow": args.start_row, "Limit": args.limit,
|
|
}
|
|
for payload in (parameters, {"data": parameters}):
|
|
result = post(opener, QUERY_PATH, payload)
|
|
if isinstance(result, list) and all(isinstance(row, list) and len(row) == 2 for row in result):
|
|
return result
|
|
raise KingdeeMaterialError("金蝶物料名称查询未返回有效数据，请核对只读查询权限。")
|