obsidian-wiki/obsidian_wiki/kingdee_api.py

"""Kingdee sales-order API adapter with local token/config management."""

from __future__ import annotations

import json
import os
import stat
import time
import uuid
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

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
DEFAULT_MAX_PAGES = 100
TOKEN_SKEW_SECONDS = 60


Urlopen = Callable[[urllib.request.Request], Any]


KINGDEE_CONFIG_TEMPLATE = f"""# 金蝶销售订单 API 本地配置。
# 这个文件只保存在本机；.runtime/ 已被 git 忽略，请不要提交密钥或 token。

# 变量：KINGDEE_SALORDER_API_URL
# 用途：查询销售订单 API 地址。
# 是否必填：可选，默认已是当前项目使用的官方 querySalOrder 接口。
KINGDEE_SALORDER_API_URL={DEFAULT_SALORDER_API_URL}

# 变量：KINGDEE_ACCESS_TOKEN
# 用途：直接指定 access token，绕过自动登录/刷新流程。
# 是否必填：可选，手动测试用；长期定时同步建议使用自动登录配置。
KINGDEE_ACCESS_TOKEN=

# 变量：KINGDEE_TOKEN_URL
# 用途：官方增强型 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_CLIENT_ID=

# 变量：KINGDEE_CLIENT_SECRET
# 用途：官方 client_secret，第三方应用 AccessToken 认证密钥，即 appSecret。
# 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。
KINGDEE_CLIENT_SECRET=

# 变量：KINGDEE_USERNAME
# 用途：官方 username，第三方应用代理用户的用户名。
# 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。
KINGDEE_USERNAME=

# 变量：KINGDEE_ACCOUNT_ID
# 用途：官方 accountId，数据中心 id。
# 是否必填：自动登录时必填；如果只填 KINGDEE_ACCESS_TOKEN，则不用填。
KINGDEE_ACCOUNT_ID=

# 变量：KINGDEE_LANGUAGE
# 用途：官方 language，返回多语言文本字段时使用。
# 是否必填：可选；默认 zh_CN。
KINGDEE_LANGUAGE=zh_CN

# 说明：nonce 和 timestamp 会按官方文档自动生成。
# nonce 每次请求随机生成；timestamp 默认使用零时区 ISO 格式，例如 2025-06-12T00:00:10.009Z。

# 变量：KINGDEE_TOKEN_BODY_EXTRA_JSON
# 用途：token JSON body 的额外固定字段；一般不需要。
# 是否必填：可选；没有额外字段时保持 {{}}。
KINGDEE_TOKEN_BODY_EXTRA_JSON={{}}

# 变量：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。
KINGDEE_TOKEN_BODY_JSON=

# 变量：KINGDEE_TOKEN_HEADERS_JSON
# 用途：调用登录/token 接口时额外发送的请求头。
# 是否必填：可选；没有额外 header 时保持 {{}}。
KINGDEE_TOKEN_HEADERS_JSON={{}}

# 变量：KINGDEE_TOKEN_FIELD
# 用途：从 token 接口返回 JSON 中读取 access token 的字段路径。
# 是否必填：可选，默认从 data.access_token 读取。
KINGDEE_TOKEN_FIELD=data.access_token

# 变量：KINGDEE_TOKEN_EXPIRES_IN_FIELD
# 用途：从 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 流程时才有用。
KINGDEE_REFRESH_TOKEN_FIELD=data.refresh_token

# 变量：KINGDEE_REFRESH_URL
# 用途：refresh token 接口地址。
# 是否必填：可选；不填时 token 过期会重新调用 KINGDEE_TOKEN_URL。
KINGDEE_REFRESH_URL=

# 变量：KINGDEE_REFRESH_BODY_JSON
# 用途：调用 refresh 接口时发送的 JSON 请求体；可使用 {{refresh_token}} 占位符。
# 是否必填：可选；填写 KINGDEE_REFRESH_URL 时通常需要填写。
KINGDEE_REFRESH_BODY_JSON=

# 变量：KINGDEE_ACGW_IDENTITY
# 用途：销售订单 API 要求的 x-acgw-identity 请求头。
# 是否必填：必填；不填无法调用销售订单查询 API。
KINGDEE_ACGW_IDENTITY=

# 变量：KINGDEE_PAGE_SIZE
# 用途：销售订单 API 分页大小，对应请求体 pageSize。
# 是否必填：可选，默认 100。
KINGDEE_PAGE_SIZE={DEFAULT_PAGE_SIZE}
"""


def ensure_kingdee_config(config_path: str | Path | None = None) -> bool:
    path = Path(config_path or DEFAULT_CONFIG_PATH).expanduser()
    if path.exists():
        return False
    _ensure_private_parent(path)
    path.write_text(KINGDEE_CONFIG_TEMPLATE, encoding="utf-8")
    path.chmod(0o600)
    return True


def load_kingdee_config(config_path: str | Path | None = None) -> dict[str, str]:
    path = Path(config_path or DEFAULT_CONFIG_PATH).expanduser()
    values = parse_config_file(path)
    aliases = {
        "KINGDEE_API_URL": "KINGDEE_SALORDER_API_URL",
        "KINGDEE_API_TOKEN": "KINGDEE_ACCESS_TOKEN",
        "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:
            values[key] = clean_config_value(value)
    for old, new in aliases.items():
        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_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


def token_cache_path_for_config(config_path: str | Path | None = None) -> Path:
    if config_path is None:
        return DEFAULT_TOKEN_CACHE_PATH
    return Path(config_path).expanduser().with_name("kingdee-token.json")


def get_kingdee_access_token(
    config: dict[str, str],
    *,
    cache_path: str | Path = DEFAULT_TOKEN_CACHE_PATH,
    urlopen: Callable[..., Any] = urllib.request.urlopen,
    now_epoch: float | None = None,
) -> str:
    now = time.time() if now_epoch is None else float(now_epoch)
    if config.get("KINGDEE_ACCESS_TOKEN"):
        return config["KINGDEE_ACCESS_TOKEN"]

    cache_path = Path(cache_path).expanduser()
    cached = _read_token_cache(cache_path)
    if cached.get("access_token") and float(cached.get("expires_at") or 0) > now + TOKEN_SKEW_SECONDS:
        return str(cached["access_token"])

    token_payload: dict[str, Any] | None = None
    refresh_token = str(cached.get("refresh_token") or "")
    if refresh_token and config.get("KINGDEE_REFRESH_URL"):
        try:
            token_payload = _request_token(
                config["KINGDEE_REFRESH_URL"],
                _render_json_template(
                    config.get("KINGDEE_REFRESH_BODY_JSON", ""),
                    _template_values(now, {"refresh_token": refresh_token}),
                ),
                config.get("KINGDEE_TOKEN_HEADERS_JSON", "{}"),
                urlopen=urlopen,
            )
        except Exception:
            token_payload = None

    if token_payload is None:
        token_url = config.get("KINGDEE_TOKEN_URL", "")
        if not token_url:
            raise RuntimeError("KINGDEE_TOKEN_URL is required when no cached KINGDEE_ACCESS_TOKEN is available")
        token_payload = _request_token(
            token_url,
            _build_token_body_json(config, now),
            config.get("KINGDEE_TOKEN_HEADERS_JSON", "{}"),
            urlopen=urlopen,
        )

    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 = _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),
        "refresh_token": str(refresh_token or ""),
        "expires_at": now + max(expires_in - TOKEN_SKEW_SECONDS, 1),
        "obtained_at": now,
    }
    _write_private_json(cache_path, cache)
    return str(access_token)


def fetch_kingdee_sales_orders(
    config: dict[str, str],
    *,
    filters: dict[str, Any] | None = None,
    token_cache_path: str | Path = DEFAULT_TOKEN_CACHE_PATH,
    page_size: int | None = None,
    max_pages: int = DEFAULT_MAX_PAGES,
    urlopen: Callable[..., Any] = urllib.request.urlopen,
    now_epoch: float | None = None,
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    access_token = get_kingdee_access_token(config, cache_path=token_cache_path, urlopen=urlopen, now_epoch=now_epoch)
    identity = config.get("KINGDEE_ACGW_IDENTITY", "")
    if not identity:
        raise RuntimeError("KINGDEE_ACGW_IDENTITY is required for Kingdee sales order API")
    api_url = config.get("KINGDEE_SALORDER_API_URL") or DEFAULT_SALORDER_API_URL
    size = page_size or _int_or_default(config.get("KINGDEE_PAGE_SIZE"), DEFAULT_PAGE_SIZE)
    rows: list[dict[str, Any]] = []
    pages: list[dict[str, Any]] = []
    total_count: int | None = None
    safe_filters = filters or {}

    for page_no in range(1, max_pages + 1):
        body = {"data": safe_filters, "pageNo": page_no, "pageSize": size}
        request = urllib.request.Request(
            api_url,
            data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
            headers={
                "Content-Type": "application/json",
                "accesstoken": access_token,
                "x-acgw-identity": identity,
                "Idempotency-Key": str(uuid.uuid4()),
            },
            method="POST",
        )
        with urlopen(request, timeout=60) as response:
            payload = json.loads(response.read().decode("utf-8"))
        if payload.get("status") is False:
            raise RuntimeError(
                f"Kingdee sales order API failed: {payload.get('errorCode', '')} {payload.get('message', '')}".strip()
            )
        page_data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
        page_rows = page_data.get("rows") if isinstance(page_data.get("rows"), list) else []
        pages.append(payload)
        rows.extend([row for row in page_rows if isinstance(row, dict)])
        total = page_data.get("totalCount")
        if isinstance(total, int) and total >= 0:
            total_count = total
        if not page_rows or len(page_rows) < size:
            break
        if total_count is not None and len(rows) >= total_count:
            break
    return rows, {"source": "kingdee-sales-order-api", "pages_fetched": len(pages), "rows": rows, "pages": pages}


def build_sales_order_filters(args: Any) -> dict[str, Any]:
    fields = {
        "start_bizdate": getattr(args, "start_bizdate", None),
        "end_bizdate": getattr(args, "end_bizdate", None),
        "start_createtime": getattr(args, "start_createtime", None),
        "end_createtime": getattr(args, "end_createtime", None),
        "start_modifytime": getattr(args, "start_modifytime", None) or getattr(args, "since", None),
        "end_modifytime": getattr(args, "end_modifytime", None),
        "start_auditdate": getattr(args, "start_auditdate", None),
        "end_auditdate": getattr(args, "end_auditdate", None),
    }
    filters = {key: value for key, value in fields.items() if value}
    array_fields = {
        "id": getattr(args, "id", None),
        "billno": getattr(args, "billno", None),
        "billstatus": getattr(args, "billstatus", None),
        "org_number": getattr(args, "org_number", None),
        "customer_number": getattr(args, "customer_number", None),
        "customer_createorg_number": getattr(args, "customer_createorg_number", None),
    }
    for key, value in array_fields.items():
        values = _split_values(value)
        if values:
            filters[key] = values
    return filters


def _request_token(
    url: str,
    body_json: str,
    headers_json: str,
    *,
    urlopen: Callable[..., Any],
) -> dict[str, Any]:
    body = json.loads(body_json or "{}")
    headers = {"Content-Type": "application/json", **json.loads(headers_json or "{}")}
    request = urllib.request.Request(
        url,
        data=json.dumps(body, ensure_ascii=False).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    with urlopen(request, timeout=60) as response:
        payload = json.loads(response.read().decode("utf-8"))
    if isinstance(payload, dict) and payload.get("status") is False:
        raise RuntimeError(f"Kingdee token API failed: {payload.get('errorCode', '')} {payload.get('message', '')}".strip())
    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] = {
        "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):
        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 "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 == "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 "{}"
    rendered = template
    for key, value in values.items():
        rendered = rendered.replace("{" + key + "}", value)
    return rendered


def _extract_path(payload: Any, path: str | None) -> Any:
    if not path:
        return None
    current = payload
    for part in path.split("."):
        if not isinstance(current, dict):
            return None
        current = current.get(part)
    return current


def _read_token_cache(path: Path) -> dict[str, Any]:
    if not path.is_file():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}
    return data if isinstance(data, dict) else {}


def _write_private_json(path: Path, data: dict[str, Any]) -> None:
    _ensure_private_parent(path)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    path.chmod(0o600)


def _ensure_private_parent(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        path.parent.chmod(0o700)
    except OSError:
        pass


def _int_or_default(value: Any, default: int) -> int:
    try:
        parsed = int(str(value))
    except (TypeError, ValueError):
        return default
    return parsed if parsed > 0 else default


def _split_values(value: Any) -> list[str]:
    if not value:
        return []
    if isinstance(value, (list, tuple)):
        raw = []
        for item in value:
            raw.extend(str(item).split(","))
    else:
        raw = str(value).split(",")
    return [item.strip() for item in raw if item.strip()]