From 4dfef7e0991633108f9a9b8b982294c225497174 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 6 Jul 2026 19:19:23 +0000 Subject: [PATCH] feat(business): add Kingdee fact queries Add a structured business facts SQLite layer for Kingdee sales orders, with fixed-template query commands for monthly order counts, sales amount, top customers, and customer trend data. Update README, SETUP, and developer docs to describe configuration, sync commands, query examples, and the boundary between structured facts and wiki ingest. Validation: python3 -m unittest discover -v --- README.md | 22 ++ SETUP.md | 15 + docs/business-facts.md | 58 ++++ obsidian_wiki/business_facts.py | 574 ++++++++++++++++++++++++++++++++ obsidian_wiki/cli.py | 81 +++++ tests/test_business_facts.py | 154 +++++++++ 6 files changed, 904 insertions(+) create mode 100644 docs/business-facts.md create mode 100644 obsidian_wiki/business_facts.py create mode 100644 tests/test_business_facts.py diff --git a/README.md b/README.md index 8a7e121..061bf2f 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ obsidian-wiki list # list the bundled skills obsidian-wiki info # show install paths, version, and config obsidian-wiki file-index rebuild # rebuild fast source-file lookup index obsidian-wiki file-index search "FH52K 图纸" -n 10 +obsidian-wiki business query "2026年6月接了多少单" # deterministic business stats obsidian-wiki setup --project . # also drop project-local skills + AGENTS.md into the current repo obsidian-wiki setup --copy # copy skill files instead of symlinking ``` @@ -313,6 +314,27 @@ obsidian-wiki file-index stage ./drawing.step --project bjd --doc-type drawing Staging is for upload/intake only. The normal ingest flow should still classify the file, archive it under `_sources/archive/`, update source registry/catalog pages, and rebuild the file index. +### Optional: Business Fact Queries + +Operational counts and amounts belong in a structured fact database, not in markdown pages. The `business` CLI stores Kingdee sales orders in `/_meta/business-facts.sqlite` and answers fixed-template statistics questions without letting an LLM invent SQL or change the counting basis. + +```bash +# 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 + +# Ask deterministic statistics questions. +obsidian-wiki business query "2026年6月接了多少单" +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. + ### Optional: Source File MCP Server Agents that support MCP can ask for original engineering files through a local stdio server: diff --git a/SETUP.md b/SETUP.md index cdffdbf..864051c 100644 --- a/SETUP.md +++ b/SETUP.md @@ -126,6 +126,9 @@ Knowledge that's project-specific goes under `projects//`. Knowledge that' | Variable | What it does | Default | |---|---|---| | `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 | `/_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)* | | `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`* | @@ -152,6 +155,18 @@ Knowledge that's project-specific goes under `projects//`. Knowledge that' | `wiki-update` | Sync current project's knowledge into the vault (works from any project) | | `skill-creator` | Create new skills to extend the framework | +## Business Facts + +Use the business fact database for exact operational questions, for example monthly order counts or sales amounts. Do not ingest every order line into the wiki. + +```bash +obsidian-wiki business sync kingdee --input ./kingdee-orders.json +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 `/_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. + ## How It Works No scripts, no dependencies. The skills are markdown files that tell an AI agent *how* to operate on your Obsidian vault: diff --git a/docs/business-facts.md b/docs/business-facts.md new file mode 100644 index 0000000..8fc730e --- /dev/null +++ b/docs/business-facts.md @@ -0,0 +1,58 @@ +# Business Facts + +Business facts are structured operational data used for exact counts, sums, and rankings. They complement the markdown wiki; they do not replace it. + +## Purpose + +Use `/_meta/business-facts.sqlite` for questions where markdown is the wrong source of truth: + +- how many sales orders were received in a month +- monthly sales amount +- top customers for a month +- customer order trend for a fixed period + +The wiki remains the place for context: customer background, project history, sales notes, email-derived signals, and later human-readable summaries. + +## Commands + +```bash +obsidian-wiki business sync kingdee --input ./kingdee-orders.json +obsidian-wiki business sync kingdee --since 2026-07-01T00:00:00Z +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`. + +## Database + +Default path: `/_meta/business-facts.sqlite`. + +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. +- `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. + +## Counting Basis + +V1 statistics use fixed templates, not free-form SQL generated by an LLM. + +- Monthly order count: `COUNT(DISTINCT order_no)` by `order_date`. +- Monthly sales amount: `SUM(total_amount)` by `order_date`. +- Monthly top customers: group by customer name or ID and rank by sales amount. +- Customer trend: group matching customer orders by order month. + +Every answer includes the statistic, counting basis, data source, and last successful sync time. If the last successful sync is before the query date, the answer warns that newer data may be missing. + +## Boundary With Wiki Ingest + +Do not create one markdown page per order or order line. That makes the wiki noisy and produces unreliable statistics. + +If business summaries are added later, generate compact daily or monthly summary sources from this database and pass those summaries through `wiki-ingest`. The database remains the fact source for exact numbers. diff --git a/obsidian_wiki/business_facts.py b/obsidian_wiki/business_facts.py new file mode 100644 index 0000000..1f400bb --- /dev/null +++ b/obsidian_wiki/business_facts.py @@ -0,0 +1,574 @@ +"""Structured business facts for deterministic operational queries. + +The markdown wiki is for distilled knowledge. This module keeps normalized +business facts, such as Kingdee sales orders, in SQLite so count and amount +questions do not depend on LLM-generated SQL or markdown scraping. +""" + +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 + +BUSINESS_DB_RELATIVE_PATH = Path("_meta") / "business-facts.sqlite" + + +def utc_now() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def business_db_path_for_vault(vault_path: str | Path) -> Path: + vault = Path(clean_config_value(str(vault_path))).expanduser() + return vault / BUSINESS_DB_RELATIVE_PATH + + +def connect(db_path: Path) -> sqlite3.Connection: + db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + return conn + + +def init_db(db_path: Path) -> None: + with connect(db_path) as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS sync_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + source TEXT NOT NULL, + status TEXT NOT NULL, + started_at TEXT NOT NULL, + finished_at TEXT, + cursor_from TEXT, + cursor_to TEXT, + records_seen INTEGER NOT NULL DEFAULT 0, + records_upserted INTEGER NOT NULL DEFAULT 0, + error TEXT + ); + + CREATE TABLE IF NOT EXISTS raw_snapshots ( + content_hash TEXT PRIMARY KEY, + source TEXT NOT NULL, + retrieved_at TEXT NOT NULL, + payload_json TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS customers ( + customer_id TEXT PRIMARY KEY, + customer_name TEXT, + updated_at TEXT + ); + + CREATE TABLE IF NOT EXISTS products ( + product_id TEXT PRIMARY KEY, + product_name TEXT, + updated_at TEXT + ); + + CREATE TABLE IF NOT EXISTS orders ( + order_id TEXT PRIMARY KEY, + order_no TEXT NOT NULL UNIQUE, + order_date TEXT NOT NULL, + customer_id TEXT, + customer_name TEXT, + total_amount REAL NOT NULL DEFAULT 0, + currency TEXT, + status TEXT, + updated_at TEXT, + raw_hash TEXT, + created_at TEXT NOT NULL, + synced_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS order_items ( + order_id TEXT NOT NULL, + line_id TEXT NOT NULL, + product_id TEXT, + product_name TEXT, + quantity REAL, + amount REAL, + PRIMARY KEY (order_id, line_id), + FOREIGN KEY (order_id) REFERENCES orders(order_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); + """ + ) + + +def load_kingdee_payload_from_json(path: str | Path) -> tuple[list[dict[str, Any]], Any]: + payload = json.loads(Path(path).expanduser().read_text(encoding="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) -> 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 extract_orders_from_payload(payload: Any) -> list[dict[str, Any]]: + if isinstance(payload, list): + 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"): + value = payload.get(key) + if isinstance(value, list): + return [item for item in value if isinstance(item, dict)] + if isinstance(value, dict): + nested = extract_orders_from_payload(value) + if nested: + return nested + return [] + + +def sync_kingdee_orders( + vault_path: str | Path, + records: list[dict[str, Any]], + *, + source_payload: Any | None = None, + synced_at: str | None = None, + db_path: Path | None = None, + cursor_from: str | None = None, +) -> dict[str, Any]: + db_path = db_path or business_db_path_for_vault(vault_path) + init_db(db_path) + synced_at = synced_at or utc_now() + payload = source_payload if source_payload is not None else {"orders": records} + payload_json = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + raw_hash = "sha256:" + hashlib.sha256(payload_json.encode("utf-8")).hexdigest() + + with connect(db_path) as conn: + cursor = conn.execute( + """ + INSERT INTO sync_runs (source, status, started_at, cursor_from) + VALUES (?, 'running', ?, ?) + """, + ("kingdee", synced_at, cursor_from), + ) + run_id = int(cursor.lastrowid) + try: + conn.execute( + """ + INSERT OR IGNORE INTO raw_snapshots (content_hash, source, retrieved_at, payload_json) + VALUES (?, 'kingdee', ?, ?) + """, + (raw_hash, synced_at, payload_json), + ) + upserted = 0 + for record in records: + normalized = normalize_order(record) + _upsert_order(conn, normalized, raw_hash=raw_hash, synced_at=synced_at) + upserted += 1 + conn.execute( + """ + UPDATE sync_runs + SET status='success', finished_at=?, cursor_to=?, records_seen=?, records_upserted=? + WHERE id=? + """, + (synced_at, synced_at, len(records), upserted, run_id), + ) + except Exception as exc: + conn.execute( + """ + UPDATE sync_runs + SET status='failed', finished_at=?, records_seen=?, error=? + WHERE id=? + """, + (synced_at, len(records), str(exc), run_id), + ) + raise + + return { + "db_path": str(db_path), + "run_id": run_id, + "records_seen": len(records), + "records_upserted": upserted, + "raw_hash": raw_hash, + "synced_at": synced_at, + } + + +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")) + 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")) + 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_name = _text(_first(record, "customer_name", "customerName", "FCustomerName") or _first(customer, "name")) + items = _items(record) + 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", "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")), + "items": items, + } + + +def query_business_question( + vault_path: str | Path, + question: str, + *, + db_path: Path | None = None, + now: str | None = None, +) -> dict[str, Any]: + db_path = db_path or business_db_path_for_vault(vault_path) + init_db(db_path) + q = question.strip() + year, month = parse_month(q, now=now) + lower = q.lower() + if "top" in lower and "customer" in lower or "top客户" in q.lower() or "客户排名" in q: + return monthly_top_customers(db_path, year, month, now=now) + if "amount" in lower or "sales" in lower or "销售额" in q or "金额" in q: + return monthly_sales_amount(db_path, year, month, now=now) + if "trend" in lower or "趋势" in q: + return customer_order_trend(db_path, q, year, month, now=now) + return monthly_order_count(db_path, year, month, now=now) + + +def monthly_order_count(db_path: Path, year: int, month: int, *, now: str | None = None) -> dict[str, Any]: + start, end = month_bounds(year, month) + with connect(db_path) as conn: + row = conn.execute( + "SELECT COUNT(DISTINCT order_no) AS value FROM orders WHERE order_date >= ? AND order_date < ?", + (start, end), + ).fetchone() + count = int(row["value"] or 0) + last_sync = last_successful_sync(db_path) + is_stale = sync_is_stale(last_sync, now=now) + return { + "template": "monthly_order_count", + "data": {"year": year, "month": month, "order_count": count}, + "last_sync": last_sync, + "is_stale": is_stale, + "answer": _answer( + f"Result: {year}-{month:02d} has {count} order(s).", + f"Scope: deduplicated by sales order number, order date from {start} to {end_exclusive_label(end)}.", + last_sync, + is_stale=is_stale, + ), + } + + +def monthly_sales_amount(db_path: Path, year: int, month: int, *, now: str | None = None) -> dict[str, Any]: + start, end = month_bounds(year, month) + with connect(db_path) as conn: + row = conn.execute( + "SELECT COALESCE(SUM(total_amount), 0) AS value FROM orders WHERE order_date >= ? AND order_date < ?", + (start, end), + ).fetchone() + amount = float(row["value"] or 0) + last_sync = last_successful_sync(db_path) + is_stale = sync_is_stale(last_sync, now=now) + return { + "template": "monthly_sales_amount", + "data": {"year": year, "month": month, "sales_amount": amount}, + "last_sync": last_sync, + "is_stale": is_stale, + "answer": _answer( + f"Result: {year}-{month:02d} sales amount is {amount:.2f}.", + f"Scope: sum of order total_amount, order date from {start} to {end_exclusive_label(end)}.", + last_sync, + is_stale=is_stale, + ), + } + + +def monthly_top_customers(db_path: Path, year: int, month: int, *, limit: int = 5, now: str | None = None) -> dict[str, Any]: + start, end = month_bounds(year, month) + with connect(db_path) as conn: + rows = conn.execute( + """ + SELECT COALESCE(NULLIF(customer_name, ''), NULLIF(customer_id, ''), 'Unknown') AS customer, + COUNT(DISTINCT order_no) AS order_count, + COALESCE(SUM(total_amount), 0) AS sales_amount + FROM orders + WHERE order_date >= ? AND order_date < ? + GROUP BY customer + ORDER BY sales_amount DESC, order_count DESC, customer ASC + LIMIT ? + """, + (start, end, limit), + ).fetchall() + customers = [ + {"customer": row["customer"], "order_count": int(row["order_count"] or 0), "sales_amount": float(row["sales_amount"] or 0)} + for row in rows + ] + last_sync = last_successful_sync(db_path) + is_stale = sync_is_stale(last_sync, now=now) + lines = ", ".join(f"{item['customer']} ({item['order_count']} orders, {item['sales_amount']:.2f})" for item in customers) or "no customers" + return { + "template": "monthly_top_customers", + "data": {"year": year, "month": month, "customers": customers}, + "last_sync": last_sync, + "is_stale": is_stale, + "answer": _answer( + f"Result: {year}-{month:02d} top customers: {lines}.", + f"Scope: ranked by sales amount, order date from {start} to {end_exclusive_label(end)}.", + last_sync, + is_stale=is_stale, + ), + } + + +def customer_order_trend(db_path: Path, question: str, year: int, month: int, *, now: str | None = None) -> dict[str, Any]: + customer = _extract_customer_name(question) + start, end = month_bounds(year, month) + with connect(db_path) as conn: + rows = conn.execute( + """ + SELECT substr(order_date, 1, 7) AS month, + COUNT(DISTINCT order_no) AS order_count, + COALESCE(SUM(total_amount), 0) AS sales_amount + FROM orders + WHERE order_date >= ? AND order_date < ? + AND (? = '' OR customer_name LIKE ? OR customer_id LIKE ?) + GROUP BY substr(order_date, 1, 7) + ORDER BY month ASC + """, + (start, end, customer, f"%{customer}%", f"%{customer}%"), + ).fetchall() + trend = [{"month": row["month"], "order_count": int(row["order_count"] or 0), "sales_amount": float(row["sales_amount"] or 0)} for row in rows] + last_sync = last_successful_sync(db_path) + is_stale = sync_is_stale(last_sync, now=now) + return { + "template": "customer_order_trend", + "data": {"customer": customer, "year": year, "month": month, "trend": trend}, + "last_sync": last_sync, + "is_stale": is_stale, + "answer": _answer( + f"Result: customer trend for {customer or 'all customers'} in {year}-{month:02d}: {trend}.", + f"Scope: grouped by order month, order date from {start} to {end_exclusive_label(end)}.", + last_sync, + is_stale=is_stale, + ), + } + + +def parse_month(text: str, *, now: str | None = None) -> tuple[int, int]: + match = re.search(r"(20\d{2})\s*[-/年.]\s*(\d{1,2})", text) + if match: + year = int(match.group(1)) + month = int(match.group(2)) + else: + month_match = re.search(r"(? 12: + raise RuntimeError(f"invalid month: {month}") + return year, month + + +def month_bounds(year: int, month: int) -> tuple[str, str]: + start = f"{year:04d}-{month:02d}-01" + if month == 12: + end = f"{year + 1:04d}-01-01" + else: + end = f"{year:04d}-{month + 1:02d}-01" + return start, end + + +def end_exclusive_label(end: str) -> str: + return f"before {end}" + + +def last_successful_sync(db_path: Path) -> str: + with connect(db_path) as conn: + row = conn.execute( + """ + SELECT finished_at + FROM sync_runs + WHERE source='kingdee' AND status='success' + ORDER BY finished_at DESC, id DESC + LIMIT 1 + """ + ).fetchone() + return str(row["finished_at"] or "") if row else "" + + +def sync_is_stale(last_sync: str, *, now: str | None = None) -> bool: + if not last_sync: + return True + return last_sync[:10] < (now or utc_now())[:10] + + +def _answer(result: str, scope: str, last_sync: str, *, is_stale: bool) -> str: + parts = [result, scope, "Data source: Kingdee API."] + if last_sync: + parts.append(f"Last sync: {last_sync}.") + if is_stale: + parts.append(f"Note: result may not include data after {last_sync}.") + else: + parts.append("Last sync: never.") + return "\n".join(parts) + + +def _upsert_order(conn: sqlite3.Connection, order: dict[str, Any], *, raw_hash: str, synced_at: str) -> None: + customer_id = order["customer_id"] + product_seen_at = order["updated_at"] or synced_at + if customer_id: + conn.execute( + """ + INSERT INTO customers (customer_id, customer_name, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(customer_id) DO UPDATE SET + customer_name=excluded.customer_name, + updated_at=excluded.updated_at + """, + (customer_id, order["customer_name"], product_seen_at), + ) + 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + 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, + 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, + ), + ) + conn.execute("DELETE FROM order_items WHERE order_id=?", (order["order_id"],)) + for item in order["items"]: + if item["product_id"]: + conn.execute( + """ + INSERT INTO products (product_id, product_name, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(product_id) DO UPDATE SET + product_name=excluded.product_name, + updated_at=excluded.updated_at + """, + (item["product_id"], item["product_name"], product_seen_at), + ) + conn.execute( + """ + INSERT INTO order_items (order_id, line_id, product_id, product_name, quantity, amount) + VALUES (?, ?, ?, ?, ?, ?) + """, + (order["order_id"], item["line_id"], item["product_id"], item["product_name"], item["quantity"], item["amount"]), + ) + + +def _items(record: dict[str, Any]) -> list[dict[str, Any]]: + raw_items = None + for key in ("items", "entries", "order_items", "FEntity"): + if isinstance(record.get(key), list): + raw_items = record[key] + break + if raw_items is None: + raw_items = [] + items = [] + for index, raw in enumerate(raw_items, start=1): + 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")) + items.append( + { + "line_id": _text(_first(raw, "line_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")), + } + ) + return items + + +def _first(record: dict[str, Any], *keys: str) -> Any: + for key in keys: + if key in record and record[key] not in (None, ""): + return record[key] + return None + + +def _text(value: Any) -> str: + return "" if value is None else str(value).strip() + + +def _date_text(value: Any) -> str: + text = _text(value) + return text[:10] if text else "" + + +def _date_time_text(value: Any) -> str: + return _text(value) + + +def _float(value: Any) -> float: + if value in (None, ""): + return 0.0 + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + + +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/obsidian_wiki/cli.py b/obsidian_wiki/cli.py index 8d4b1df..1237e79 100644 --- a/obsidian_wiki/cli.py +++ b/obsidian_wiki/cli.py @@ -18,6 +18,13 @@ from pathlib import Path from obsidian_wiki import __version__ from obsidian_wiki.access_control import audit_profile, export_profile, list_profiles +from obsidian_wiki.business_facts import ( + business_db_path_for_vault, + fetch_kingdee_payload, + load_kingdee_payload_from_json, + query_business_question, + sync_kingdee_orders, +) from obsidian_wiki.file_assets import ( clean_config_value, db_path_for_vault, @@ -419,6 +426,20 @@ def _resolve_file_asset_db(args: argparse.Namespace, vault_path: str) -> Path: return db_path +def _resolve_business_db(args: argparse.Namespace, vault_path: str) -> Path: + raw = ( + getattr(args, "db", None) + or os.environ.get("OBSIDIAN_BUSINESS_FACTS_DB") + or _read_config_value("OBSIDIAN_BUSINESS_FACTS_DB") + ) + if not raw: + return business_db_path_for_vault(vault_path) + db_path = Path(clean_config_value(raw)).expanduser() + if not db_path.is_absolute(): + db_path = Path(vault_path).expanduser() / db_path + return db_path + + def cmd_file_index_init(args: argparse.Namespace) -> int: vault_path = _require_vault(args) db_path = _resolve_file_asset_db(args, vault_path) @@ -480,6 +501,43 @@ def cmd_file_index_stage(args: argparse.Namespace) -> int: return 0 +def cmd_business_sync_kingdee(args: argparse.Namespace) -> int: + vault_path = _require_vault(args) + 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) + result = sync_kingdee_orders( + vault_path, + records, + source_payload=payload, + db_path=_resolve_business_db(args, vault_path), + cursor_from=args.since, + ) + if args.json: + print(json.dumps(result, ensure_ascii=False, indent=2)) + else: + print(f"✅ Synced Kingdee orders: {result['records_upserted']} record(s)") + print(f" DB: {result['db_path']}") + print(f" Last sync: {result['synced_at']}") + return 0 + + +def cmd_business_query(args: argparse.Namespace) -> int: + vault_path = _require_vault(args) + question = " ".join(args.question).strip() + result = query_business_question( + vault_path, + question, + db_path=_resolve_business_db(args, vault_path), + ) + if args.json: + print(json.dumps(result, ensure_ascii=False, indent=2)) + else: + print(result["answer"]) + return 0 + + def cmd_mcp_server(args: argparse.Namespace) -> int: from obsidian_wiki.mcp_server import main as mcp_main @@ -602,6 +660,29 @@ def build_parser() -> argparse.ArgumentParser: ftp.add_argument("--visibility", default="internal", help="initial visibility") ftp.set_defaults(func=cmd_file_index_stage) + bp = sub.add_parser("business", help="sync and query structured business facts") + business_sub = bp.add_subparsers(dest="business_command") + + bsp = business_sub.add_parser("sync", help="sync business facts from an external system") + business_sync_sub = bsp.add_subparsers(dest="business_sync_source") + + bkp = business_sync_sub.add_parser("kingdee", help="sync Kingdee sales orders") + _add_vault_arg(bkp) + bkp.add_argument("--db", metavar="PATH", help="SQLite DB path (defaults to /_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("--since", help="optional updated_since cursor passed to the API and sync log") + bkp.add_argument("--json", action="store_true", help="print machine-readable sync result") + bkp.set_defaults(func=cmd_business_sync_kingdee) + + bqp = business_sub.add_parser("query", help="answer fixed-template business statistics questions") + _add_vault_arg(bqp) + bqp.add_argument("--db", metavar="PATH", help="SQLite DB path (defaults to /_meta/business-facts.sqlite)") + bqp.add_argument("--json", action="store_true", help="print machine-readable query result") + bqp.add_argument("question", nargs="+", help="question, e.g. '2026-06 order count'") + bqp.set_defaults(func=cmd_business_query) + mp = sub.add_parser("mcp-server", help="run the MCP server for safe source-file lookup") _add_file_index_storage_args(mp) mp.add_argument("--profile", help="bot profile name for ACL checks") diff --git a/tests/test_business_facts.py b/tests/test_business_facts.py new file mode 100644 index 0000000..2af724b --- /dev/null +++ b/tests/test_business_facts.py @@ -0,0 +1,154 @@ +import json +import tempfile +import unittest +from contextlib import redirect_stdout +from io import StringIO +from pathlib import Path + +from obsidian_wiki.business_facts import ( + business_db_path_for_vault, + query_business_question, + sync_kingdee_orders, +) + + +class BusinessFactsTests(unittest.TestCase): + def setUp(self): + self.tempdir = tempfile.TemporaryDirectory() + self.vault = Path(self.tempdir.name) + (self.vault / "_meta").mkdir() + self.db_path = business_db_path_for_vault(self.vault) + + def tearDown(self): + self.tempdir.cleanup() + + def test_monthly_order_count_deduplicates_order_lines(self): + sync_kingdee_orders( + self.vault, + [ + { + "order_id": "1001", + "order_no": "SO-001", + "order_date": "2026-06-03", + "customer_id": "C01", + "customer_name": "Customer A", + "total_amount": 1200, + "updated_at": "2026-06-03T08:00:00Z", + "items": [ + {"line_id": "1", "product_id": "P01", "product_name": "Part 1", "quantity": 10, "amount": 500}, + {"line_id": "2", "product_id": "P02", "product_name": "Part 2", "quantity": 20, "amount": 700}, + ], + }, + { + "order_id": "1002", + "order_no": "SO-002", + "order_date": "2026-06-15", + "customer_id": "C02", + "customer_name": "Customer B", + "total_amount": 900, + "updated_at": "2026-06-15T08:00:00Z", + "items": [ + {"line_id": "1", "product_id": "P03", "product_name": "Part 3", "quantity": 5, "amount": 900}, + ], + }, + { + "order_id": "1003", + "order_no": "SO-003", + "order_date": "2026-07-01", + "customer_id": "C01", + "customer_name": "Customer A", + "total_amount": 300, + "updated_at": "2026-07-01T08:00:00Z", + }, + ], + source_payload={"orders": 3}, + synced_at="2026-07-06T09:00:00Z", + ) + + result = query_business_question(self.vault, "2026-06 order count") + + self.assertEqual(result["template"], "monthly_order_count") + self.assertEqual(result["data"]["order_count"], 2) + self.assertIn("2", result["answer"]) + self.assertIn("sales order number", result["answer"]) + self.assertIn("2026-07-06T09:00:00Z", result["answer"]) + + def test_repeated_sync_updates_existing_orders_without_duplicates(self): + first = [ + { + "order_id": "1001", + "order_no": "SO-001", + "order_date": "2026-06-03", + "customer_id": "C01", + "customer_name": "Customer A", + "total_amount": 1200, + "updated_at": "2026-06-03T08:00:00Z", + } + ] + second = [{**first[0], "total_amount": 1500, "updated_at": "2026-06-04T08:00:00Z"}] + + sync_kingdee_orders(self.vault, first, source_payload={"orders": first}, synced_at="2026-07-06T09:00:00Z") + sync_kingdee_orders(self.vault, second, source_payload={"orders": second}, synced_at="2026-07-06T10: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"], 1500.0) + self.assertIn("2026-07-06T10:00:00Z", amount["answer"]) + + def test_query_warns_when_last_sync_is_before_today(self): + sync_kingdee_orders( + self.vault, + [ + { + "order_id": "1001", + "order_no": "SO-001", + "order_date": "2026-06-03", + "customer_name": "Customer A", + "total_amount": 1200, + } + ], + source_payload={"orders": 1}, + synced_at="2026-07-03T09:00:00Z", + ) + + result = query_business_question(self.vault, "2026-06 order count", now="2026-07-06T00:00:00Z") + + self.assertTrue(result["is_stale"]) + self.assertIn("may not include data after 2026-07-03T09:00:00Z", result["answer"]) + + def test_cli_sync_payload_can_be_loaded_from_json_file(self): + payload_path = self.vault / "kingdee.json" + payload_path.write_text( + json.dumps( + { + "orders": [ + { + "order_id": "1001", + "order_no": "SO-001", + "order_date": "2026-06-03", + "customer_name": "Customer A", + "total_amount": 1200, + } + ] + } + ), + encoding="utf-8", + ) + + from obsidian_wiki.cli import main + + with redirect_stdout(StringIO()): + sync_code = main(["business", "sync", "kingdee", "--vault", str(self.vault), "--input", str(payload_path)]) + query_output = StringIO() + with redirect_stdout(query_output): + query_code = main(["business", "query", "2026-06 order count", "--vault", str(self.vault), "--json"]) + + self.assertEqual(sync_code, 0) + self.assertEqual(query_code, 0) + self.assertEqual(json.loads(query_output.getvalue())["data"]["order_count"], 1) + + +if __name__ == "__main__": + unittest.main()