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

feat(business): preserve full Kingdee order structure

Browse Source 
Store expanded Kingdee sales order header, line, and delivery-plan fields in the business facts database while preserving raw JSON at each level.

Add migration-safe columns, a delivery-plan table, coverage for the official response shape, and database documentation.

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:31:39 +00:00
parent 72c666f165
commit dae51ab570
3 changed files with 377 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/business-facts.md
+6 -3
View File
@@ -62,12 +62,15 @@ Tables:
- `sync_runs`: one row per Kingdee sync attempt, including status, cursor, record counts, and error text.
- `raw_snapshots`: content-hashed raw JSON payloads for audit and replay.
- `orders`: one normalized row per sales order.
- `order_items`: normalized detail lines for product/quantity/line amount analysis.
- `orders`: one normalized row per sales order. It keeps the query-critical fields (`order_id`, `order_no`, `order_date`, customer, status, organization, department, operator, currency, total amount, net amount, tax amount, address, comment) plus `raw_json` for the full original order header.
- `order_items`: normalized material detail lines. It keeps material number/name/model, quantity, unit, price, tax-inclusive price, net amount, tax-inclusive amount, tax amount, delivery date, project, warehouse, and `raw_json` for the full original line.
- `order_deliveries`: normalized delivery-plan rows from `billentry[].orderdeliverentry[]`. It keeps planned demand date, planned delivery date, planned quantity, planned base quantity, unit, receive address/place, remark, transport lead time, and `raw_json` for the full original delivery-plan row.
- `customers`: customer lookup data observed from orders.
- `products`: product lookup data observed from order lines.
Order upsert key: `order_id`, falling back to `order_no` when no explicit ID is present. Re-syncing the same order updates the existing row and replaces that order's detail lines.
Order upsert key: `order_id`, falling back to `order_no` when no explicit ID is present. Re-syncing the same order updates the existing row and replaces that order's detail lines and delivery-plan rows.
`total_amount` is the statistic field used by monthly sales queries. For Kingdee sales orders it prefers `totalallamount` (价税合计) and falls back to other total fields only when the tax-inclusive total is absent. `total_net_amount`, `total_tax_amount`, and `total_all_amount` are stored separately so later reports can choose a different accounting basis without re-fetching historical payloads.
## Counting Basis
obsidian_wiki/business_facts.py
+287 -35
View File
@@ -20,6 +20,77 @@ from obsidian_wiki.kingdee_api import fetch_kingdee_sales_orders, load_kingdee_c
BUSINESS_DB_RELATIVE_PATH = Path("_meta") / "business-facts.sqlite"
ORDER_COLUMNS = (
"order_id",
"order_no",
"order_date",
"customer_id",
"customer_name",
"total_amount",
"currency",
"status",
"updated_at",
"billstatus",
"orderstatus",
"closestatus",
"org_id",
"org_number",
"org_name",
"dept_id",
"dept_number",
"dept_name",
"operator_id",
"operator_number",
"operator_name",
"total_net_amount",
"total_tax_amount",
"total_all_amount",
"receive_address",
"comment",
"raw_json",
)
ORDER_ITEM_COLUMNS = (
"order_id",
"line_id",
"product_id",
"product_name",
"quantity",
"amount",
"seq",
"material_model",
"unit_number",
"unit_name",
"price",
"price_and_tax",
"amount_and_tax",
"tax_amount",
"delivery_date",
"project_number",
"project_name",
"warehouse_number",
"warehouse_name",
"raw_json",
)
ORDER_DELIVERY_COLUMNS = (
"order_id",
"line_id",
"delivery_id",
"plan_date",
"plan_delivery_date",
"plan_qty",
"plan_base_qty",
"plan_unit_number",
"plan_unit_name",
"receive_address",
"receive_place_number",
"receive_place_name",
"remark",
"transport_lead_time",
"raw_json",
)
def utc_now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
@@ -85,6 +156,24 @@ def init_db(db_path: Path) -> None:
currency TEXT,
status TEXT,
updated_at TEXT,
billstatus TEXT,
orderstatus TEXT,
closestatus TEXT,
org_id TEXT,
org_number TEXT,
org_name TEXT,
dept_id TEXT,
dept_number TEXT,
dept_name TEXT,
operator_id TEXT,
operator_number TEXT,
operator_name TEXT,
total_net_amount REAL NOT NULL DEFAULT 0,
total_tax_amount REAL NOT NULL DEFAULT 0,
total_all_amount REAL NOT NULL DEFAULT 0,
receive_address TEXT,
comment TEXT,
raw_json TEXT,
raw_hash TEXT,
created_at TEXT NOT NULL,
synced_at TEXT NOT NULL
@@ -97,15 +186,101 @@ def init_db(db_path: Path) -> None:
product_name TEXT,
quantity REAL,
amount REAL,
seq INTEGER,
material_model TEXT,
unit_number TEXT,
unit_name TEXT,
price REAL,
price_and_tax REAL,
amount_and_tax REAL,
tax_amount REAL,
delivery_date TEXT,
project_number TEXT,
project_name TEXT,
warehouse_number TEXT,
warehouse_name TEXT,
raw_json TEXT,
PRIMARY KEY (order_id, line_id),
FOREIGN KEY (order_id) REFERENCES orders(order_id)
);
CREATE TABLE IF NOT EXISTS order_deliveries (
order_id TEXT NOT NULL,
line_id TEXT NOT NULL,
delivery_id TEXT NOT NULL,
plan_date TEXT,
plan_delivery_date TEXT,
plan_qty REAL,
plan_base_qty REAL,
plan_unit_number TEXT,
plan_unit_name TEXT,
receive_address TEXT,
receive_place_number TEXT,
receive_place_name TEXT,
remark TEXT,
transport_lead_time INTEGER,
raw_json TEXT,
PRIMARY KEY (order_id, line_id, delivery_id),
FOREIGN KEY (order_id, line_id) REFERENCES order_items(order_id, line_id)
);
CREATE INDEX IF NOT EXISTS idx_orders_order_date ON orders(order_date);
CREATE INDEX IF NOT EXISTS idx_orders_customer_name ON orders(customer_name);
CREATE INDEX IF NOT EXISTS idx_order_items_product_id ON order_items(product_id);
CREATE INDEX IF NOT EXISTS idx_order_deliveries_plan_date ON order_deliveries(plan_date);
"""
)
_ensure_columns(
conn,
"orders",
{
"billstatus": "TEXT",
"orderstatus": "TEXT",
"closestatus": "TEXT",
"org_id": "TEXT",
"org_number": "TEXT",
"org_name": "TEXT",
"dept_id": "TEXT",
"dept_number": "TEXT",
"dept_name": "TEXT",
"operator_id": "TEXT",
"operator_number": "TEXT",
"operator_name": "TEXT",
"total_net_amount": "REAL NOT NULL DEFAULT 0",
"total_tax_amount": "REAL NOT NULL DEFAULT 0",
"total_all_amount": "REAL NOT NULL DEFAULT 0",
"receive_address": "TEXT",
"comment": "TEXT",
"raw_json": "TEXT",
},
)
_ensure_columns(
conn,
"order_items",
{
"seq": "INTEGER",
"material_model": "TEXT",
"unit_number": "TEXT",
"unit_name": "TEXT",
"price": "REAL",
"price_and_tax": "REAL",
"amount_and_tax": "REAL",
"tax_amount": "REAL",
"delivery_date": "TEXT",
"project_number": "TEXT",
"project_name": "TEXT",
"warehouse_number": "TEXT",
"warehouse_name": "TEXT",
"raw_json": "TEXT",
},
)
def _ensure_columns(conn: sqlite3.Connection, table: str, columns: dict[str, str]) -> None:
existing = {row["name"] for row in conn.execute(f"PRAGMA table_info({table})").fetchall()}
for name, column_type in columns.items():
if name not in existing:
conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {column_type}")
def load_kingdee_payload_from_json(path: str | Path) -> tuple[list[dict[str, Any]], Any]:
@@ -240,16 +415,37 @@ def normalize_order(record: dict[str, Any]) -> dict[str, Any]:
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)
total_amount_value = _first(record, "total_amount", "totalallamount", "totalamount", "amount", "amount_total", "FAllAmount", "FTotalAmount")
total_all_amount = _float(total_amount_value)
explicit_total_all_amount = _first(record, "totalallamount", "total_all_amount")
return {
"order_id": order_id,
"order_no": order_no,
"order_date": order_date,
"customer_id": customer_id,
"customer_name": customer_name,
"total_amount": _float(_first(record, "total_amount", "totalallamount", "totalamount", "amount", "amount_total", "FAllAmount", "FTotalAmount")),
"total_amount": total_all_amount,
"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")),
"billstatus": _text(record.get("billstatus")),
"orderstatus": _text(record.get("orderstatus")),
"closestatus": _text(record.get("closestatus")),
"org_id": _text(record.get("org_id")),
"org_number": _text(record.get("org_number")),
"org_name": _text(record.get("org_name")),
"dept_id": _text(record.get("dept_id")),
"dept_number": _text(record.get("dept_number")),
"dept_name": _text(record.get("dept_name")),
"operator_id": _text(record.get("operator_id")),
"operator_number": _text(_first(record, "operator_operatornumber", "operator_number")),
"operator_name": _text(_first(record, "operator_operatorname", "operator_name")),
"total_net_amount": _float(_first(record, "totalamount", "total_net_amount")),
"total_tax_amount": _float(_first(record, "totaltaxamount", "total_tax_amount")),
"total_all_amount": _float(explicit_total_all_amount) if explicit_total_all_amount is not None else total_all_amount,
"receive_address": _text(_first(record, "receiveaddress", "receive_address")),
"comment": _text(record.get("comment")),
"raw_json": _json_dumps(record),
"items": items,
}
@@ -468,41 +664,26 @@ def _upsert_order(conn: sqlite3.Connection, order: dict[str, Any], *, raw_hash:
""",
(customer_id, order["customer_name"], product_seen_at),
)
order_columns_sql = ", ".join(ORDER_COLUMNS)
order_placeholders = ", ".join("?" for _ in (*ORDER_COLUMNS, "raw_hash", "created_at", "synced_at"))
order_updates = ",\n ".join(f"{column}=excluded.{column}" for column in ORDER_COLUMNS if column != "order_id")
conn.execute(
"""
INSERT INTO orders (
order_id, order_no, order_date, customer_id, customer_name,
total_amount, currency, status, updated_at, raw_hash, created_at, synced_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
f"""
INSERT INTO orders ({order_columns_sql}, raw_hash, created_at, synced_at)
VALUES ({order_placeholders})
ON CONFLICT(order_id) DO UPDATE SET
order_no=excluded.order_no,
order_date=excluded.order_date,
customer_id=excluded.customer_id,
customer_name=excluded.customer_name,
total_amount=excluded.total_amount,
currency=excluded.currency,
status=excluded.status,
updated_at=excluded.updated_at,
{order_updates},
raw_hash=excluded.raw_hash,
synced_at=excluded.synced_at
""",
(
order["order_id"],
order["order_no"],
order["order_date"],
order["customer_id"],
order["customer_name"],
order["total_amount"],
order["currency"],
order["status"],
order["updated_at"],
raw_hash,
synced_at,
synced_at,
),
tuple(order[column] for column in ORDER_COLUMNS) + (raw_hash, synced_at, synced_at),
)
conn.execute("DELETE FROM order_deliveries WHERE order_id=?", (order["order_id"],))
conn.execute("DELETE FROM order_items WHERE order_id=?", (order["order_id"],))
item_columns_sql = ", ".join(ORDER_ITEM_COLUMNS)
item_placeholders = ", ".join("?" for _ in ORDER_ITEM_COLUMNS)
delivery_columns_sql = ", ".join(ORDER_DELIVERY_COLUMNS)
delivery_placeholders = ", ".join("?" for _ in ORDER_DELIVERY_COLUMNS)
for item in order["items"]:
if item["product_id"]:
conn.execute(
@@ -515,13 +696,23 @@ def _upsert_order(conn: sqlite3.Connection, order: dict[str, Any], *, raw_hash:
""",
(item["product_id"], item["product_name"], product_seen_at),
)
item_row = {"order_id": order["order_id"], **item}
conn.execute(
"""
INSERT INTO order_items (order_id, line_id, product_id, product_name, quantity, amount)
VALUES (?, ?, ?, ?, ?, ?)
f"""
INSERT INTO order_items ({item_columns_sql})
VALUES ({item_placeholders})
""",
(order["order_id"], item["line_id"], item["product_id"], item["product_name"], item["quantity"], item["amount"]),
tuple(item_row[column] for column in ORDER_ITEM_COLUMNS),
)
for delivery in item["deliveries"]:
delivery_row = {"order_id": order["order_id"], "line_id": item["line_id"], **delivery}
conn.execute(
f"""
INSERT INTO order_deliveries ({delivery_columns_sql})
VALUES ({delivery_placeholders})
""",
tuple(delivery_row[column] for column in ORDER_DELIVERY_COLUMNS),
)
def _items(record: dict[str, Any]) -> list[dict[str, Any]]:
@@ -539,18 +730,66 @@ def _items(record: dict[str, Any]) -> list[dict[str, Any]]:
product = raw.get("product") if isinstance(raw.get("product"), dict) else {}
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"))
line_id = _text(_first(raw, "line_id", "id", "line_no", "entry_id", "FEntryID", "seq")) or str(index)
items.append(
{
"line_id": _text(_first(raw, "line_id", "id", "line_no", "entry_id", "FEntryID", "seq")) or str(index),
"line_id": line_id,
"product_id": product_id,
"product_name": product_name,
"quantity": _float(_first(raw, "quantity", "qty", "FQty")),
"amount": _float(_first(raw, "amountandtax", "amount", "line_amount", "FAmount")),
"amount": _float(_first(raw, "amount", "line_amount", "FAmount", "amountandtax")),
"seq": _int(_first(raw, "seq", "line_no")),
"material_model": _text(_first(raw, "material_masterid_modelnum", "material_model", "model")),
"unit_number": _text(_first(raw, "unit_number", "unitNumber", "FUnitNumber")),
"unit_name": _text(_first(raw, "unit_name", "unitName", "FUnitName")),
"price": _float(_first(raw, "price", "FPrice")),
"price_and_tax": _float(_first(raw, "priceandtax", "price_and_tax")),
"amount_and_tax": _float(_first(raw, "amountandtax", "amount_and_tax")),
"tax_amount": _float(_first(raw, "taxamount", "tax_amount")),
"delivery_date": _date_text(_first(raw, "deliverydate", "delivery_date")),
"project_number": _text(_first(raw, "project_number", "projectNumber")),
"project_name": _text(_first(raw, "project_name", "projectName")),
"warehouse_number": _text(_first(raw, "warehouse_number", "warehouseNumber")),
"warehouse_name": _text(_first(raw, "warehouse_name", "warehouseName")),
"raw_json": _json_dumps(raw),
"deliveries": _deliveries(raw),
}
)
return items
def _deliveries(item: dict[str, Any]) -> list[dict[str, Any]]:
raw_deliveries = None
for key in ("orderdeliverentry", "deliveries", "delivery_entries", "plan_entries"):
if isinstance(item.get(key), list):
raw_deliveries = item[key]
break
if raw_deliveries is None:
raw_deliveries = []
deliveries = []
for index, raw in enumerate(raw_deliveries, start=1):
if not isinstance(raw, dict):
continue
deliveries.append(
{
"delivery_id": _text(_first(raw, "id", "delivery_id", "entry_id")) or str(index),
"plan_date": _date_text(_first(raw, "d_plandate", "plan_date")),
"plan_delivery_date": _date_text(_first(raw, "d_plandeliverydate", "plan_delivery_date")),
"plan_qty": _float(_first(raw, "d_planqty", "plan_qty")),
"plan_base_qty": _float(_first(raw, "d_planbaseqty", "plan_base_qty")),
"plan_unit_number": _text(_first(raw, "d_planunit_number", "plan_unit_number")),
"plan_unit_name": _text(_first(raw, "d_planunit_name", "plan_unit_name")),
"receive_address": _text(_first(raw, "d_receiveaddress", "receive_address")),
"receive_place_number": _text(_first(raw, "d_receiveaddressf7_number", "receive_place_number")),
"receive_place_name": _text(_first(raw, "d_receiveaddressf7_name", "receive_place_name")),
"remark": _text(_first(raw, "d_remark", "remark")),
"transport_lead_time": _int(_first(raw, "d_transportleadtime", "transport_lead_time")),
"raw_json": _json_dumps(raw),
}
)
return deliveries
def _first(record: dict[str, Any], *keys: str) -> Any:
for key in keys:
if key in record and record[key] not in (None, ""):
@@ -580,6 +819,19 @@ def _float(value: Any) -> float:
return 0.0
def _int(value: Any) -> int | None:
if value in (None, ""):
return None
try:
return int(value)
except (TypeError, ValueError):
return None
def _json_dumps(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)
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 ""
tests/test_business_facts.py
+84
View File
@@ -1,4 +1,5 @@
import json
import sqlite3
import tempfile
import unittest
from contextlib import redirect_stdout
@@ -134,15 +135,48 @@ class BusinessFactsTests(unittest.TestCase):
"customer_name": "真实客户",
"totalallamount": "1234.56",
"totalamount": "1000.00",
"totaltaxamount": "234.56",
"org_number": "ORG-001",
"org_name": "销售组织",
"dept_number": "DEPT-001",
"dept_name": "销售部",
"operator_operatornumber": "OP-001",
"operator_operatorname": "张三",
"settlecurrency_number": "CNY",
"billentry": [
{
"id": "4568544278879012864",
"seq": 1,
"material_masterid_number": "MAT-001",
"material_masterid_name": "车载连接器",
"material_masterid_modelnum": "MODEL-A",
"qty": "711.97",
"priceandtax": "1.73",
"amountandtax": "1234.56",
"amount": "1000.00",
"unit_number": "PCS",
"unit_name": "个",
"deliverydate": "2026-06-20",
"project_number": "PRJ-001",
"project_name": "重点项目",
"warehouse_number": "WH-001",
"warehouse_name": "成品仓",
"orderdeliverentry": [
{
"id": "7690495588165568512",
"d_plandate": "2026-06-18",
"d_plandeliverydate": "2026-06-16",
"d_planqty": "500.00",
"d_planbaseqty": "500.00",
"d_planunit_number": "PCS",
"d_planunit_name": "个",
"d_receiveaddress": "测试地址",
"d_receiveaddressf7_number": "ADDR-001",
"d_receiveaddressf7_name": "测试收货点",
"d_remark": "首批",
"d_transportleadtime": 2,
}
],
}
],
}
@@ -161,6 +195,56 @@ class BusinessFactsTests(unittest.TestCase):
self.assertEqual(count["data"]["order_count"], 1)
self.assertEqual(amount["data"]["sales_amount"], 1234.56)
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
order = conn.execute(
"""
SELECT billstatus, orderstatus, closestatus, org_number, org_name,
dept_number, dept_name, operator_number, operator_name,
total_net_amount, total_tax_amount, total_all_amount, raw_json
FROM orders
WHERE order_id=?
""",
("3551625973803228160",),
).fetchone()
item = conn.execute(
"""
SELECT seq, product_id, product_name, material_model, unit_number, unit_name,
price_and_tax, amount, amount_and_tax, delivery_date,
project_number, project_name, warehouse_number, warehouse_name, raw_json
FROM order_items
WHERE order_id=? AND line_id=?
""",
("3551625973803228160", "4568544278879012864"),
).fetchone()
delivery = conn.execute(
"""
SELECT delivery_id, plan_date, plan_delivery_date, plan_qty, plan_base_qty,
plan_unit_number, plan_unit_name, receive_address,
receive_place_number, receive_place_name, remark, transport_lead_time, raw_json
FROM order_deliveries
WHERE order_id=? AND line_id=?
""",
("3551625973803228160", "4568544278879012864"),
).fetchone()
self.assertEqual(order["billstatus"], "C")
self.assertEqual(order["org_number"], "ORG-001")
self.assertEqual(order["operator_name"], "张三")
self.assertEqual(order["total_net_amount"], 1000.0)
self.assertEqual(order["total_tax_amount"], 234.56)
self.assertEqual(order["total_all_amount"], 1234.56)
self.assertEqual(json.loads(order["raw_json"])["billno"], "SO-REAL-001")
self.assertEqual(item["seq"], 1)
self.assertEqual(item["material_model"], "MODEL-A")
self.assertEqual(item["price_and_tax"], 1.73)
self.assertEqual(item["amount_and_tax"], 1234.56)
self.assertEqual(json.loads(item["raw_json"])["material_masterid_number"], "MAT-001")
self.assertEqual(delivery["delivery_id"], "7690495588165568512")
self.assertEqual(delivery["plan_date"], "2026-06-18")
self.assertEqual(delivery["plan_qty"], 500.0)
self.assertEqual(delivery["transport_lead_time"], 2)
self.assertEqual(json.loads(delivery["raw_json"])["d_remark"], "首批")
def test_cli_sync_payload_can_be_loaded_from_json_file(self):
payload_path = self.vault / "kingdee.json"