gitea-admin/obsidian-wiki
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(business): follow Kingdee token request body

Browse Source 
Align the generated Kingdee token request with the official enhanced-token document: client_id, client_secret, username, accountId, language, automatic nonce, and automatic current timestamp. Treat expires_in as milliseconds by default and update docs/tests accordingly.

Validation: python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests; git diff --check.
This commit is contained in:
Codex
2026-07-06 20:11:01 +00:00
parent 4e34e5b3fe
commit 72c666f165
4 changed files with 108 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/business-facts.md
+6 -5
View File
@@ -37,14 +37,15 @@ When `--input` is omitted, sync reads `.runtime/kingdee.env`, gets or refreshes
The config template includes:
- Token defaults follow Kingdee's enhanced OpenAPI token authentication doc: `https://vip.kingdee.com/knowledge/specialDetail/226337046514476288?category=239331354741842688&id=489812471545485056&productLineId=29`.
- `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_URL`: defaults to the official enhanced-token endpoint `/kapi/oauth2/getToken`.
- `KINGDEE_CLIENT_ID`, `KINGDEE_CLIENT_SECRET`, `KINGDEE_USERNAME`, `KINGDEE_ACCOUNT_ID`, `KINGDEE_LANGUAGE`: fixed fields used to build the official getToken request body.
- `nonce` and `timestamp`: generated automatically for every token request; `timestamp` uses the official zero-time-zone format by default.
- `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_BODY_JSON`: advanced override for nonstandard token request bodies.
- `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_TOKEN_FIELD`, `KINGDEE_TOKEN_EXPIRES_IN_FIELD`, `KINGDEE_TOKEN_EXPIRES_IN_UNIT`, `KINGDEE_REFRESH_TOKEN_FIELD`: dot paths and units used to read token response fields. The official `expires_in` unit is milliseconds.
- `KINGDEE_REFRESH_URL` and `KINGDEE_REFRESH_BODY_JSON`: optional refresh request. Use `{refresh_token}` in the JSON body template.
- `KINGDEE_ACGW_IDENTITY`: value sent as the required `x-acgw-identity` sales-order API header.
- `KINGDEE_PAGE_SIZE`: default sales-order page size.
obsidian_wiki/cli.py
+4 -1
View File
@@ -518,7 +518,10 @@ def cmd_business_sync_kingdee(args: argparse.Namespace) -> int:
overrides = _kingdee_cli_overrides(args)
config = load_kingdee_config(config_path)
config.update(overrides)
if created and not (config.get("KINGDEE_ACCESS_TOKEN") or config.get("KINGDEE_TOKEN_URL")):
has_official_token_credentials = all(
config.get(key) for key in ("KINGDEE_CLIENT_ID", "KINGDEE_CLIENT_SECRET", "KINGDEE_USERNAME", "KINGDEE_ACCOUNT_ID")
)
if created and not (config.get("KINGDEE_ACCESS_TOKEN") or config.get("KINGDEE_TOKEN_BODY_JSON") or has_official_token_credentials):
raise RuntimeError(f"Kingdee config template created at {Path(config_path or '.runtime/kingdee.env').expanduser()}; fill it and rerun")
if args.dry_run_auth:
token = get_kingdee_access_token(config, cache_path=token_cache_path_for_config(config_path))
obsidian_wiki/kingdee_api.py
+79 -56
View File
@@ -16,6 +16,7 @@ from obsidian_wiki.config import parse_config_file
from obsidian_wiki.file_assets import clean_config_value
DEFAULT_SALORDER_API_URL = "https://beijida.kdgalaxy.com/kapi/v2/h5a3/sm/sm_salorder/querySalOrder"
DEFAULT_TOKEN_API_URL = "https://beijida.kdgalaxy.com/kapi/oauth2/getToken"
DEFAULT_CONFIG_PATH = Path(".runtime") / "kingdee.env"
DEFAULT_TOKEN_CACHE_PATH = Path(".runtime") / "kingdee-token.json"
DEFAULT_PAGE_SIZE = 100
@@ -40,51 +41,53 @@ KINGDEE_SALORDER_API_URL={DEFAULT_SALORDER_API_URL}
KINGDEE_ACCESS_TOKEN=
# 变量：KINGDEE_TOKEN_URL
# 用途：获取 access token 的登录/token 接口地址。
# 用途：官方增强型 Token 接口 /kapi/oauth2/getToken。
# 是否必填：可选，默认已按当前项目域名填写；如果只填 KINGDEE_ACCESS_TOKEN，则不会调用它。
KINGDEE_TOKEN_URL={DEFAULT_TOKEN_API_URL}
# 下面 5 项对应官方 getToken 请求 Body 的固定身份字段。
# 程序会按官方文档自动生成 nonce 和 timestamp；不用手写当前时间。
# 变量：KINGDEE_CLIENT_ID
# 用途：官方 client_id，第三方应用系统编码，即 appId。
# 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。
KINGDEE_TOKEN_URL=
KINGDEE_CLIENT_ID=
# 推荐填法：用下面这些拆分变量让程序自动拼 token JSON body。
# 如果你的 token 接口字段名不是 app_key/app_secret/timestamp，就改对应 FIELD。
# 变量：KINGDEE_CLIENT_SECRET
# 用途：官方 client_secret，第三方应用 AccessToken 认证密钥，即 appSecret。
# 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。
KINGDEE_CLIENT_SECRET=
# 变量：KINGDEE_APP_KEY
# 用途：金蝶 OpenAPI 应用的 app key/client id。
# 是否必填：自动登录时通常必填；具体以金蝶 token 文档为准。
KINGDEE_APP_KEY=
# 变量：KINGDEE_USERNAME
# 用途：官方 username，第三方应用代理用户的用户名。
# 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。
KINGDEE_USERNAME=
# 变量：KINGDEE_APP_SECRET
# 用途：金蝶 OpenAPI 应用的 app secret/client secret。
# 是否必填：自动登录时通常必填；具体以金蝶 token 文档为准。
KINGDEE_APP_SECRET=
# 变量：KINGDEE_ACCOUNT_ID
# 用途：官方 accountId，数据中心 id。
# 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。
KINGDEE_ACCOUNT_ID=
# 变量：KINGDEE_TOKEN_APP_KEY_FIELD
# 用途：token JSON body 中 app key 对应的字段名。
# 是否必填：可选，默认 app_key；如果金蝶文档写 client_id，就改成 client_id。
KINGDEE_TOKEN_APP_KEY_FIELD=app_key
# 变量：KINGDEE_LANGUAGE
# 用途：官方 language，返回多语言文本字段时使用。
# 是否必填：可选；默认 zh_CN。
KINGDEE_LANGUAGE=zh_CN
# 变量：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
# 说明：nonce 和 timestamp 会按官方文档自动生成。
# nonce 每次请求随机生成；timestamp 默认使用零时区 ISO 格式，例如 2025-06-12T00:00:10.009Z。
# 变量：KINGDEE_TOKEN_BODY_EXTRA_JSON
# 用途：token JSON body 的额外固定字段，例如租户、账套等；会合并到自动拼出的 body。
# 用途：token JSON 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_TIMESTAMP_FORMAT
# 用途：timestamp 格式。可选值：iso_utc、datetime_utc。
# 是否必填：可选，默认 iso_utc；不确定时不要改。
KINGDEE_TOKEN_TIMESTAMP_FORMAT=iso_utc
# 高级填法：如果金蝶环境不是增强型 Token，或 body 结构特殊，可直接填写完整 JSON。
# 填了 KINGDEE_TOKEN_BODY_JSON 后，会覆盖上面的官方默认 body。
# 变量：KINGDEE_TOKEN_BODY_JSON
# 用途：调用登录/token 接口时发送的完整 JSON 请求体。
# 是否必填：可选；只有不用拆分变量时才填写，必须是一行合法 JSON。
@@ -101,10 +104,15 @@ KINGDEE_TOKEN_HEADERS_JSON={{}}
KINGDEE_TOKEN_FIELD=data.access_token
# 变量：KINGDEE_TOKEN_EXPIRES_IN_FIELD
# 用途：从 token 接口返回 JSON 中读取有效期秒数的字段路径。
# 是否必填：可选，默认从 data.expires_in 读取；读不到时按 3600 秒处理。
# 用途：从 token 接口返回 JSON 中读取有效期的字段路径。
# 是否必填：可选，默认从 data.expires_in 读取；官方单位是毫秒。
KINGDEE_TOKEN_EXPIRES_IN_FIELD=data.expires_in
# 变量：KINGDEE_TOKEN_EXPIRES_IN_UNIT
# 用途：KINGDEE_TOKEN_EXPIRES_IN_FIELD 读到的单位。
# 是否必填：可选，默认 milliseconds，符合官方文档。
KINGDEE_TOKEN_EXPIRES_IN_UNIT=milliseconds
# 变量：KINGDEE_REFRESH_TOKEN_FIELD
# 用途：从 token 接口返回 JSON 中读取 refresh token 的字段路径。
# 是否必填：可选；只有配置 refresh 流程时才有用。
@@ -151,6 +159,8 @@ def load_kingdee_config(config_path: str | Path | None = None) -> dict[str, str]
"KINGDEE_LOGIN_URL": "KINGDEE_TOKEN_URL",
"KINGDEE_LOGIN_BODY_JSON": "KINGDEE_TOKEN_BODY_JSON",
"KINGDEE_LOGIN_HEADERS_JSON": "KINGDEE_TOKEN_HEADERS_JSON",
"KINGDEE_APP_KEY": "KINGDEE_CLIENT_ID",
"KINGDEE_APP_SECRET": "KINGDEE_CLIENT_SECRET",
}
for key, value in os.environ.items():
if key.startswith("KINGDEE_") and value:
@@ -159,13 +169,14 @@ def load_kingdee_config(config_path: str | Path | None = None) -> dict[str, str]
if values.get(old) and not values.get(new):
values[new] = values[old]
values.setdefault("KINGDEE_SALORDER_API_URL", DEFAULT_SALORDER_API_URL)
values.setdefault("KINGDEE_TOKEN_URL", DEFAULT_TOKEN_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_LANGUAGE", "zh_CN")
values.setdefault("KINGDEE_TOKEN_TIMESTAMP_FORMAT", "iso_utc")
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_TOKEN_EXPIRES_IN_UNIT", "milliseconds")
values.setdefault("KINGDEE_REFRESH_TOKEN_FIELD", "data.refresh_token")
values.setdefault("KINGDEE_PAGE_SIZE", str(DEFAULT_PAGE_SIZE))
return values
@@ -223,7 +234,10 @@ def get_kingdee_access_token(
access_token = _extract_path(token_payload, config.get("KINGDEE_TOKEN_FIELD", "data.access_token"))
if not access_token:
raise RuntimeError("Kingdee token response did not contain an access token")
expires_in = _int_or_default(_extract_path(token_payload, config.get("KINGDEE_TOKEN_EXPIRES_IN_FIELD", "data.expires_in")), 3600)
expires_in = _expires_in_seconds(
_extract_path(token_payload, config.get("KINGDEE_TOKEN_EXPIRES_IN_FIELD", "data.expires_in")),
config.get("KINGDEE_TOKEN_EXPIRES_IN_UNIT", "milliseconds"),
)
refresh_token = _extract_path(token_payload, config.get("KINGDEE_REFRESH_TOKEN_FIELD", "data.refresh_token")) or refresh_token
cache = {
"access_token": str(access_token),
@@ -343,19 +357,16 @@ def _build_token_body_json(config: dict[str, str], now_epoch: float) -> str:
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"))
body: dict[str, Any] = {
"client_id": config.get("KINGDEE_CLIENT_ID", ""),
"client_secret": config.get("KINGDEE_CLIENT_SECRET", ""),
"username": config.get("KINGDEE_USERNAME", ""),
"accountId": config.get("KINGDEE_ACCOUNT_ID", ""),
"nonce": config.get("KINGDEE_NONCE") or uuid.uuid4().hex,
"timestamp": _format_timestamp(now_epoch, config.get("KINGDEE_TOKEN_TIMESTAMP_FORMAT", "iso_utc")),
}
if config.get("KINGDEE_LANGUAGE"):
body["language"] = config["KINGDEE_LANGUAGE"]
extra = json.loads(config.get("KINGDEE_TOKEN_BODY_EXTRA_JSON", "{}") or "{}")
if not isinstance(extra, dict):
@@ -379,21 +390,33 @@ def _template_values(now_epoch: float, extra: dict[str, str] | None = None) -> d
def _format_timestamp(now_epoch: float, timestamp_format: str) -> str | int:
normalized = (timestamp_format or "epoch_ms").strip().lower()
normalized = (timestamp_format or "iso_utc").strip().lower()
dt = datetime.fromtimestamp(now_epoch, tz=timezone.utc)
if normalized in {"iso_utc", "iso_utc_ms"}:
return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(dt.microsecond / 1000):03d}Z"
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 _expires_in_seconds(value: Any, unit: str) -> int:
normalized = (unit or "milliseconds").strip().lower()
if normalized in {"milliseconds", "millisecond", "ms"}:
return max(int(_int_or_default(value, 7_200_000) / 1000), 1)
if normalized in {"seconds", "second", "s"}:
return _int_or_default(value, 7200)
if normalized == "auto":
parsed = _int_or_default(value, 7_200_000)
return max(int(parsed / 1000), 1) if parsed > 86_400 else parsed
raise RuntimeError(f"unsupported KINGDEE_TOKEN_EXPIRES_IN_UNIT: {unit}")
def _render_json_template(template: str, values: dict[str, str]) -> str:
if not template:
return "{}"
tests/test_kingdee_api.py
+19 -32
View File
@@ -67,9 +67,10 @@ 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)
self.assertIn("变量：KINGDEE_CLIENT_ID", text)
self.assertIn("变量：KINGDEE_USERNAME", text)
self.assertIn("变量：KINGDEE_ACCOUNT_ID", text)
self.assertIn("nonce 和 timestamp 会按官方文档自动生成", text)
def test_default_config_path_accepts_none(self):
cwd = Path.cwd()
@@ -109,7 +110,7 @@ class KingdeeApiTests(unittest.TestCase):
{
"data": {
"access_token": "access-1",
"expires_in": 3600,
"expires_in": 3600000,
"refresh_token": "refresh-1",
}
}
@@ -129,44 +130,30 @@ 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):
def test_access_token_body_defaults_to_official_fields_with_nonce_and_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_CLIENT_ID": "thirdapp",
"KINGDEE_CLIENT_SECRET": "secret",
"KINGDEE_USERNAME": "zhangSan",
"KINGDEE_ACCOUNT_ID": "1355633519610561531",
"KINGDEE_LANGUAGE": "zh_CN",
"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)
token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
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_id"], "thirdapp")
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")
self.assertEqual(body["username"], "zhangSan")
self.assertEqual(body["accountId"], "1355633519610561531")
self.assertEqual(body["language"], "zh_CN")
self.assertRegex(body["nonce"], r"^[0-9a-f]{32}$")
self.assertEqual(body["timestamp"], "1970-01-01T00:16:40.000Z")
def test_expired_token_uses_refresh_endpoint(self):
cache_path = self.root / ".runtime" / "kingdee-token.json"
@@ -196,7 +183,7 @@ class KingdeeApiTests(unittest.TestCase):
{
"data": {
"access_token": "new-token",
"expires_in": 3600,
"expires_in": 3600000,
"refresh_token": "refresh-2",
}
}