diff --git a/docs/business-facts.md b/docs/business-facts.md index afae718..eb9e51d 100644 --- a/docs/business-facts.md +++ b/docs/business-facts.md @@ -39,6 +39,10 @@ The config template includes: - `KINGDEE_SALORDER_API_URL`: defaults to the official `querySalOrder` endpoint. - `KINGDEE_TOKEN_URL` and `KINGDEE_TOKEN_BODY_JSON`: login/access-token request. +- `KINGDEE_APP_KEY`, `KINGDEE_APP_SECRET`, `KINGDEE_TOKEN_APP_KEY_FIELD`, `KINGDEE_TOKEN_APP_SECRET_FIELD`: preferred split-field way to build the token request body without hand-writing the full JSON. +- `KINGDEE_TOKEN_TIMESTAMP_FIELD`, `KINGDEE_TOKEN_TIMESTAMP_FORMAT`: optional current-time field for token APIs that require a timestamp. Supported formats are `epoch_ms`, `epoch_seconds`, `datetime_utc`, `iso_utc`, and `date_utc`. +- `KINGDEE_TOKEN_BODY_EXTRA_JSON`: optional fixed extra fields merged into the generated token request body. +- `KINGDEE_TOKEN_BODY_JSON`: advanced override for the full token request JSON. It supports `{now_ms}`, `{now_epoch}`, `{now_seconds}`, `{now_datetime_utc}`, `{now_iso_utc}`, and `{today_utc}` placeholders. - `KINGDEE_TOKEN_HEADERS_JSON`: optional token request headers, such as app-specific headers. - `KINGDEE_TOKEN_FIELD`, `KINGDEE_TOKEN_EXPIRES_IN_FIELD`, `KINGDEE_REFRESH_TOKEN_FIELD`: dot paths used to read token response fields. - `KINGDEE_REFRESH_URL` and `KINGDEE_REFRESH_BODY_JSON`: optional refresh request. Use `{refresh_token}` in the JSON body template. diff --git a/obsidian_wiki/kingdee_api.py b/obsidian_wiki/kingdee_api.py index 93a43c4..fec986e 100644 --- a/obsidian_wiki/kingdee_api.py +++ b/obsidian_wiki/kingdee_api.py @@ -8,6 +8,7 @@ import stat import time import uuid import urllib.request +from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable @@ -43,9 +44,50 @@ KINGDEE_ACCESS_TOKEN= # 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。 KINGDEE_TOKEN_URL= +# 推荐填法：用下面这些拆分变量让程序自动拼 token JSON body。 +# 如果你的 token 接口字段名不是 app_key/app_secret/timestamp，就改对应 FIELD。 + +# 变量：KINGDEE_APP_KEY +# 用途：金蝶 OpenAPI 应用的 app key/client id。 +# 是否必填：自动登录时通常必填；具体以金蝶 token 文档为准。 +KINGDEE_APP_KEY= + +# 变量：KINGDEE_APP_SECRET +# 用途：金蝶 OpenAPI 应用的 app secret/client secret。 +# 是否必填：自动登录时通常必填；具体以金蝶 token 文档为准。 +KINGDEE_APP_SECRET= + +# 变量：KINGDEE_TOKEN_APP_KEY_FIELD +# 用途：token JSON body 中 app key 对应的字段名。 +# 是否必填：可选，默认 app_key；如果金蝶文档写 client_id，就改成 client_id。 +KINGDEE_TOKEN_APP_KEY_FIELD=app_key + +# 变量：KINGDEE_TOKEN_APP_SECRET_FIELD +# 用途：token JSON body 中 app secret 对应的字段名。 +# 是否必填：可选，默认 app_secret；如果金蝶文档写 client_secret，就改成 client_secret。 +KINGDEE_TOKEN_APP_SECRET_FIELD=app_secret + +# 变量：KINGDEE_TOKEN_TIMESTAMP_FIELD +# 用途：token JSON body 中当前时间/时间戳对应的字段名；留空则不发送时间。 +# 是否必填：仅当金蝶 token 文档要求当前时间时填写，例如 timestamp。 +KINGDEE_TOKEN_TIMESTAMP_FIELD= + +# 变量：KINGDEE_TOKEN_TIMESTAMP_FORMAT +# 用途：当前时间格式。可选值：epoch_ms、epoch_seconds、datetime_utc、iso_utc、date_utc。 +# 是否必填：可选，默认 epoch_ms。 +KINGDEE_TOKEN_TIMESTAMP_FORMAT=epoch_ms + +# 变量：KINGDEE_TOKEN_BODY_EXTRA_JSON +# 用途：token JSON body 的额外固定字段，例如租户、账套等；会合并到自动拼出的 body。 +# 是否必填：可选；没有额外字段时保持 {{}}。 +KINGDEE_TOKEN_BODY_EXTRA_JSON={{}} + +# 高级填法：如果 token body 结构特殊，可直接填写完整 JSON。 +# 填了 KINGDEE_TOKEN_BODY_JSON 后，会覆盖上面的拆分变量。 +# 支持占位符：{{now_ms}}、{{now_epoch}}、{{now_seconds}}、{{now_datetime_utc}}、{{now_iso_utc}}、{{today_utc}}。 # 变量：KINGDEE_TOKEN_BODY_JSON -# 用途：调用登录/token 接口时发送的 JSON 请求体，例如 app key、app secret、租户等。 -# 是否必填：自动登录时必填；必须是一行合法 JSON。 +# 用途：调用登录/token 接口时发送的完整 JSON 请求体。 +# 是否必填：可选；只有不用拆分变量时才填写，必须是一行合法 JSON。 KINGDEE_TOKEN_BODY_JSON= # 变量：KINGDEE_TOKEN_HEADERS_JSON @@ -118,6 +160,10 @@ def load_kingdee_config(config_path: str | Path | None = None) -> dict[str, str] values[new] = values[old] values.setdefault("KINGDEE_SALORDER_API_URL", DEFAULT_SALORDER_API_URL) values.setdefault("KINGDEE_TOKEN_HEADERS_JSON", "{}") + values.setdefault("KINGDEE_TOKEN_APP_KEY_FIELD", "app_key") + values.setdefault("KINGDEE_TOKEN_APP_SECRET_FIELD", "app_secret") + values.setdefault("KINGDEE_TOKEN_TIMESTAMP_FORMAT", "epoch_ms") + values.setdefault("KINGDEE_TOKEN_BODY_EXTRA_JSON", "{}") values.setdefault("KINGDEE_TOKEN_FIELD", "data.access_token") values.setdefault("KINGDEE_TOKEN_EXPIRES_IN_FIELD", "data.expires_in") values.setdefault("KINGDEE_REFRESH_TOKEN_FIELD", "data.refresh_token") @@ -153,7 +199,10 @@ def get_kingdee_access_token( try: token_payload = _request_token( config["KINGDEE_REFRESH_URL"], - _render_json_template(config.get("KINGDEE_REFRESH_BODY_JSON", ""), {"refresh_token": refresh_token}), + _render_json_template( + config.get("KINGDEE_REFRESH_BODY_JSON", ""), + _template_values(now, {"refresh_token": refresh_token}), + ), config.get("KINGDEE_TOKEN_HEADERS_JSON", "{}"), urlopen=urlopen, ) @@ -166,7 +215,7 @@ def get_kingdee_access_token( raise RuntimeError("KINGDEE_TOKEN_URL is required when no cached KINGDEE_ACCESS_TOKEN is available") token_payload = _request_token( token_url, - config.get("KINGDEE_TOKEN_BODY_JSON", ""), + _build_token_body_json(config, now), config.get("KINGDEE_TOKEN_HEADERS_JSON", "{}"), urlopen=urlopen, ) @@ -289,6 +338,62 @@ def _request_token( return payload +def _build_token_body_json(config: dict[str, str], now_epoch: float) -> str: + body_json = config.get("KINGDEE_TOKEN_BODY_JSON", "") + if body_json: + return _render_json_template(body_json, _template_values(now_epoch)) + + body: dict[str, Any] = {} + app_key = config.get("KINGDEE_APP_KEY", "") + app_secret = config.get("KINGDEE_APP_SECRET", "") + app_key_field = config.get("KINGDEE_TOKEN_APP_KEY_FIELD") or "app_key" + app_secret_field = config.get("KINGDEE_TOKEN_APP_SECRET_FIELD") or "app_secret" + if app_key: + body[app_key_field] = app_key + if app_secret: + body[app_secret_field] = app_secret + + timestamp_field = config.get("KINGDEE_TOKEN_TIMESTAMP_FIELD", "") + if timestamp_field: + body[timestamp_field] = _format_timestamp(now_epoch, config.get("KINGDEE_TOKEN_TIMESTAMP_FORMAT", "epoch_ms")) + + extra = json.loads(config.get("KINGDEE_TOKEN_BODY_EXTRA_JSON", "{}") or "{}") + if not isinstance(extra, dict): + raise RuntimeError("KINGDEE_TOKEN_BODY_EXTRA_JSON must be a JSON object") + body.update(extra) + return json.dumps(body, ensure_ascii=False, separators=(",", ":")) + + +def _template_values(now_epoch: float, extra: dict[str, str] | None = None) -> dict[str, str]: + dt = datetime.fromtimestamp(now_epoch, tz=timezone.utc) + values = { + "now_ms": str(int(now_epoch * 1000)), + "now_epoch": str(int(now_epoch)), + "now_seconds": str(int(now_epoch)), + "now_datetime_utc": dt.strftime("%Y-%m-%d %H:%M:%S"), + "now_iso_utc": dt.replace(microsecond=0).isoformat().replace("+00:00", "Z"), + "today_utc": dt.strftime("%Y-%m-%d"), + } + values.update(extra or {}) + return values + + +def _format_timestamp(now_epoch: float, timestamp_format: str) -> str | int: + normalized = (timestamp_format or "epoch_ms").strip().lower() + dt = datetime.fromtimestamp(now_epoch, tz=timezone.utc) + if normalized in {"epoch_ms", "milliseconds", "millis"}: + return int(now_epoch * 1000) + if normalized in {"epoch_seconds", "seconds", "unix"}: + return int(now_epoch) + if normalized == "datetime_utc": + return dt.strftime("%Y-%m-%d %H:%M:%S") + if normalized == "iso_utc": + return dt.replace(microsecond=0).isoformat().replace("+00:00", "Z") + if normalized == "date_utc": + return dt.strftime("%Y-%m-%d") + raise RuntimeError(f"unsupported KINGDEE_TOKEN_TIMESTAMP_FORMAT: {timestamp_format}") + + def _render_json_template(template: str, values: dict[str, str]) -> str: if not template: return "{}" diff --git a/tests/test_kingdee_api.py b/tests/test_kingdee_api.py index 1e85e90..0d266aa 100644 --- a/tests/test_kingdee_api.py +++ b/tests/test_kingdee_api.py @@ -67,6 +67,9 @@ class KingdeeApiTests(unittest.TestCase): self.assertIn("用途：销售订单 API 要求的 x-acgw-identity 请求头。", text) self.assertIn("是否必填：可选，手动测试用；长期定时同步建议使用自动登录配置。", text) self.assertIn("是否必填：可选，默认从 data.access_token 读取。", text) + self.assertIn("变量：KINGDEE_APP_KEY", text) + self.assertIn("变量：KINGDEE_TOKEN_TIMESTAMP_FIELD", text) + self.assertIn("支持占位符：{now_ms}", text) def test_default_config_path_accepts_none(self): cwd = Path.cwd() @@ -126,6 +129,45 @@ class KingdeeApiTests(unittest.TestCase): self.assertEqual(request.get_header("X-app"), "demo") self.assertEqual(stat.S_IMODE(cache_path.stat().st_mode), 0o600) + def test_access_token_body_can_be_built_from_split_fields_with_current_time(self): + cache_path = self.root / ".runtime" / "kingdee-token.json" + config = { + "KINGDEE_TOKEN_URL": "https://token.example/access", + "KINGDEE_APP_KEY": "abc", + "KINGDEE_APP_SECRET": "secret", + "KINGDEE_TOKEN_APP_KEY_FIELD": "client_id", + "KINGDEE_TOKEN_APP_SECRET_FIELD": "client_secret", + "KINGDEE_TOKEN_TIMESTAMP_FIELD": "timestamp", + "KINGDEE_TOKEN_TIMESTAMP_FORMAT": "epoch_ms", + "KINGDEE_TOKEN_BODY_EXTRA_JSON": '{"tenant":"demo"}', + "KINGDEE_TOKEN_FIELD": "data.access_token", + } + http = RecordingUrlopen([{"data": {"access_token": "access-1"}}]) + + token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000.5) + + self.assertEqual(token, "access-1") + body = json.loads(http.calls[0]["request"].data.decode("utf-8")) + self.assertEqual(body["client_id"], "abc") + self.assertEqual(body["client_secret"], "secret") + self.assertEqual(body["timestamp"], 1000500) + self.assertEqual(body["tenant"], "demo") + + def test_token_body_json_supports_current_time_placeholders(self): + cache_path = self.root / ".runtime" / "kingdee-token.json" + config = { + "KINGDEE_TOKEN_URL": "https://token.example/access", + "KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc","timestamp":{now_ms},"requested_at":"{now_datetime_utc}"}', + "KINGDEE_TOKEN_FIELD": "data.access_token", + } + http = RecordingUrlopen([{"data": {"access_token": "access-1"}}]) + + get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000) + + body = json.loads(http.calls[0]["request"].data.decode("utf-8")) + self.assertEqual(body["timestamp"], 1000000) + self.assertEqual(body["requested_at"], "1970-01-01 00:16:40") + def test_expired_token_uses_refresh_endpoint(self): cache_path = self.root / ".runtime" / "kingdee-token.json" cache_path.parent.mkdir(parents=True)