feat(business): add Kingdee sales order API sync

Implement private Kingdee config generation, token login/refresh caching, paginated sales order fetching, official response-field normalization, and CLI filters for deterministic business fact sync. Update docs to keep credentials in .runtime and cover dry-run auth.

Validation: python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests; git diff --check.

.env.example

@@ -28,6 +28,14 @@ OBSIDIAN_SOURCE_CATALOG_DIR=_sources/catalog
 # SQLite source-file locator. Leave empty to use _meta/file-assets.sqlite.
 OBSIDIAN_FILE_ASSET_DB=
 
+# --- Business Facts / Kingdee Sales Orders (optional) ---
+#
+# Exact operational statistics use <vault>/_meta/business-facts.sqlite.
+# Run `obsidian-wiki business config kingdee` to create the local private
+# .runtime/kingdee.env file for real Kingdee credentials and token settings.
+# Do not put real app secrets or access tokens in this example file.
+OBSIDIAN_BUSINESS_FACTS_DB=
+
 # --- Source File MCP Server (optional) ---
 #
 # Run with: obsidian-wiki-mcp

README.md

@@ -322,10 +322,16 @@ Operational counts and amounts belong in a structured fact database, not in mark
 # Sync from an exported or API-fetched Kingdee JSON payload.
 obsidian-wiki business sync kingdee --input ./kingdee-orders.json
 
-# Or fetch from a simple JSON endpoint.
-KINGDEE_API_URL="https://example.internal/orders" \
-KINGDEE_API_TOKEN="..." \
-obsidian-wiki business sync kingdee --since 2026-07-01T00:00:00Z
+# Or create the private local config for the real Kingdee API/token flow.
+obsidian-wiki business config kingdee
+vim .runtime/kingdee.env
+obsidian-wiki business sync kingdee --dry-run-auth
+
+# Then fetch reviewed June orders from Kingdee.
+obsidian-wiki business sync kingdee \
+  --start-bizdate 2026-06-01 \
+  --end-bizdate 2026-06-30 \
+  --billstatus C
 
 # Ask deterministic statistics questions.
 obsidian-wiki business query "2026年6月接了多少单"
@@ -333,7 +339,7 @@ obsidian-wiki business query "2026-06 sales amount"
 obsidian-wiki business query "2026-06 top customer"
 ```
 
-V1 uses fixed templates for monthly order count, monthly sales amount, monthly top customers, and a customer trend query. The default order-count basis is distinct Kingdee sales order number by order date. Detailed order rows stay in SQLite; the wiki remains for business context, customer/project knowledge, and later summary pages when you choose to generate them.
+V1 uses fixed templates for monthly order count, monthly sales amount, monthly top customers, and a customer trend query. The default order-count basis is distinct Kingdee sales order number by order date. Detailed order rows stay in SQLite; the wiki remains for business context, customer/project knowledge, and later summary pages when you choose to generate them. Real Kingdee credentials live in `.runtime/kingdee.env`, and fetched tokens are cached in `.runtime/kingdee-token.json`; both files are local-only and ignored by git.
 
 ### Optional: Source File MCP Server
 

SETUP.md

@@ -127,8 +127,7 @@ Knowledge that's project-specific goes under `projects/<name>/`. Knowledge that'
 |---|---|---|
 | `OBSIDIAN_SOURCES_DIR` | Directories with docs to ingest (comma-separated) | *(empty — point agent at specific files)* |
 | `OBSIDIAN_BUSINESS_FACTS_DB` | Optional path for structured business facts such as Kingdee orders | `<vault>/_meta/business-facts.sqlite` |
-| `KINGDEE_API_URL` | JSON endpoint used by `obsidian-wiki business sync kingdee` when `--input` is omitted | *(unset)* |
-| `KINGDEE_API_TOKEN` | Bearer token for the Kingdee-compatible endpoint | *(unset)* |
+| `.runtime/kingdee.env` | Private local Kingdee sales-order API config created by `obsidian-wiki business config kingdee` | official sales-order URL, blank credentials |
 | `OBSIDIAN_CATEGORIES` | Wiki page categories | `concepts,entities,skills,references,synthesis,journal` |
 | `OBSIDIAN_MAX_PAGES_PER_INGEST` | Max pages updated per ingest | `15` |
 | `CLAUDE_HISTORY_PATH` | Where to find Claude data | *auto-discovers from `~/.claude`* |
@@ -161,11 +160,23 @@ Use the business fact database for exact operational questions, for example mont
 
 ```bash
 obsidian-wiki business sync kingdee --input ./kingdee-orders.json
+
+obsidian-wiki business config kingdee
+# Fill .runtime/kingdee.env, then test login/refresh without syncing orders:
+obsidian-wiki business sync kingdee --dry-run-auth
+
+obsidian-wiki business sync kingdee \
+  --start-bizdate 2026-06-01 \
+  --end-bizdate 2026-06-30 \
+  --billstatus C
+
 obsidian-wiki business query "2026年6月接了多少单"
 obsidian-wiki business query "2026-06 sales amount"
 ```
 
-Without `--input`, the Kingdee sync command reads `KINGDEE_API_URL` and optional `KINGDEE_API_TOKEN`, then stores the fetched JSON response in `<vault>/_meta/business-facts.sqlite`. The answer reports the statistic, counting basis, data source, and last successful sync time. If the latest successful sync is before today, the answer warns that newer data may be missing.
+Without `--input`, the Kingdee sync command creates/reads `.runtime/kingdee.env`, obtains or refreshes an access token, calls the official sales order query API, then stores the fetched JSON response in `<vault>/_meta/business-facts.sqlite`. The private config supports configurable login/token URL, request body, response field paths, refresh URL/body, and required `x-acgw-identity`; fetched tokens are cached in `.runtime/kingdee-token.json`. `.runtime/` is ignored by git.
+
+The answer reports the statistic, counting basis, data source, and last successful sync time. If the latest successful sync is before today, the answer warns that newer data may be missing.
 
 ## How It Works
 

docs/business-facts.md

@@ -17,13 +17,37 @@ The wiki remains the place for context: customer background, project history, sa
 
 ```bash
 obsidian-wiki business sync kingdee --input ./kingdee-orders.json
-obsidian-wiki business sync kingdee --since 2026-07-01T00:00:00Z
+
+obsidian-wiki business config kingdee
+# Fill .runtime/kingdee.env, then test the login/refresh flow:
+obsidian-wiki business sync kingdee --dry-run-auth
+
+obsidian-wiki business sync kingdee \
+  --start-bizdate 2026-06-01 \
+  --end-bizdate 2026-06-30 \
+  --billstatus C
+
+obsidian-wiki business sync kingdee --start-modifytime "2026-07-01 00:00:00"
 obsidian-wiki business query "2026年6月接了多少单"
 obsidian-wiki business query "2026-06 sales amount"
 obsidian-wiki business query "2026-06 top customer"
 ```
 
-When `--input` is omitted, sync reads `KINGDEE_API_URL` and optional `KINGDEE_API_TOKEN`. The API response must be JSON and can be either a list of order objects or an object containing `orders`, `data`, `records`, or `items`.
+When `--input` is omitted, sync reads `.runtime/kingdee.env`, gets or refreshes an access token, calls the official Kingdee sales order API, and stores the returned rows. If the file does not exist, `business sync kingdee` creates the template and asks you to fill it before retrying.
+
+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_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.
+- `KINGDEE_ACGW_IDENTITY`: value sent as the required `x-acgw-identity` sales-order API header.
+- `KINGDEE_PAGE_SIZE`: default sales-order page size.
+
+`KINGDEE_LOGIN_URL`, `KINGDEE_LOGIN_BODY_JSON`, and `KINGDEE_LOGIN_HEADERS_JSON` are accepted as aliases for the token settings. `KINGDEE_ACCESS_TOKEN` can also be set for one-off manual testing. Fetched tokens are cached in `.runtime/kingdee-token.json`; both local files are ignored by git and should not be copied into `.env.example`.
+
+The API response can still be imported from a JSON file for replay or testing. JSON may be a list of order objects or an object containing `orders`, `rows`, `data`, `records`, or `items`.
 
 ## Database
 

obsidian_wiki/business_facts.py

@@ -9,16 +9,14 @@ from __future__ import annotations
 
 import hashlib
 import json
-import os
 import re
 import sqlite3
-import urllib.parse
-import urllib.request
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
 from obsidian_wiki.file_assets import clean_config_value
+from obsidian_wiki.kingdee_api import fetch_kingdee_sales_orders, load_kingdee_config, token_cache_path_for_config
 
 BUSINESS_DB_RELATIVE_PATH = Path("_meta") / "business-facts.sqlite"
 
@@ -115,20 +113,33 @@ def load_kingdee_payload_from_json(path: str | Path) -> tuple[list[dict[str, Any
     return extract_orders_from_payload(payload), payload
 
 
-def fetch_kingdee_payload(*, api_url: str | None = None, token: str | None = None, since: str | None = None) -> tuple[list[dict[str, Any]], Any]:
-    url = api_url or os.environ.get("KINGDEE_API_URL", "")
-    if not url:
-        raise RuntimeError("KINGDEE_API_URL is required when --input is not provided")
-    separator = "&" if "?" in url else "?"
-    if since:
-        url = f"{url}{separator}updated_since={urllib.parse.quote(since)}"
-    request = urllib.request.Request(url)
-    auth_token = token or os.environ.get("KINGDEE_API_TOKEN", "")
-    if auth_token:
-        request.add_header("Authorization", f"Bearer {auth_token}")
-    with urllib.request.urlopen(request, timeout=60) as response:
-        payload = json.loads(response.read().decode("utf-8"))
-    return extract_orders_from_payload(payload), payload
+def fetch_kingdee_payload(
+    *,
+    api_url: str | None = None,
+    token: str | None = None,
+    since: str | None = None,
+    config_path: str | Path | None = None,
+    config_overrides: dict[str, str] | None = None,
+    filters: dict[str, Any] | None = None,
+    page_size: int | None = None,
+    max_pages: int = 100,
+) -> tuple[list[dict[str, Any]], Any]:
+    config = load_kingdee_config(config_path)
+    config.update(config_overrides or {})
+    if api_url:
+        config["KINGDEE_SALORDER_API_URL"] = api_url
+    if token:
+        config["KINGDEE_ACCESS_TOKEN"] = token
+    safe_filters = dict(filters or {})
+    if since and not safe_filters.get("start_modifytime"):
+        safe_filters["start_modifytime"] = since
+    return fetch_kingdee_sales_orders(
+        config,
+        filters=safe_filters,
+        token_cache_path=token_cache_path_for_config(config_path),
+        page_size=page_size,
+        max_pages=max_pages,
+    )
 
 
 def extract_orders_from_payload(payload: Any) -> list[dict[str, Any]]:
@@ -136,7 +147,7 @@ def extract_orders_from_payload(payload: Any) -> list[dict[str, Any]]:
         return [item for item in payload if isinstance(item, dict)]
     if not isinstance(payload, dict):
         raise RuntimeError("Kingdee payload must be a JSON object or array")
-    for key in ("orders", "data", "records", "items"):
+    for key in ("orders", "rows", "data", "records", "items"):
         value = payload.get(key)
         if isinstance(value, list):
             return [item for item in value if isinstance(item, dict)]
@@ -216,17 +227,17 @@ def sync_kingdee_orders(
 
 def normalize_order(record: dict[str, Any]) -> dict[str, Any]:
     order_id = _text(_first(record, "order_id", "id", "fid", "FID", "FId", "f_id"))
-    order_no = _text(_first(record, "order_no", "bill_no", "billNo", "FBillNo", "f_bill_no", "number"))
+    order_no = _text(_first(record, "order_no", "billno", "bill_no", "billNo", "FBillNo", "f_bill_no", "number"))
     if not order_id:
         order_id = order_no
     if not order_no:
         order_no = order_id
-    order_date = _date_text(_first(record, "order_date", "date", "FDate", "orderDate", "bill_date"))
+    order_date = _date_text(_first(record, "order_date", "bizdate", "date", "FDate", "orderDate", "bill_date"))
     if not order_no or not order_date:
         raise RuntimeError("Kingdee order record must include order_no/order_id and order_date")
 
     customer = record.get("customer") if isinstance(record.get("customer"), dict) else {}
-    customer_id = _text(_first(record, "customer_id", "customerId", "FCustomerId") or _first(customer, "id", "number"))
+    customer_id = _text(_first(record, "customer_number", "customer_id", "customerId", "FCustomerId") or _first(customer, "number", "id"))
     customer_name = _text(_first(record, "customer_name", "customerName", "FCustomerName") or _first(customer, "name"))
     items = _items(record)
     return {
@@ -235,10 +246,10 @@ def normalize_order(record: dict[str, Any]) -> dict[str, Any]:
         "order_date": order_date,
         "customer_id": customer_id,
         "customer_name": customer_name,
-        "total_amount": _float(_first(record, "total_amount", "amount", "amount_total", "FAllAmount", "FTotalAmount")),
-        "currency": _text(_first(record, "currency", "FCurrency")),
-        "status": _text(_first(record, "status", "document_status", "FDocumentStatus")),
-        "updated_at": _date_time_text(_first(record, "updated_at", "updatedAt", "FModifyDate", "modify_date")),
+        "total_amount": _float(_first(record, "total_amount", "totalallamount", "totalamount", "amount", "amount_total", "FAllAmount", "FTotalAmount")),
+        "currency": _text(_first(record, "settlecurrency_number", "currency_number", "currency", "FCurrency")),
+        "status": _status_text(record),
+        "updated_at": _date_time_text(_first(record, "updated_at", "modifytime", "updatedAt", "FModifyDate", "modify_date")),
         "items": items,
     }
 
@@ -515,7 +526,7 @@ def _upsert_order(conn: sqlite3.Connection, order: dict[str, Any], *, raw_hash:
 
 def _items(record: dict[str, Any]) -> list[dict[str, Any]]:
     raw_items = None
-    for key in ("items", "entries", "order_items", "FEntity"):
+    for key in ("items", "billentry", "entries", "order_items", "FEntity"):
         if isinstance(record.get(key), list):
             raw_items = record[key]
             break
@@ -526,15 +537,15 @@ def _items(record: dict[str, Any]) -> list[dict[str, Any]]:
         if not isinstance(raw, dict):
             continue
         product = raw.get("product") if isinstance(raw.get("product"), dict) else {}
-        product_id = _text(_first(raw, "product_id", "productId", "material_id", "FMaterialId") or _first(product, "id", "number"))
-        product_name = _text(_first(raw, "product_name", "productName", "material_name", "FMaterialName") or _first(product, "name"))
+        product_id = _text(_first(raw, "material_masterid_number", "product_id", "productId", "material_id", "FMaterialId") or _first(product, "number", "id"))
+        product_name = _text(_first(raw, "material_masterid_name", "product_name", "productName", "material_name", "FMaterialName") or _first(product, "name"))
         items.append(
             {
-                "line_id": _text(_first(raw, "line_id", "line_no", "entry_id", "FEntryID", "seq")) or str(index),
+                "line_id": _text(_first(raw, "line_id", "id", "line_no", "entry_id", "FEntryID", "seq")) or str(index),
                 "product_id": product_id,
                 "product_name": product_name,
                 "quantity": _float(_first(raw, "quantity", "qty", "FQty")),
-                "amount": _float(_first(raw, "amount", "line_amount", "FAmount")),
+                "amount": _float(_first(raw, "amountandtax", "amount", "line_amount", "FAmount")),
             }
         )
     return items
@@ -572,3 +583,12 @@ def _float(value: Any) -> float:
 def _extract_customer_name(question: str) -> str:
     match = re.search(r"customer\s+(.+?)\s+trend", question, re.IGNORECASE)
     return match.group(1).strip() if match else ""
+
+
+def _status_text(record: dict[str, Any]) -> str:
+    parts = []
+    for key in ("billstatus", "orderstatus", "closestatus", "status", "document_status", "FDocumentStatus"):
+        value = _text(record.get(key))
+        if value:
+            parts.append(f"{key}={value}")
+    return ";".join(parts)

obsidian_wiki/cli.py

@@ -34,6 +34,13 @@ from obsidian_wiki.file_assets import (
     search_assets,
     stage_files,
 )
+from obsidian_wiki.kingdee_api import (
+    build_sales_order_filters,
+    ensure_kingdee_config,
+    get_kingdee_access_token,
+    load_kingdee_config,
+    token_cache_path_for_config,
+)
 
 HOME = Path.home()
 GLOBAL_CONFIG_DIR = HOME / ".obsidian-wiki"
@@ -506,7 +513,27 @@ def cmd_business_sync_kingdee(args: argparse.Namespace) -> int:
     if args.input:
         records, payload = load_kingdee_payload_from_json(args.input)
     else:
-        records, payload = fetch_kingdee_payload(api_url=args.api_url, token=args.token, since=args.since)
+        config_path = getattr(args, "config", None)
+        created = ensure_kingdee_config(config_path)
+        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")):
+            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))
+            print(f"✅  Kingdee auth OK: token length {len(token)}")
+            return 0
+        records, payload = fetch_kingdee_payload(
+            api_url=args.api_url,
+            token=args.token or args.access_token,
+            since=args.since,
+            config_path=config_path,
+            config_overrides=overrides,
+            filters=build_sales_order_filters(args),
+            page_size=args.page_size,
+            max_pages=args.max_pages,
+        )
     result = sync_kingdee_orders(
         vault_path,
         records,
@@ -523,6 +550,29 @@ def cmd_business_sync_kingdee(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_business_config_kingdee(args: argparse.Namespace) -> int:
+    config_path = Path(args.config or ".runtime/kingdee.env").expanduser()
+    created = ensure_kingdee_config(config_path)
+    state = "created" if created else "exists"
+    print(f"✅  Kingdee config {state}: {config_path}")
+    return 0
+
+
+def _kingdee_cli_overrides(args: argparse.Namespace) -> dict[str, str]:
+    overrides: dict[str, str] = {}
+    if getattr(args, "api_url", None):
+        overrides["KINGDEE_SALORDER_API_URL"] = args.api_url
+    if getattr(args, "token", None):
+        overrides["KINGDEE_ACCESS_TOKEN"] = args.token
+    if getattr(args, "access_token", None):
+        overrides["KINGDEE_ACCESS_TOKEN"] = args.access_token
+    if getattr(args, "identity", None):
+        overrides["KINGDEE_ACGW_IDENTITY"] = args.identity
+    if getattr(args, "page_size", None):
+        overrides["KINGDEE_PAGE_SIZE"] = str(args.page_size)
+    return overrides
+
+
 def cmd_business_query(args: argparse.Namespace) -> int:
     vault_path = _require_vault(args)
     question = " ".join(args.question).strip()
@@ -663,6 +713,13 @@ def build_parser() -> argparse.ArgumentParser:
     bp = sub.add_parser("business", help="sync and query structured business facts")
     business_sub = bp.add_subparsers(dest="business_command")
 
+    bcp = business_sub.add_parser("config", help="create local business integration config files")
+    business_config_sub = bcp.add_subparsers(dest="business_config_target")
+
+    bckp = business_config_sub.add_parser("kingdee", help="create a local Kingdee API config template")
+    bckp.add_argument("--config", metavar="PATH", help="Kingdee config path (defaults to .runtime/kingdee.env)")
+    bckp.set_defaults(func=cmd_business_config_kingdee)
+
     bsp = business_sub.add_parser("sync", help="sync business facts from an external system")
     business_sync_sub = bsp.add_subparsers(dest="business_sync_source")
 
@@ -670,9 +727,29 @@ def build_parser() -> argparse.ArgumentParser:
     _add_vault_arg(bkp)
     bkp.add_argument("--db", metavar="PATH", help="SQLite DB path (defaults to <vault>/_meta/business-facts.sqlite)")
     bkp.add_argument("--input", metavar="PATH", help="JSON payload exported from or fetched from Kingdee")
-    bkp.add_argument("--api-url", help="Kingdee-compatible JSON API URL; defaults to KINGDEE_API_URL")
-    bkp.add_argument("--token", help="API bearer token; defaults to KINGDEE_API_TOKEN")
+    bkp.add_argument("--config", metavar="PATH", help="Kingdee config path (defaults to .runtime/kingdee.env)")
+    bkp.add_argument("--api-url", help="Kingdee sales order API URL; defaults to KINGDEE_SALORDER_API_URL")
+    bkp.add_argument("--token", help="direct access token override; compatibility alias for --access-token")
+    bkp.add_argument("--access-token", help="direct access token override")
+    bkp.add_argument("--identity", help="x-acgw-identity override")
     bkp.add_argument("--since", help="optional updated_since cursor passed to the API and sync log")
+    bkp.add_argument("--start-bizdate")
+    bkp.add_argument("--end-bizdate")
+    bkp.add_argument("--start-createtime")
+    bkp.add_argument("--end-createtime")
+    bkp.add_argument("--start-modifytime")
+    bkp.add_argument("--end-modifytime")
+    bkp.add_argument("--start-auditdate")
+    bkp.add_argument("--end-auditdate")
+    bkp.add_argument("--id", action="append", help="sales order ID; repeat or comma-separate")
+    bkp.add_argument("--billno", action="append", help="sales order bill number; repeat or comma-separate")
+    bkp.add_argument("--billstatus", action="append", help="A/B/C status; repeat or comma-separate")
+    bkp.add_argument("--org-number", action="append", help="sales org code; repeat or comma-separate")
+    bkp.add_argument("--customer-number", action="append", help="customer code; repeat or comma-separate")
+    bkp.add_argument("--customer-createorg-number", action="append", help="customer create org code; repeat or comma-separate")
+    bkp.add_argument("--page-size", type=int, help="Kingdee pageSize")
+    bkp.add_argument("--max-pages", type=int, default=100, help="maximum pages to fetch")
+    bkp.add_argument("--dry-run-auth", action="store_true", help="fetch/refresh access token and exit without syncing orders")
     bkp.add_argument("--json", action="store_true", help="print machine-readable sync result")
     bkp.set_defaults(func=cmd_business_sync_kingdee)
 

obsidian_wiki/kingdee_api.py

@@ -0,0 +1,310 @@
+"""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 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_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"""# Kingdee sales order API configuration.
+# This file is local-only; .runtime/ is ignored by git.
+KINGDEE_SALORDER_API_URL={DEFAULT_SALORDER_API_URL}
+
+# Token endpoint details from your Kingdee OpenAPI app.
+KINGDEE_TOKEN_URL=
+KINGDEE_TOKEN_BODY_JSON=
+KINGDEE_TOKEN_HEADERS_JSON={{}}
+KINGDEE_TOKEN_FIELD=data.access_token
+KINGDEE_TOKEN_EXPIRES_IN_FIELD=data.expires_in
+KINGDEE_REFRESH_TOKEN_FIELD=data.refresh_token
+
+# Optional refresh endpoint. Use {{refresh_token}} in the JSON body template.
+KINGDEE_REFRESH_URL=
+KINGDEE_REFRESH_BODY_JSON=
+
+# Required by the sales order API.
+KINGDEE_ACGW_IDENTITY=
+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",
+    }
+    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_HEADERS_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")
+    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", ""), {"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,
+            config.get("KINGDEE_TOKEN_BODY_JSON", ""),
+            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 = _int_or_default(_extract_path(token_payload, config.get("KINGDEE_TOKEN_EXPIRES_IN_FIELD", "data.expires_in")), 3600)
+    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 _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()]

tests/test_business_facts.py

@@ -118,6 +118,50 @@ class BusinessFactsTests(unittest.TestCase):
         self.assertTrue(result["is_stale"])
         self.assertIn("may not include data after 2026-07-03T09:00:00Z", result["answer"])
 
+    def test_official_kingdee_sales_order_shape_maps_into_business_facts(self):
+        payload = {
+            "data": {
+                "rows": [
+                    {
+                        "id": "3551625973803228160",
+                        "billno": "SO-REAL-001",
+                        "billstatus": "C",
+                        "orderstatus": "C",
+                        "closestatus": "A",
+                        "bizdate": "2026-06-08",
+                        "modifytime": "2026-06-09 10:30:00",
+                        "customer_number": "CUST-001",
+                        "customer_name": "真实客户",
+                        "totalallamount": "1234.56",
+                        "totalamount": "1000.00",
+                        "settlecurrency_number": "CNY",
+                        "billentry": [
+                            {
+                                "id": "4568544278879012864",
+                                "material_masterid_number": "MAT-001",
+                                "material_masterid_name": "车载连接器",
+                                "qty": "711.97",
+                                "amountandtax": "1234.56",
+                                "amount": "1000.00",
+                            }
+                        ],
+                    }
+                ]
+            }
+        }
+
+        sync_kingdee_orders(
+            self.vault,
+            payload["data"]["rows"],
+            source_payload=payload,
+            synced_at="2026-07-06T09:00:00Z",
+        )
+        count = query_business_question(self.vault, "2026-06 order count")
+        amount = query_business_question(self.vault, "2026-06 sales amount")
+
+        self.assertEqual(count["data"]["order_count"], 1)
+        self.assertEqual(amount["data"]["sales_amount"], 1234.56)
+
     def test_cli_sync_payload_can_be_loaded_from_json_file(self):
         payload_path = self.vault / "kingdee.json"
         payload_path.write_text(

tests/test_kingdee_api.py

@@ -0,0 +1,242 @@
+import json
+import os
+import stat
+import tempfile
+import unittest
+from contextlib import redirect_stdout
+from io import StringIO
+from pathlib import Path
+
+from obsidian_wiki.kingdee_api import (
+    DEFAULT_SALORDER_API_URL,
+    ensure_kingdee_config,
+    fetch_kingdee_sales_orders,
+    get_kingdee_access_token,
+)
+
+
+class FakeResponse:
+    def __init__(self, payload):
+        self.payload = payload
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def read(self):
+        return json.dumps(self.payload).encode("utf-8")
+
+
+class RecordingUrlopen:
+    def __init__(self, responses):
+        self.responses = list(responses)
+        self.calls = []
+
+    def __call__(self, request, timeout=60):
+        self.calls.append({"request": request, "timeout": timeout})
+        if not self.responses:
+            raise AssertionError("unexpected HTTP request")
+        return FakeResponse(self.responses.pop(0))
+
+
+class KingdeeApiTests(unittest.TestCase):
+    def setUp(self):
+        self.tempdir = tempfile.TemporaryDirectory()
+        self.root = Path(self.tempdir.name)
+
+    def tearDown(self):
+        self.tempdir.cleanup()
+
+    def test_config_command_creates_private_template_without_overwriting(self):
+        config_path = self.root / ".runtime" / "kingdee.env"
+
+        created = ensure_kingdee_config(config_path)
+        config_path.write_text(config_path.read_text(encoding="utf-8") + "\nKINGDEE_TOKEN_URL=https://token.example\n", encoding="utf-8")
+        created_again = ensure_kingdee_config(config_path)
+
+        self.assertTrue(created)
+        self.assertFalse(created_again)
+        self.assertEqual(stat.S_IMODE(config_path.stat().st_mode), 0o600)
+        text = config_path.read_text(encoding="utf-8")
+        self.assertIn(f"KINGDEE_SALORDER_API_URL={DEFAULT_SALORDER_API_URL}", text)
+        self.assertIn("KINGDEE_TOKEN_BODY_JSON=", text)
+        self.assertIn("KINGDEE_TOKEN_URL=https://token.example", text)
+
+    def test_default_config_path_accepts_none(self):
+        cwd = Path.cwd()
+        try:
+            os.chdir(self.root)
+            created = ensure_kingdee_config(None)
+        finally:
+            os.chdir(cwd)
+
+        self.assertTrue(created)
+        self.assertTrue((self.root / ".runtime" / "kingdee.env").is_file())
+
+    def test_cli_config_kingdee_writes_template(self):
+        from obsidian_wiki.cli import main
+
+        config_path = self.root / ".runtime" / "kingdee.env"
+        output = StringIO()
+        with redirect_stdout(output):
+            code = main(["business", "config", "kingdee", "--config", str(config_path)])
+
+        self.assertEqual(code, 0)
+        self.assertTrue(config_path.is_file())
+        self.assertIn("Kingdee config", output.getvalue())
+
+    def test_access_token_is_fetched_and_reused_from_private_cache(self):
+        cache_path = self.root / ".runtime" / "kingdee-token.json"
+        config = {
+            "KINGDEE_TOKEN_URL": "https://token.example/access",
+            "KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc","app_secret":"secret"}',
+            "KINGDEE_TOKEN_HEADERS_JSON": '{"X-App":"demo"}',
+            "KINGDEE_TOKEN_FIELD": "data.access_token",
+            "KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
+            "KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
+        }
+        http = RecordingUrlopen(
+            [
+                {
+                    "data": {
+                        "access_token": "access-1",
+                        "expires_in": 3600,
+                        "refresh_token": "refresh-1",
+                    }
+                }
+            ]
+        )
+
+        token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
+        cached = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1100)
+
+        self.assertEqual(token, "access-1")
+        self.assertEqual(cached, "access-1")
+        self.assertEqual(len(http.calls), 1)
+        request = http.calls[0]["request"]
+        self.assertEqual(request.get_method(), "POST")
+        self.assertEqual(request.full_url, "https://token.example/access")
+        self.assertEqual(json.loads(request.data.decode("utf-8"))["app_key"], "abc")
+        self.assertEqual(request.get_header("X-app"), "demo")
+        self.assertEqual(stat.S_IMODE(cache_path.stat().st_mode), 0o600)
+
+    def test_expired_token_uses_refresh_endpoint(self):
+        cache_path = self.root / ".runtime" / "kingdee-token.json"
+        cache_path.parent.mkdir(parents=True)
+        cache_path.write_text(
+            json.dumps(
+                {
+                    "access_token": "old-token",
+                    "refresh_token": "refresh-1",
+                    "expires_at": 999,
+                    "obtained_at": 100,
+                }
+            ),
+            encoding="utf-8",
+        )
+        config = {
+            "KINGDEE_TOKEN_URL": "https://token.example/access",
+            "KINGDEE_TOKEN_BODY_JSON": '{"app_key":"abc"}',
+            "KINGDEE_REFRESH_URL": "https://token.example/refresh",
+            "KINGDEE_REFRESH_BODY_JSON": '{"refresh_token":"{refresh_token}"}',
+            "KINGDEE_TOKEN_FIELD": "data.access_token",
+            "KINGDEE_TOKEN_EXPIRES_IN_FIELD": "data.expires_in",
+            "KINGDEE_REFRESH_TOKEN_FIELD": "data.refresh_token",
+        }
+        http = RecordingUrlopen(
+            [
+                {
+                    "data": {
+                        "access_token": "new-token",
+                        "expires_in": 3600,
+                        "refresh_token": "refresh-2",
+                    }
+                }
+            ]
+        )
+
+        token = get_kingdee_access_token(config, cache_path=cache_path, urlopen=http, now_epoch=1000)
+
+        self.assertEqual(token, "new-token")
+        request = http.calls[0]["request"]
+        self.assertEqual(request.full_url, "https://token.example/refresh")
+        self.assertEqual(json.loads(request.data.decode("utf-8"))["refresh_token"], "refresh-1")
+
+    def test_sales_order_fetch_posts_filters_headers_and_paginates(self):
+        cache_path = self.root / ".runtime" / "kingdee-token.json"
+        cache_path.parent.mkdir(parents=True)
+        cache_path.write_text(
+            json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
+            encoding="utf-8",
+        )
+        config = {
+            "KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
+            "KINGDEE_ACGW_IDENTITY": "identity-1",
+            "KINGDEE_PAGE_SIZE": "2",
+        }
+        page_1 = {
+            "status": True,
+            "errorCode": "",
+            "message": "",
+            "data": {
+                "pageNo": 1,
+                "pageSize": 2,
+                "rows": [{"id": "1", "billno": "SO-001"}, {"id": "2", "billno": "SO-002"}],
+                "totalCount": 3,
+            },
+        }
+        page_2 = {
+            "status": True,
+            "errorCode": "",
+            "message": "",
+            "data": {
+                "pageNo": 2,
+                "pageSize": 2,
+                "rows": [{"id": "3", "billno": "SO-003"}],
+                "totalCount": 3,
+            },
+        }
+        http = RecordingUrlopen([page_1, page_2])
+
+        rows, payload = fetch_kingdee_sales_orders(
+            config,
+            filters={"start_bizdate": "2026-06-01", "billstatus": ["C"]},
+            token_cache_path=cache_path,
+            urlopen=http,
+            now_epoch=1000,
+        )
+
+        self.assertEqual([row["billno"] for row in rows], ["SO-001", "SO-002", "SO-003"])
+        self.assertEqual(payload["pages_fetched"], 2)
+        self.assertEqual(len(http.calls), 2)
+        first = http.calls[0]["request"]
+        self.assertEqual(first.get_method(), "POST")
+        self.assertEqual(first.get_header("Accesstoken"), "access-1")
+        self.assertEqual(first.get_header("X-acgw-identity"), "identity-1")
+        body = json.loads(first.data.decode("utf-8"))
+        self.assertEqual(body["data"]["start_bizdate"], "2026-06-01")
+        self.assertEqual(body["data"]["billstatus"], ["C"])
+        self.assertEqual(body["pageNo"], 1)
+        self.assertEqual(body["pageSize"], 2)
+
+    def test_sales_order_api_failure_reports_code_and_message(self):
+        cache_path = self.root / ".runtime" / "kingdee-token.json"
+        cache_path.parent.mkdir(parents=True)
+        cache_path.write_text(
+            json.dumps({"access_token": "access-1", "expires_at": 9999, "obtained_at": 1000}),
+            encoding="utf-8",
+        )
+        config = {
+            "KINGDEE_SALORDER_API_URL": "https://kingdee.example/querySalOrder",
+            "KINGDEE_ACGW_IDENTITY": "identity-1",
+        }
+        http = RecordingUrlopen([{"status": False, "errorCode": "E001", "message": "bad filter"}])
+
+        with self.assertRaisesRegex(RuntimeError, "E001.*bad filter"):
+            fetch_kingdee_sales_orders(config, filters={}, token_cache_path=cache_path, urlopen=http, now_epoch=1000)
+
+
+if __name__ == "__main__":
+    unittest.main()