diff --git a/docs/business-facts.md b/docs/business-facts.md index 1b531e3..1b2f9dd 100644 --- a/docs/business-facts.md +++ b/docs/business-facts.md @@ -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 diff --git a/obsidian_wiki/business_facts.py b/obsidian_wiki/business_facts.py index 64bd465..7f9a9ba 100644 --- a/obsidian_wiki/business_facts.py +++ b/obsidian_wiki/business_facts.py @@ -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 "" diff --git a/tests/test_business_facts.py b/tests/test_business_facts.py index 6b47916..bd7f519 100644 --- a/tests/test_business_facts.py +++ b/tests/test_business_facts.py @@ -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"