diff --git a/.skills/wiki-query/SKILL.md b/.skills/wiki-query/SKILL.md index 52b0e4d..94d3bd1 100644 --- a/.skills/wiki-query/SKILL.md +++ b/.skills/wiki-query/SKILL.md @@ -52,6 +52,30 @@ Also decide the **mode**: - **Index-only mode** — triggered by "quick answer", "just scan", "don't read the pages", "fast lookup". Stops at Step 3. Answers from frontmatter + `index.md` only. - **Normal mode** — the full tiered pipeline below. +### Step 1a: Business Facts Route (LLM-decided, before wiki retrieval) + +If filtered mode is active, skip this step. The business fact database is operator-internal and should not be used for public/user-facing answers. + +Before searching wiki pages, decide semantically whether the user is asking for exact operational sales-order facts instead of narrative knowledge. Treat it as a business facts query when the intent is about counts, sums, averages, rankings, trends, lists, or breakdowns of Kingdee sales orders, customers, products/materials, salespeople, organizations, departments, projects, warehouses, statuses, currencies, or delivery plans. + +If it is a business facts query: + +1. Load the machine-readable schema and template catalog: + ```bash + obsidian-wiki business schema --json + ``` +2. Prefer a template whenever one fits. Convert the user's natural date range into either `--year/--month` or `--start-date` plus exclusive `--end-date`, then run: + ```bash + obsidian-wiki business template --vault "$OBSIDIAN_VAULT_PATH" --json + ``` + Common parameters include `--customer`, `--product`, `--staff`, `--org`, `--department`, `--project`, `--warehouse`, `--billstatus`, `--orderstatus`, `--closestatus`, `--currency`, `--limit`, `--granularity day|month`, and `--staff-role-scope primary|all`. Salesperson questions default to `primary`; choose `all` only when the user asks to include secondary sales/follow-up roles. +3. If no template fits, generate one guarded read-only SQL statement from the schema. Use only `SELECT` or `WITH ... SELECT`, only the documented business tables, never `raw_json` or `raw_snapshots`, and never write/DDL/PRAGMA. Then run: + ```bash + obsidian-wiki business sql --vault "$OBSIDIAN_VAULT_PATH" "" --json + ``` +4. If the template or guarded SQL returns an answer, answer from that JSON result as structured business facts. Do not cite wiki pages unless you also used wiki pages for business context. Mention stale sync warnings when `is_stale` is true. +5. If the CLI/database is unavailable or the question is not a business facts query, continue with the normal wiki retrieval protocol below. + ### Step 2: Index Pass (cheap) Build a candidate set *without opening any page bodies*: diff --git a/README.md b/README.md index 3506175..65102f1 100644 --- a/README.md +++ b/README.md @@ -33,12 +33,14 @@ 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 business schema --json # business facts catalog used by wiki-query 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 ``` -When working from a cloned repo, `scripts/obsidian-wiki-ops.py` is the single repo-local operations entrypoint. It wraps common `obsidian-wiki` commands, Kingdee business sync/query, the business sync timer, daily update, and the managed MCP service: +For operational questions such as "2026 年 6 月接了多少单", ask your agent with `wiki-query`. The `wiki-query` skill decides whether the question needs structured business facts, then uses the business template catalog or guarded read-only SQL before falling back to normal wiki retrieval. + +When working from a cloned repo, `scripts/obsidian-wiki-ops.py` is the single repo-local operations entrypoint. It wraps common `obsidian-wiki` commands, Kingdee business sync/templates, the business sync timer, daily update, and the managed MCP service: ```bash scripts/obsidian-wiki-ops.py help @@ -325,7 +327,7 @@ Staging is for upload/intake only. The normal ingest flow should still classify ### 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. +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`. Normal users should ask through `wiki-query`; the skill decides when to use structured business facts, prefers templates, and only falls back to guarded read-only SQL when no template fits. ```bash # Sync from an exported or API-fetched Kingdee JSON payload. @@ -345,16 +347,20 @@ scripts/obsidian-wiki-ops.py kingdee sync \ # Or fetch incrementally by Kingdee modify time. scripts/obsidian-wiki-ops.py kingdee sync --incremental -# Ask deterministic statistics questions. -scripts/obsidian-wiki-ops.py kingdee query "2026年6月接了多少单" -scripts/obsidian-wiki-ops.py kingdee query "2026-06 sales amount" -scripts/obsidian-wiki-ops.py kingdee query "2026-06 top customer" -scripts/obsidian-wiki-ops.py kingdee query "徐丽最近一周订单" -scripts/obsidian-wiki-ops.py kingdee query "徐丽最近一周销售额" --all-staff +# Inspect the schema/template catalog used by wiki-query. +scripts/obsidian-wiki-ops.py kingdee schema --json + +# Run explicit templates for debugging or automation. +scripts/obsidian-wiki-ops.py kingdee template order_count --vault /path/to/vault --year 2026 --month 6 --json +scripts/obsidian-wiki-ops.py kingdee template sales_amount --vault /path/to/vault --year 2026 --month 6 --json +scripts/obsidian-wiki-ops.py kingdee template top_customers --vault /path/to/vault --year 2026 --month 6 --limit 10 --json +scripts/obsidian-wiki-ops.py kingdee template staff_order_list --vault /path/to/vault --start-date 2026-06-30 --end-date 2026-07-07 --staff 徐丽 --json + +# Run guarded read-only SQL only when templates do not fit. scripts/obsidian-wiki-ops.py kingdee sql "SELECT COUNT(*) AS order_count FROM orders" ``` -V1 uses fixed templates for monthly order count, monthly sales amount, monthly top customers, customer trend, and salesperson order list/count/amount queries. Salesperson queries default to the main Kingdee salesperson and can include secondary/follow-up roles with `--all-staff`. If no template fits, `business sql` and `business query --fallback-sql` allow guarded read-only SQL against structured tables only; raw JSON and write statements are blocked. 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. +Templates cover order counts, sales amounts, customer/product counts, quantity, averages, trends, rankings, breakdowns, order lists, customer/product/staff summaries, status breakdowns, and delivery-plan queries. Salesperson templates default to the main Kingdee salesperson and can include secondary/follow-up roles with `--staff-role-scope all`. If no template fits, `business sql` allows guarded read-only SQL against structured tables only; raw JSON and write statements are blocked. Detailed order rows stay in SQLite; the wiki remains for business context, customer/project knowledge, and cross-linked notes. 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. For daily unattended sync on Linux: diff --git a/SETUP.md b/SETUP.md index c8d9bbd..23d4335 100644 --- a/SETUP.md +++ b/SETUP.md @@ -156,7 +156,7 @@ Knowledge that's project-specific goes under `projects//`. Knowledge that' ## 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. +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. Ask normal questions through `wiki-query`; it will decide when to use structured business facts. ```bash scripts/obsidian-wiki-ops.py kingdee sync --input ./kingdee-orders.json @@ -171,10 +171,10 @@ scripts/obsidian-wiki-ops.py kingdee sync \ --billstatus C scripts/obsidian-wiki-ops.py kingdee sync --incremental -scripts/obsidian-wiki-ops.py kingdee query "2026年6月接了多少单" -scripts/obsidian-wiki-ops.py kingdee query "2026-06 sales amount" -scripts/obsidian-wiki-ops.py kingdee query "徐丽最近一周订单" -scripts/obsidian-wiki-ops.py kingdee query "徐丽最近一周销售额" --all-staff +scripts/obsidian-wiki-ops.py kingdee schema --json +scripts/obsidian-wiki-ops.py kingdee template order_count --vault /path/to/vault --year 2026 --month 6 --json +scripts/obsidian-wiki-ops.py kingdee template sales_amount --vault /path/to/vault --year 2026 --month 6 --json +scripts/obsidian-wiki-ops.py kingdee template staff_order_list --vault /path/to/vault --start-date 2026-06-30 --end-date 2026-07-07 --staff 徐丽 --json scripts/obsidian-wiki-ops.py kingdee sql "SELECT COUNT(*) AS order_count FROM orders" scripts/obsidian-wiki-ops.py timer install --vault /path/to/vault @@ -182,7 +182,7 @@ scripts/obsidian-wiki-ops.py timer install --vault /path/to/vault 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 `/_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. Fixed templates are tried first; guarded SQL fallback is available for structured tables only. Salesperson queries default to the main salesperson and use `--all-staff` to include sales 2/3/4 and follow-up roles. The Linux timer uses `systemd --user`, runs daily by default, and syncs incrementally with a 48-hour lookback. +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. `wiki-query` loads the business schema, chooses a template when one fits, and uses guarded read-only SQL only for gaps. Salesperson templates default to the main salesperson and use `--staff-role-scope all` to include sales 2/3/4 and follow-up roles. The Linux timer uses `systemd --user`, runs daily by default, and syncs incrementally with a 48-hour lookback. ## How It Works diff --git a/docs/business-facts.md b/docs/business-facts.md index f4fab7d..efeb654 100644 --- a/docs/business-facts.md +++ b/docs/business-facts.md @@ -8,11 +8,11 @@ Use `/_meta/business-facts.sqlite` for questions where markdown is the wr - how many sales orders were received in a month - monthly sales amount -- top customers for a month -- customer order trend for a fixed period -- sales order counts, lists, and amounts by salesperson +- customer, product, salesperson, organization, department, project, warehouse, status, and currency rankings or breakdowns +- customer, product, and salesperson summaries +- sales order counts, lists, trends, quantities, averages, and delivery-plan quantities -The wiki remains the place for context: customer background, project history, sales notes, email-derived signals, and later human-readable summaries. +The wiki remains the place for context: customer background, project history, sales notes, email-derived signals, and cross-linked narrative knowledge. ## Commands @@ -30,13 +30,15 @@ scripts/obsidian-wiki-ops.py kingdee sync \ scripts/obsidian-wiki-ops.py kingdee sync --start-modifytime "2026-07-01 00:00:00" scripts/obsidian-wiki-ops.py kingdee sync --incremental -scripts/obsidian-wiki-ops.py kingdee query "2026年6月接了多少单" -scripts/obsidian-wiki-ops.py kingdee query "2026-06 sales amount" -scripts/obsidian-wiki-ops.py kingdee query "2026-06 top customer" -scripts/obsidian-wiki-ops.py kingdee query "徐丽最近一周订单" -scripts/obsidian-wiki-ops.py kingdee query "徐丽最近一周接了多少单" -scripts/obsidian-wiki-ops.py kingdee query "徐丽最近一周销售额" --all-staff -scripts/obsidian-wiki-ops.py kingdee query "按组织列出订单" --fallback-sql "SELECT org_name, COUNT(*) AS order_count FROM orders GROUP BY org_name" + +# wiki-query uses this schema/template catalog to answer business questions. +scripts/obsidian-wiki-ops.py kingdee schema --json +scripts/obsidian-wiki-ops.py kingdee template order_count --vault /path/to/vault --year 2026 --month 6 --json +scripts/obsidian-wiki-ops.py kingdee template sales_amount --vault /path/to/vault --year 2026 --month 6 --json +scripts/obsidian-wiki-ops.py kingdee template top_products --vault /path/to/vault --year 2026 --month 6 --limit 10 --json +scripts/obsidian-wiki-ops.py kingdee template staff_summary --vault /path/to/vault --start-date 2026-06-30 --end-date 2026-07-07 --staff 徐丽 --staff-role-scope all --json + +# Guarded SQL is for gaps where no template fits. scripts/obsidian-wiki-ops.py kingdee sql "SELECT COUNT(*) AS order_count FROM orders" ``` @@ -91,21 +93,32 @@ Order upsert key: `order_id`, falling back to `order_no` when no explicit ID is `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 +## Query Flow And Counting Basis -V1 statistics use fixed templates, not free-form SQL generated by an LLM. +`wiki-query` decides semantically whether a user question needs business facts. When it does, it loads `obsidian-wiki business schema --json`, chooses a template if one fits, and only then uses guarded SQL for gaps. -- 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. -- Staff order list/count/amount: join `orders` to `order_staff`, defaulting to `primary_sales`. +Template families include: + +- Order totals: `order_count`, `sales_amount`, `customer_count`, `product_count`, `quantity_sum`, `avg_order_amount`. +- Trends: `order_trend`, `sales_trend`, `quantity_trend`, grouped by day or month. +- Rankings and breakdowns: customers, products, staff, organizations, departments, projects, warehouses, statuses, currencies. +- Lists: orders, customer orders, product orders, staff orders. +- Summaries: customer, product, and staff summaries. +- Delivery plans: delivery-plan count, quantity, and list by planned date. + +Core counting basis: + +- Order count: `COUNT(DISTINCT order_no)` by `orders.order_date`. +- Sales amount: `SUM(orders.total_amount)` by `orders.order_date`. +- Ordered quantity: `SUM(order_items.quantity)` by `orders.order_date`. +- Delivery-plan quantity: `SUM(order_deliveries.plan_qty)` by `order_deliveries.plan_date`. +- Staff templates join `orders` to `order_staff`, defaulting to `primary_sales`. 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. -Salesperson queries default to the main Kingdee salesperson (`operator_operatorname`). Use `--all-staff`, or phrases such as "全部角色" or "含跟单", to include sales 2/3/4 and follow-up roles. +Salesperson templates default to the main Kingdee salesperson (`operator_operatorname`). Use `--staff-role-scope all`, or wording such as "全部角色" or "含跟单", to include sales 2/3/4 and follow-up roles. -If no fixed template matches, callers may provide guarded SQL through `business sql` or `business query --fallback-sql`. The SQL fallback is still restricted: +If no template matches, `wiki-query` may generate guarded SQL and execute it through `business sql`. The SQL fallback is still restricted: - only one `SELECT` or `WITH ... SELECT` statement - only structured tables: `orders`, `order_items`, `order_deliveries`, `order_staff`, `customers`, `products`, `sync_runs` @@ -126,5 +139,4 @@ The 48-hour lookback intentionally re-fetches recent orders so late edits and bo ## 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. +The database remains the fact source for exact numbers; the wiki remains the context source. diff --git a/obsidian_wiki/business_facts.py b/obsidian_wiki/business_facts.py index b80fde1..9b16519 100644 --- a/obsidian_wiki/business_facts.py +++ b/obsidian_wiki/business_facts.py @@ -116,6 +116,45 @@ FORBIDDEN_SQL_COLUMNS = {"raw_json"} SQL_DEFAULT_LIMIT = 100 SQL_MAX_LIMIT = 1000 +BUSINESS_TEMPLATE_NAMES = ( + "order_count", + "sales_amount", + "customer_count", + "product_count", + "quantity_sum", + "avg_order_amount", + "order_trend", + "sales_trend", + "quantity_trend", + "top_customers", + "top_products", + "top_staff", + "top_orgs", + "top_departments", + "top_projects", + "top_warehouses", + "breakdown_by_customer", + "breakdown_by_product", + "breakdown_by_staff", + "breakdown_by_org", + "breakdown_by_department", + "breakdown_by_status", + "breakdown_by_currency", + "order_list", + "customer_order_list", + "product_order_list", + "staff_order_list", + "customer_summary", + "product_summary", + "staff_summary", + "delivery_plan_count", + "delivery_plan_quantity", + "delivery_plan_list", + "order_status_breakdown", + "billstatus_breakdown", + "close_status_breakdown", +) + def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") @@ -376,6 +415,886 @@ def extract_orders_from_payload(payload: Any) -> list[dict[str, Any]]: return [] +def business_schema() -> dict[str, Any]: + return { + "purpose": "Structured business facts for exact Kingdee sales-order statistics. Use this before free-form wiki retrieval when the user asks for operational counts, sums, rankings, trends, or lists.", + "tables": [ + {"name": "orders", "fields": [field for field in ORDER_COLUMNS if field != "raw_json"]}, + {"name": "order_items", "fields": [field for field in ORDER_ITEM_COLUMNS if field != "raw_json"]}, + {"name": "order_deliveries", "fields": [field for field in ORDER_DELIVERY_COLUMNS if field != "raw_json"]}, + {"name": "order_staff", "fields": list(ORDER_STAFF_COLUMNS)}, + {"name": "customers", "fields": ["customer_id", "customer_name", "updated_at"]}, + {"name": "products", "fields": ["product_id", "product_name", "updated_at"]}, + {"name": "sync_runs", "fields": ["id", "source", "status", "started_at", "finished_at", "cursor_from", "cursor_to", "records_seen", "records_upserted", "error"]}, + ], + "metric_guidance": { + "order_count": "Use COUNT(DISTINCT order_no) for sales-order counts.", + "sales_amount": "Use orders.total_amount for tax-inclusive order amount; total_net_amount and total_tax_amount are available when that split matters.", + "quantity": "Use order_items.quantity for ordered material quantity; use order_deliveries.plan_qty only for delivery-plan questions.", + "date": "Use orders.order_date for order statistics and order_deliveries.plan_date for planned demand/delivery-plan questions.", + "staff": "Use order_staff.role='primary_sales' by default; use staff_role_scope=all when the user asks to include secondary sales or follow-up roles.", + }, + "templates": [_template_schema(name) for name in BUSINESS_TEMPLATE_NAMES], + "sql_guardrails": { + "allowed_statements": ["SELECT", "WITH ... SELECT"], + "allowed_tables": sorted(ALLOWED_SQL_TABLES), + "forbidden_columns": sorted(FORBIDDEN_SQL_COLUMNS), + "forbidden_tables": ["raw_snapshots"], + "default_limit": SQL_DEFAULT_LIMIT, + "max_limit": SQL_MAX_LIMIT, + }, + } + + +def _template_schema(name: str) -> dict[str, Any]: + descriptions = { + "order_count": "Count distinct sales orders in a date range.", + "sales_amount": "Sum sales order amount in a date range.", + "customer_count": "Count distinct customers in a date range.", + "product_count": "Count distinct products/materials in a date range.", + "quantity_sum": "Sum ordered material quantity in a date range.", + "avg_order_amount": "Average order amount in a date range.", + "order_trend": "Trend of order count by day or month.", + "sales_trend": "Trend of sales amount by day or month.", + "quantity_trend": "Trend of ordered material quantity by day or month.", + "top_customers": "Rank customers by sales amount and order count.", + "top_products": "Rank products by quantity and amount.", + "top_staff": "Rank staff by order count and sales amount.", + "top_orgs": "Rank sales organizations.", + "top_departments": "Rank sales departments.", + "top_projects": "Rank item project names.", + "top_warehouses": "Rank item warehouses.", + "breakdown_by_customer": "Break down orders by customer.", + "breakdown_by_product": "Break down order items by product.", + "breakdown_by_staff": "Break down orders by staff.", + "breakdown_by_org": "Break down orders by sales organization.", + "breakdown_by_department": "Break down orders by department.", + "breakdown_by_status": "Break down orders by combined order/bill/status value.", + "breakdown_by_currency": "Break down orders by currency.", + "order_list": "List sales orders in a date range.", + "customer_order_list": "List orders for one customer.", + "product_order_list": "List orders containing one product/material.", + "staff_order_list": "List orders owned by one staff member.", + "customer_summary": "Summarize one customer.", + "product_summary": "Summarize one product/material.", + "staff_summary": "Summarize one staff member.", + "delivery_plan_count": "Count delivery-plan rows by plan_date.", + "delivery_plan_quantity": "Sum planned delivery quantity by plan_date.", + "delivery_plan_list": "List delivery-plan rows by plan_date.", + "order_status_breakdown": "Break down by Kingdee orderstatus.", + "billstatus_breakdown": "Break down by Kingdee billstatus.", + "close_status_breakdown": "Break down by Kingdee closestatus.", + } + required: list[str] = [] + if name in {"customer_order_list", "customer_summary"}: + required.append("customer") + if name in {"product_order_list", "product_summary"}: + required.append("product") + if name in {"staff_order_list", "staff_summary"}: + required.append("staff") + return { + "name": name, + "description": descriptions[name], + "date_parameters": ["year/month", "start_date + end_date_exclusive"], + "required_parameters": required, + "optional_parameters": [ + "limit", + "granularity=day|month", + "staff_role_scope=primary|all", + "customer", + "product", + "staff", + "org", + "department", + "project", + "warehouse", + "billstatus", + "orderstatus", + "closestatus", + "currency", + ], + } + + +def execute_business_template(db_path: Path, template_name: str, **params: Any) -> dict[str, Any]: + init_db(db_path) + name = template_name.strip().lower().replace("-", "_") + if name not in BUSINESS_TEMPLATE_NAMES: + raise RuntimeError(f"unknown business template: {template_name}") + clean_params = {key: value for key, value in params.items() if value not in (None, "")} + start, end = _template_date_range(name, clean_params) + limit = _safe_limit(clean_params.get("limit", 10)) + granularity = str(clean_params.get("granularity", "month")).lower() + if granularity not in {"day", "month"}: + raise RuntimeError("granularity must be day or month") + + if name in {"order_count", "sales_amount", "customer_count", "avg_order_amount"}: + return _order_scalar_template(db_path, name, start, end, clean_params) + if name in {"product_count", "quantity_sum"}: + return _item_scalar_template(db_path, name, start, end, clean_params) + if name in {"order_trend", "sales_trend"}: + return _order_trend_template(db_path, name, start, end, clean_params, granularity) + if name == "quantity_trend": + return _item_trend_template(db_path, name, start, end, clean_params, granularity) + if name in {"top_customers", "breakdown_by_customer"}: + return _order_group_template(db_path, name, start, end, clean_params, "customer", limit) + if name in {"top_orgs", "breakdown_by_org"}: + return _order_group_template(db_path, name, start, end, clean_params, "org", limit) + if name in {"top_departments", "breakdown_by_department"}: + return _order_group_template(db_path, name, start, end, clean_params, "department", limit) + if name == "breakdown_by_currency": + return _order_group_template(db_path, name, start, end, clean_params, "currency", limit) + if name in {"breakdown_by_status", "order_status_breakdown", "billstatus_breakdown", "close_status_breakdown"}: + status_group = { + "breakdown_by_status": "status", + "order_status_breakdown": "orderstatus", + "billstatus_breakdown": "billstatus", + "close_status_breakdown": "closestatus", + }[name] + return _order_group_template(db_path, name, start, end, clean_params, status_group, limit) + if name in {"top_products", "breakdown_by_product", "top_projects", "top_warehouses"}: + group = {"top_products": "product", "breakdown_by_product": "product", "top_projects": "project", "top_warehouses": "warehouse"}[name] + return _item_group_template(db_path, name, start, end, clean_params, group, limit) + if name in {"top_staff", "breakdown_by_staff"}: + return _staff_group_template(db_path, name, start, end, clean_params, limit) + if name in {"order_list", "customer_order_list"}: + if name == "customer_order_list": + _require_param(name, clean_params, "customer") + return _order_list_template(db_path, name, start, end, clean_params, limit) + if name == "product_order_list": + _require_param(name, clean_params, "product") + return _product_order_list_template(db_path, name, start, end, clean_params, limit) + if name == "staff_order_list": + _require_param(name, clean_params, "staff") + return _staff_template(db_path, name, start, end, clean_params, list_mode=True, limit=limit) + if name == "customer_summary": + _require_param(name, clean_params, "customer") + return _customer_summary_template(db_path, name, start, end, clean_params) + if name == "product_summary": + _require_param(name, clean_params, "product") + return _product_summary_template(db_path, name, start, end, clean_params) + if name == "staff_summary": + _require_param(name, clean_params, "staff") + return _staff_template(db_path, name, start, end, clean_params) + if name in {"delivery_plan_count", "delivery_plan_quantity"}: + return _delivery_scalar_template(db_path, name, start, end, clean_params) + if name == "delivery_plan_list": + return _delivery_list_template(db_path, name, start, end, clean_params, limit) + raise RuntimeError(f"unimplemented business template: {name}") + + +def _template_date_range(template_name: str, params: dict[str, Any]) -> tuple[str, str]: + if params.get("start_date") and params.get("end_date"): + return str(params["start_date"]), str(params["end_date"]) + if params.get("start_date") and params.get("end_date_exclusive"): + return str(params["start_date"]), str(params["end_date_exclusive"]) + if params.get("year") and params.get("month"): + return month_bounds(int(params["year"]), int(params["month"])) + raise RuntimeError(f"{template_name} requires year/month or start_date and end_date") + + +def _safe_limit(value: Any) -> int: + return max(1, min(int(value or 10), SQL_MAX_LIMIT)) + + +def _require_param(template_name: str, params: dict[str, Any], key: str) -> None: + if not params.get(key): + raise RuntimeError(f"{template_name} requires {key}") + + +def _template_response( + db_path: Path, + name: str, + params: dict[str, Any], + data: dict[str, Any], + *, + rows: list[dict[str, Any]] | None = None, + now: str | None = None, +) -> dict[str, Any]: + last_sync = last_successful_sync(db_path) + is_stale = sync_is_stale(last_sync, now=now) + result = { + "template": name, + "params": params, + "data": data, + "last_sync": last_sync, + "is_stale": is_stale, + "answer": _answer( + f"Result: {name} returned {data}.", + "Scope: structured business facts from the Kingdee sales-order database.", + last_sync, + is_stale=is_stale, + ), + } + if rows is not None: + result["rows"] = rows + result["row_count"] = len(rows) + return result + + +def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]: + return {key: row[key] for key in row.keys()} + + +def _where_date(column: str, start: str, end: str) -> tuple[list[str], list[Any]]: + return [f"{column} >= ?", f"{column} < ?"], [start, end] + + +def _order_filter_sql(params: dict[str, Any], alias: str = "o") -> tuple[list[str], list[Any]]: + clauses: list[str] = [] + args: list[Any] = [] + like_fields = { + "customer": (f"{alias}.customer_name", f"{alias}.customer_id"), + "org": (f"{alias}.org_name", f"{alias}.org_number"), + "department": (f"{alias}.dept_name", f"{alias}.dept_number"), + } + for param, fields in like_fields.items(): + if params.get(param): + clauses.append("(" + " OR ".join(f"{field} LIKE ?" for field in fields) + ")") + args.extend([f"%{params[param]}%"] * len(fields)) + exact_fields = { + "currency": f"{alias}.currency", + "billstatus": f"{alias}.billstatus", + "orderstatus": f"{alias}.orderstatus", + "closestatus": f"{alias}.closestatus", + } + for param, field in exact_fields.items(): + if params.get(param): + clauses.append(f"{field} = ?") + args.append(params[param]) + return clauses, args + + +def _item_filter_sql(params: dict[str, Any], alias: str = "i") -> tuple[list[str], list[Any]]: + clauses: list[str] = [] + args: list[Any] = [] + like_fields = { + "product": (f"{alias}.product_id", f"{alias}.product_name", f"{alias}.material_model"), + "project": (f"{alias}.project_number", f"{alias}.project_name"), + "warehouse": (f"{alias}.warehouse_number", f"{alias}.warehouse_name"), + } + for param, fields in like_fields.items(): + if params.get(param): + clauses.append("(" + " OR ".join(f"{field} LIKE ?" for field in fields) + ")") + args.extend([f"%{params[param]}%"] * len(fields)) + return clauses, args + + +def _order_where(params: dict[str, Any], start: str, end: str, alias: str = "o") -> tuple[str, list[Any]]: + clauses, args = _where_date(f"{alias}.order_date", start, end) + extra, extra_args = _order_filter_sql(params, alias=alias) + clauses.extend(extra) + args.extend(extra_args) + return " AND ".join(clauses), args + + +def _item_where(params: dict[str, Any], start: str, end: str) -> tuple[str, list[Any]]: + clauses, args = _where_date("o.order_date", start, end) + extra, extra_args = _order_filter_sql(params) + clauses.extend(extra) + args.extend(extra_args) + item_extra, item_args = _item_filter_sql(params) + clauses.extend(item_extra) + args.extend(item_args) + return " AND ".join(clauses), args + + +def _selected_staff_roles(params: dict[str, Any]) -> tuple[str, ...]: + return ALL_STAFF_ROLES if params.get("staff_role_scope") == "all" else PRIMARY_STAFF_ROLES + + +def _period_expr(column: str, granularity: str) -> str: + return column if granularity == "day" else f"substr({column}, 1, 7)" + + +def _staff_filter_sql(params: dict[str, Any], alias: str = "s") -> tuple[list[str], list[Any]]: + clauses: list[str] = [] + args: list[Any] = [] + roles = _selected_staff_roles(params) + clauses.append("(" + " OR ".join(f"{alias}.role = ?" for _role in roles) + ")") + args.extend(roles) + if params.get("staff"): + clauses.append(f"({alias}.staff_name LIKE ? OR {alias}.staff_number LIKE ?)") + args.extend([f"%{params['staff']}%", f"%{params['staff']}%"]) + return clauses, args + + +def _order_where_with_staff(params: dict[str, Any], start: str, end: str) -> tuple[str, list[Any], str]: + where, args = _order_where(params, start, end) + staff_join = "" + if params.get("staff"): + staff_join = "JOIN order_staff s ON s.order_id = o.order_id" + staff_clauses, staff_args = _staff_filter_sql(params) + where = f"{where} AND " + " AND ".join(staff_clauses) + args.extend(staff_args) + return where, args, staff_join + + +def _item_where_with_staff(params: dict[str, Any], start: str, end: str) -> tuple[str, list[Any], str]: + where, args = _item_where(params, start, end) + staff_join = "" + if params.get("staff"): + staff_join = "JOIN order_staff s ON s.order_id = o.order_id" + staff_clauses, staff_args = _staff_filter_sql(params) + where = f"{where} AND " + " AND ".join(staff_clauses) + args.extend(staff_args) + return where, args, staff_join + + +def _order_scalar_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any]) -> dict[str, Any]: + metric_sql = { + "order_count": "COUNT(DISTINCT order_no)", + "sales_amount": "COALESCE(SUM(total_amount), 0)", + "customer_count": "COUNT(DISTINCT customer_key)", + "avg_order_amount": "COALESCE(AVG(total_amount), 0)", + }[name] + data_key = { + "order_count": "order_count", + "sales_amount": "sales_amount", + "customer_count": "customer_count", + "avg_order_amount": "avg_order_amount", + }[name] + where, args, staff_join = _order_where_with_staff(params, start, end) + with connect(db_path) as conn: + row = conn.execute( + f""" + WITH matched_orders AS ( + SELECT DISTINCT o.order_id, o.order_no, o.total_amount, + COALESCE(NULLIF(o.customer_id, ''), NULLIF(o.customer_name, '')) AS customer_key + FROM orders o + {staff_join} + WHERE {where} + ) + SELECT {metric_sql} AS value + FROM matched_orders + """, + args, + ).fetchone() + value = float(row["value"] or 0) if name in {"sales_amount", "avg_order_amount"} else int(row["value"] or 0) + data = {"start_date": start, "end_date_exclusive": end, data_key: value} + return _template_response(db_path, name, params, data) + + +def _item_scalar_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any]) -> dict[str, Any]: + metric_sql = { + "product_count": "COUNT(DISTINCT product_key)", + "quantity_sum": "COALESCE(SUM(quantity), 0)", + }[name] + data_key = {"product_count": "product_count", "quantity_sum": "quantity_sum"}[name] + where, args, staff_join = _item_where_with_staff(params, start, end) + with connect(db_path) as conn: + row = conn.execute( + f""" + WITH matched_items AS ( + SELECT DISTINCT i.order_id, i.line_id, i.quantity, + COALESCE(NULLIF(i.product_id, ''), NULLIF(i.product_name, '')) AS product_key + FROM orders o + JOIN order_items i ON i.order_id = o.order_id + {staff_join} + WHERE {where} + ) + SELECT {metric_sql} AS value + FROM matched_items + """, + args, + ).fetchone() + value = float(row["value"] or 0) if name == "quantity_sum" else int(row["value"] or 0) + data = {"start_date": start, "end_date_exclusive": end, data_key: value} + return _template_response(db_path, name, params, data) + + +def _order_trend_template( + db_path: Path, + name: str, + start: str, + end: str, + params: dict[str, Any], + granularity: str, +) -> dict[str, Any]: + metric_sql = { + "order_trend": "COUNT(DISTINCT order_no)", + "sales_trend": "COALESCE(SUM(total_amount), 0)", + }[name] + value_key = {"order_trend": "order_count", "sales_trend": "sales_amount"}[name] + where, args, staff_join = _order_where_with_staff(params, start, end) + period = _period_expr("o.order_date", granularity) + with connect(db_path) as conn: + rows = conn.execute( + f""" + WITH matched_orders AS ( + SELECT DISTINCT o.order_id, o.order_no, o.total_amount, {period} AS period + FROM orders o + {staff_join} + WHERE {where} + ) + SELECT period, {metric_sql} AS value + FROM matched_orders + GROUP BY period + ORDER BY period ASC + """, + args, + ).fetchall() + result_rows = [ + {"period": row["period"], value_key: float(row["value"] or 0) if value_key == "sales_amount" else int(row["value"] or 0)} + for row in rows + ] + data = {"start_date": start, "end_date_exclusive": end, "granularity": granularity, "rows": result_rows} + return _template_response(db_path, name, params, data, rows=result_rows) + + +def _item_trend_template( + db_path: Path, + name: str, + start: str, + end: str, + params: dict[str, Any], + granularity: str, +) -> dict[str, Any]: + where, args, staff_join = _item_where_with_staff(params, start, end) + period = _period_expr("o.order_date", granularity) + with connect(db_path) as conn: + rows = conn.execute( + f""" + WITH matched_items AS ( + SELECT DISTINCT i.order_id, i.line_id, i.quantity, {period} AS period + FROM orders o + JOIN order_items i ON i.order_id = o.order_id + {staff_join} + WHERE {where} + ) + SELECT period, COALESCE(SUM(quantity), 0) AS quantity_sum + FROM matched_items + GROUP BY period + ORDER BY period ASC + """, + args, + ).fetchall() + result_rows = [{"period": row["period"], "quantity_sum": float(row["quantity_sum"] or 0)} for row in rows] + data = {"start_date": start, "end_date_exclusive": end, "granularity": granularity, "rows": result_rows} + return _template_response(db_path, name, params, data, rows=result_rows) + + +def _order_group_template( + db_path: Path, + name: str, + start: str, + end: str, + params: dict[str, Any], + group: str, + limit: int, +) -> dict[str, Any]: + group_specs = { + "customer": ("COALESCE(NULLIF(o.customer_name, ''), NULLIF(o.customer_id, ''), 'Unknown')", "customer"), + "org": ("COALESCE(NULLIF(o.org_name, ''), NULLIF(o.org_number, ''), 'Unknown')", "org"), + "department": ("COALESCE(NULLIF(o.dept_name, ''), NULLIF(o.dept_number, ''), 'Unknown')", "department"), + "currency": ("COALESCE(NULLIF(o.currency, ''), 'Unknown')", "currency"), + "status": ("COALESCE(NULLIF(o.status, ''), 'Unknown')", "status"), + "orderstatus": ("COALESCE(NULLIF(o.orderstatus, ''), 'Unknown')", "status"), + "billstatus": ("COALESCE(NULLIF(o.billstatus, ''), 'Unknown')", "status"), + "closestatus": ("COALESCE(NULLIF(o.closestatus, ''), 'Unknown')", "status"), + } + group_expr, output_key = group_specs[group] + where, args, staff_join = _order_where_with_staff(params, start, end) + status_group = group in {"status", "orderstatus", "billstatus", "closestatus"} + order_sql = "label ASC" if status_group else "sales_amount DESC, order_count DESC, label ASC" + with connect(db_path) as conn: + rows = conn.execute( + f""" + WITH matched_orders AS ( + SELECT DISTINCT o.order_id, o.order_no, o.total_amount, {group_expr} AS label + FROM orders o + {staff_join} + WHERE {where} + ) + SELECT label, COUNT(DISTINCT order_no) AS order_count, + COALESCE(SUM(total_amount), 0) AS sales_amount + FROM matched_orders + GROUP BY label + ORDER BY {order_sql} + LIMIT ? + """, + (*args, limit), + ).fetchall() + result_rows = [ + { + output_key: row["label"], + "order_count": int(row["order_count"] or 0), + "sales_amount": float(row["sales_amount"] or 0), + } + for row in rows + ] + data = {"start_date": start, "end_date_exclusive": end, "group_by": group, "rows": result_rows} + return _template_response(db_path, name, params, data, rows=result_rows) + + +def _item_group_template( + db_path: Path, + name: str, + start: str, + end: str, + params: dict[str, Any], + group: str, + limit: int, +) -> dict[str, Any]: + group_specs = { + "product": ( + "COALESCE(NULLIF(i.product_id, ''), 'Unknown')", + "COALESCE(NULLIF(i.product_name, ''), '')", + "product_id", + "product_name", + ), + "project": ( + "COALESCE(NULLIF(i.project_number, ''), NULLIF(i.project_name, ''), 'Unknown')", + "COALESCE(NULLIF(i.project_name, ''), '')", + "project", + "project_name", + ), + "warehouse": ( + "COALESCE(NULLIF(i.warehouse_number, ''), NULLIF(i.warehouse_name, ''), 'Unknown')", + "COALESCE(NULLIF(i.warehouse_name, ''), '')", + "warehouse", + "warehouse_name", + ), + } + label_expr, name_expr, output_key, name_key = group_specs[group] + where, args, staff_join = _item_where_with_staff(params, start, end) + with connect(db_path) as conn: + rows = conn.execute( + f""" + WITH matched_items AS ( + SELECT DISTINCT i.order_id, i.line_id, o.order_no, i.quantity, + COALESCE(i.amount_and_tax, i.amount, 0) AS amount, + {label_expr} AS label, {name_expr} AS label_name + FROM orders o + JOIN order_items i ON i.order_id = o.order_id + {staff_join} + WHERE {where} + ) + SELECT label, label_name, COUNT(DISTINCT order_no) AS order_count, + COALESCE(SUM(quantity), 0) AS quantity, + COALESCE(SUM(amount), 0) AS amount + FROM matched_items + GROUP BY label, label_name + ORDER BY quantity DESC, amount DESC, order_count DESC, label ASC + LIMIT ? + """, + (*args, limit), + ).fetchall() + result_rows = [ + { + output_key: row["label"], + name_key: row["label_name"], + "order_count": int(row["order_count"] or 0), + "quantity": float(row["quantity"] or 0), + "amount": float(row["amount"] or 0), + } + for row in rows + ] + data = {"start_date": start, "end_date_exclusive": end, "group_by": group, "rows": result_rows} + return _template_response(db_path, name, params, data, rows=result_rows) + + +def _staff_group_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any], limit: int) -> dict[str, Any]: + where, args = _order_where(params, start, end) + staff_clauses, staff_args = _staff_filter_sql(params) + where = f"{where} AND " + " AND ".join(staff_clauses) + args.extend(staff_args) + with connect(db_path) as conn: + rows = conn.execute( + f""" + WITH matched_staff AS ( + SELECT DISTINCT COALESCE(NULLIF(s.staff_name, ''), NULLIF(s.staff_number, ''), 'Unknown') AS staff, + s.staff_number, o.order_id, o.order_no, o.total_amount + FROM orders o + JOIN order_staff s ON s.order_id = o.order_id + WHERE {where} + ) + SELECT staff, staff_number, COUNT(DISTINCT order_no) AS order_count, + COALESCE(SUM(total_amount), 0) AS sales_amount + FROM matched_staff + GROUP BY staff, staff_number + ORDER BY sales_amount DESC, order_count DESC, staff ASC + LIMIT ? + """, + (*args, limit), + ).fetchall() + result_rows = [ + { + "staff": row["staff"], + "staff_number": row["staff_number"], + "order_count": int(row["order_count"] or 0), + "sales_amount": float(row["sales_amount"] or 0), + } + for row in rows + ] + data = { + "start_date": start, + "end_date_exclusive": end, + "staff_roles": list(_selected_staff_roles(params)), + "rows": result_rows, + } + return _template_response(db_path, name, params, data, rows=result_rows) + + +def _order_list_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any], limit: int) -> dict[str, Any]: + where, args, staff_join = _order_where_with_staff(params, start, end) + with connect(db_path) as conn: + rows = conn.execute( + f""" + SELECT DISTINCT o.order_id, o.order_no, o.order_date, o.customer_id, o.customer_name, + o.total_amount, o.currency, o.billstatus, o.orderstatus, o.closestatus, + o.org_name, o.dept_name, o.operator_name, o.updated_at + FROM orders o + {staff_join} + WHERE {where} + ORDER BY o.order_date DESC, o.order_no ASC + LIMIT ? + """, + (*args, limit), + ).fetchall() + result_rows = [_row_to_dict(row) for row in rows] + for row in result_rows: + row["total_amount"] = float(row["total_amount"] or 0) + data = {"start_date": start, "end_date_exclusive": end, "orders": result_rows} + return _template_response(db_path, name, params, data, rows=result_rows) + + +def _product_order_list_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any], limit: int) -> dict[str, Any]: + where, args, staff_join = _item_where_with_staff(params, start, end) + with connect(db_path) as conn: + rows = conn.execute( + f""" + SELECT DISTINCT o.order_id, o.order_no, o.order_date, o.customer_name, + o.total_amount, i.line_id, i.product_id, i.product_name, i.material_model, + i.quantity, COALESCE(i.amount_and_tax, i.amount, 0) AS amount, + i.project_name, i.warehouse_name + FROM orders o + JOIN order_items i ON i.order_id = o.order_id + {staff_join} + WHERE {where} + ORDER BY o.order_date DESC, o.order_no ASC, i.seq ASC + LIMIT ? + """, + (*args, limit), + ).fetchall() + result_rows = [_row_to_dict(row) for row in rows] + for row in result_rows: + row["total_amount"] = float(row["total_amount"] or 0) + row["quantity"] = float(row["quantity"] or 0) + row["amount"] = float(row["amount"] or 0) + data = {"start_date": start, "end_date_exclusive": end, "orders": result_rows} + return _template_response(db_path, name, params, data, rows=result_rows) + + +def _customer_summary_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any]) -> dict[str, Any]: + where, args, staff_join = _order_where_with_staff(params, start, end) + with connect(db_path) as conn: + row = conn.execute( + f""" + WITH matched_orders AS ( + SELECT DISTINCT o.order_id, o.order_no, o.total_amount + FROM orders o + {staff_join} + WHERE {where} + ) + SELECT COUNT(DISTINCT order_no) AS order_count, + COALESCE(SUM(total_amount), 0) AS sales_amount, + COALESCE(AVG(total_amount), 0) AS avg_order_amount + FROM matched_orders + """, + args, + ).fetchone() + data = { + "customer": params.get("customer"), + "start_date": start, + "end_date_exclusive": end, + "order_count": int(row["order_count"] or 0), + "sales_amount": float(row["sales_amount"] or 0), + "avg_order_amount": float(row["avg_order_amount"] or 0), + } + return _template_response(db_path, name, params, data) + + +def _product_summary_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any]) -> dict[str, Any]: + where, args, staff_join = _item_where_with_staff(params, start, end) + with connect(db_path) as conn: + row = conn.execute( + f""" + WITH matched_items AS ( + SELECT DISTINCT o.order_no, o.customer_id, o.customer_name, i.order_id, i.line_id, + i.quantity, COALESCE(i.amount_and_tax, i.amount, 0) AS amount + FROM orders o + JOIN order_items i ON i.order_id = o.order_id + {staff_join} + WHERE {where} + ) + SELECT COUNT(DISTINCT order_no) AS order_count, + COUNT(DISTINCT COALESCE(NULLIF(customer_id, ''), NULLIF(customer_name, ''))) AS customer_count, + COALESCE(SUM(quantity), 0) AS quantity, + COALESCE(SUM(amount), 0) AS amount + FROM matched_items + """, + args, + ).fetchone() + data = { + "product": params.get("product"), + "start_date": start, + "end_date_exclusive": end, + "order_count": int(row["order_count"] or 0), + "customer_count": int(row["customer_count"] or 0), + "quantity": float(row["quantity"] or 0), + "amount": float(row["amount"] or 0), + } + return _template_response(db_path, name, params, data) + + +def _staff_template( + db_path: Path, + name: str, + start: str, + end: str, + params: dict[str, Any], + *, + list_mode: bool = False, + limit: int = 10, +) -> dict[str, Any]: + where, args = _order_where(params, start, end) + staff_clauses, staff_args = _staff_filter_sql(params) + where = f"{where} AND " + " AND ".join(staff_clauses) + args.extend(staff_args) + roles = _selected_staff_roles(params) + if list_mode: + with connect(db_path) as conn: + rows = conn.execute( + f""" + SELECT DISTINCT o.order_id, o.order_no, o.order_date, o.customer_name, + o.total_amount, o.currency, o.billstatus, o.orderstatus, + o.org_name, o.dept_name + FROM orders o + JOIN order_staff s ON s.order_id = o.order_id + WHERE {where} + ORDER BY o.order_date DESC, o.order_no ASC + LIMIT ? + """, + (*args, limit), + ).fetchall() + result_rows = [_row_to_dict(row) for row in rows] + for row in result_rows: + row["total_amount"] = float(row["total_amount"] or 0) + data = { + "staff": params.get("staff"), + "staff_roles": list(roles), + "start_date": start, + "end_date_exclusive": end, + "orders": result_rows, + } + return _template_response(db_path, name, params, data, rows=result_rows) + + with connect(db_path) as conn: + row = conn.execute( + f""" + WITH matched_orders AS ( + SELECT DISTINCT o.order_id, o.order_no, o.customer_id, o.customer_name, o.total_amount + FROM orders o + JOIN order_staff s ON s.order_id = o.order_id + WHERE {where} + ) + SELECT COUNT(DISTINCT order_no) AS order_count, + COUNT(DISTINCT COALESCE(NULLIF(customer_id, ''), NULLIF(customer_name, ''))) AS customer_count, + COALESCE(SUM(total_amount), 0) AS sales_amount, + COALESCE(AVG(total_amount), 0) AS avg_order_amount + FROM matched_orders + """, + args, + ).fetchone() + data = { + "staff": params.get("staff"), + "staff_roles": list(roles), + "start_date": start, + "end_date_exclusive": end, + "order_count": int(row["order_count"] or 0), + "customer_count": int(row["customer_count"] or 0), + "sales_amount": float(row["sales_amount"] or 0), + "avg_order_amount": float(row["avg_order_amount"] or 0), + } + return _template_response(db_path, name, params, data) + + +def _delivery_where(params: dict[str, Any], start: str, end: str) -> tuple[str, list[Any], str]: + clauses, args = _where_date("d.plan_date", start, end) + extra, extra_args = _order_filter_sql(params) + clauses.extend(extra) + args.extend(extra_args) + item_extra, item_args = _item_filter_sql(params) + clauses.extend(item_extra) + args.extend(item_args) + staff_join = "" + if params.get("staff"): + staff_join = "JOIN order_staff s ON s.order_id = o.order_id" + staff_clauses, staff_args = _staff_filter_sql(params) + clauses.extend(staff_clauses) + args.extend(staff_args) + return " AND ".join(clauses), args, staff_join + + +def _delivery_scalar_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any]) -> dict[str, Any]: + metric_sql = { + "delivery_plan_count": "COUNT(*)", + "delivery_plan_quantity": "COALESCE(SUM(plan_qty), 0)", + }[name] + data_key = { + "delivery_plan_count": "delivery_plan_count", + "delivery_plan_quantity": "delivery_plan_quantity", + }[name] + where, args, staff_join = _delivery_where(params, start, end) + with connect(db_path) as conn: + row = conn.execute( + f""" + WITH matched_deliveries AS ( + SELECT DISTINCT d.order_id, d.line_id, d.delivery_id, d.plan_qty + FROM order_deliveries d + JOIN orders o ON o.order_id = d.order_id + JOIN order_items i ON i.order_id = d.order_id AND i.line_id = d.line_id + {staff_join} + WHERE {where} + ) + SELECT {metric_sql} AS value + FROM matched_deliveries + """, + args, + ).fetchone() + value = float(row["value"] or 0) if name == "delivery_plan_quantity" else int(row["value"] or 0) + data = {"start_date": start, "end_date_exclusive": end, data_key: value} + return _template_response(db_path, name, params, data) + + +def _delivery_list_template(db_path: Path, name: str, start: str, end: str, params: dict[str, Any], limit: int) -> dict[str, Any]: + where, args, staff_join = _delivery_where(params, start, end) + with connect(db_path) as conn: + rows = conn.execute( + f""" + SELECT DISTINCT d.order_id, o.order_no, o.order_date, o.customer_name, + i.line_id, i.product_id, i.product_name, d.delivery_id, d.plan_date, + d.plan_delivery_date, d.plan_qty, d.plan_base_qty, d.plan_unit_name, + d.receive_place_name, d.remark + FROM order_deliveries d + JOIN orders o ON o.order_id = d.order_id + JOIN order_items i ON i.order_id = d.order_id AND i.line_id = d.line_id + {staff_join} + WHERE {where} + ORDER BY d.plan_date ASC, o.order_no ASC, i.seq ASC + LIMIT ? + """, + (*args, limit), + ).fetchall() + result_rows = [_row_to_dict(row) for row in rows] + for row in result_rows: + row["plan_qty"] = float(row["plan_qty"] or 0) + row["plan_base_qty"] = float(row["plan_base_qty"] or 0) + data = {"start_date": start, "end_date_exclusive": end, "deliveries": result_rows} + return _template_response(db_path, name, params, data, rows=result_rows) + + def sync_kingdee_orders( vault_path: str | Path, records: list[dict[str, Any]], diff --git a/obsidian_wiki/cli.py b/obsidian_wiki/cli.py index c7ac2bb..290cb6f 100644 --- a/obsidian_wiki/cli.py +++ b/obsidian_wiki/cli.py @@ -19,9 +19,11 @@ 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_schema, business_db_path_for_vault, compute_incremental_start_modifytime, execute_business_sql, + execute_business_template, fetch_kingdee_payload, load_kingdee_payload_from_json, query_business_question, @@ -609,6 +611,64 @@ def cmd_business_query(args: argparse.Namespace) -> int: return 0 +def cmd_business_schema(args: argparse.Namespace) -> int: + schema = business_schema() + if args.json: + print(json.dumps(schema, ensure_ascii=False, indent=2)) + else: + print("Business facts schema") + print("Tables:") + for table in schema["tables"]: + print(f" - {table['name']}: {', '.join(table['fields'])}") + print("Templates:") + for template in schema["templates"]: + required = ", ".join(template["required_parameters"]) or "none" + print(f" - {template['name']}: {template['description']} (required: {required})") + return 0 + + +def cmd_business_template(args: argparse.Namespace) -> int: + vault_path = _require_vault(args) + params = _business_template_params(args) + result = execute_business_template( + _resolve_business_db(args, vault_path), + args.template_name, + **params, + ) + if args.json: + print(json.dumps(result, ensure_ascii=False, indent=2)) + else: + print(result["answer"]) + for row in result.get("rows", []): + print(json.dumps(row, ensure_ascii=False, sort_keys=True)) + return 0 + + +def _business_template_params(args: argparse.Namespace) -> dict[str, object]: + keys = ( + "year", + "month", + "start_date", + "end_date", + "end_date_exclusive", + "limit", + "granularity", + "staff_role_scope", + "customer", + "product", + "staff", + "org", + "department", + "project", + "warehouse", + "billstatus", + "orderstatus", + "closestatus", + "currency", + ) + return {key: getattr(args, key) for key in keys if getattr(args, key, None) not in (None, "")} + + def cmd_business_sql(args: argparse.Namespace) -> int: vault_path = _require_vault(args) sql = " ".join(args.sql).strip() @@ -803,6 +863,41 @@ def build_parser() -> argparse.ArgumentParser: bqp.add_argument("question", nargs="+", help="question, e.g. '2026-06 order count'") bqp.set_defaults(func=cmd_business_query) + bsch = business_sub.add_parser("schema", help="print business fact tables, templates, and SQL guardrails") + bsch.add_argument("--json", action="store_true", help="print machine-readable schema") + bsch.set_defaults(func=cmd_business_schema) + + btp = business_sub.add_parser("template", help="run one business query template with explicit parameters") + _add_vault_arg(btp) + btp.add_argument("--db", metavar="PATH", help="SQLite DB path (defaults to /_meta/business-facts.sqlite)") + btp.add_argument("--year", type=int, help="calendar year, e.g. 2026") + btp.add_argument("--month", type=int, help="calendar month, 1-12") + btp.add_argument("--start-date", help="inclusive date, YYYY-MM-DD") + btp.add_argument("--end-date", help="exclusive date, YYYY-MM-DD") + btp.add_argument("--end-date-exclusive", help="explicit exclusive end date, YYYY-MM-DD") + btp.add_argument("--limit", type=int, default=10, help="maximum rows for list/ranking templates") + btp.add_argument("--granularity", choices=("day", "month"), default="month", help="trend grouping") + btp.add_argument( + "--staff-role-scope", + choices=("primary", "all"), + default="primary", + help="primary sales only by default; all includes secondary sales and follow-up roles", + ) + btp.add_argument("--customer", help="customer name or code filter") + btp.add_argument("--product", help="product/material name, code, or model filter") + btp.add_argument("--staff", help="staff name or code filter") + btp.add_argument("--org", help="sales organization name or code filter") + btp.add_argument("--department", help="sales department name or code filter") + btp.add_argument("--project", help="project name or code filter") + btp.add_argument("--warehouse", help="warehouse name or code filter") + btp.add_argument("--billstatus", help="Kingdee billstatus filter") + btp.add_argument("--orderstatus", help="Kingdee orderstatus filter") + btp.add_argument("--closestatus", help="Kingdee closestatus filter") + btp.add_argument("--currency", help="currency code filter") + btp.add_argument("--json", action="store_true", help="print machine-readable template result") + btp.add_argument("template_name", help="template name from `business schema`") + btp.set_defaults(func=cmd_business_template) + bsp_sql = business_sub.add_parser("sql", help="run guarded read-only SQL against structured business facts") _add_vault_arg(bsp_sql) bsp_sql.add_argument("--db", metavar="PATH", help="SQLite DB path (defaults to /_meta/business-facts.sqlite)") diff --git a/scripts/obsidian-wiki-ops.py b/scripts/obsidian-wiki-ops.py index 80a700b..2b210b1 100755 --- a/scripts/obsidian-wiki-ops.py +++ b/scripts/obsidian-wiki-ops.py @@ -34,7 +34,8 @@ Usage: scripts/obsidian-wiki-ops.py kingdee config [args...] scripts/obsidian-wiki-ops.py kingdee auth-test [args...] scripts/obsidian-wiki-ops.py kingdee sync [args...] - scripts/obsidian-wiki-ops.py kingdee query "2026年6月接了多少单" [args...] + scripts/obsidian-wiki-ops.py kingdee schema --json + scripts/obsidian-wiki-ops.py kingdee template order_count --vault /path/to/vault --year 2026 --month 6 --json scripts/obsidian-wiki-ops.py kingdee sql "SELECT COUNT(*) FROM orders" [args...] scripts/obsidian-wiki-ops.py timer install [args...] @@ -89,6 +90,10 @@ def route_kingdee(args: Sequence[str]) -> int: return obsidian_cli_main(["business", "sync", "kingdee", "--dry-run-auth", *rest]) if action == "sync": return obsidian_cli_main(["business", "sync", "kingdee", *rest]) + if action == "schema": + return obsidian_cli_main(["business", "schema", *rest]) + if action == "template": + return obsidian_cli_main(["business", "template", *rest]) if action == "query": return obsidian_cli_main(["business", "query", *rest]) if action == "sql": diff --git a/tests/test_business_facts.py b/tests/test_business_facts.py index 19dd134..8fb32e7 100644 --- a/tests/test_business_facts.py +++ b/tests/test_business_facts.py @@ -8,9 +8,11 @@ from pathlib import Path from unittest.mock import patch from obsidian_wiki.business_facts import ( + business_schema, business_db_path_for_vault, compute_incremental_start_modifytime, execute_business_sql, + execute_business_template, query_business_question, sync_kingdee_orders, ) @@ -461,6 +463,231 @@ class BusinessFactsTests(unittest.TestCase): self.assertEqual(fallback["template"], "sql_fallback") self.assertEqual(fallback["rows"], [{"order_no": "SO-001", "customer_name": "Customer A"}]) + def test_business_schema_describes_templates_fields_and_sql_guardrails(self): + schema = business_schema() + + template_names = {template["name"] for template in schema["templates"]} + templates = {template["name"]: template for template in schema["templates"]} + self.assertTrue( + { + "order_count", + "sales_amount", + "customer_count", + "product_count", + "quantity_sum", + "avg_order_amount", + "order_trend", + "sales_trend", + "quantity_trend", + "top_customers", + "top_products", + "top_staff", + "top_orgs", + "top_departments", + "top_projects", + "top_warehouses", + "breakdown_by_customer", + "breakdown_by_product", + "breakdown_by_staff", + "breakdown_by_org", + "breakdown_by_department", + "breakdown_by_status", + "breakdown_by_currency", + "order_list", + "customer_order_list", + "product_order_list", + "staff_order_list", + "customer_summary", + "product_summary", + "staff_summary", + "delivery_plan_count", + "delivery_plan_quantity", + "delivery_plan_list", + "order_status_breakdown", + "billstatus_breakdown", + "close_status_breakdown", + }.issubset(template_names) + ) + table_names = {table["name"] for table in schema["tables"]} + self.assertIn("orders", table_names) + self.assertIn("order_items", table_names) + self.assertIn("order_staff", table_names) + self.assertIn("order_deliveries", table_names) + self.assertIn("COUNT(DISTINCT order_no)", schema["metric_guidance"]["order_count"]) + self.assertEqual(templates["customer_count"]["required_parameters"], []) + self.assertEqual(templates["product_count"]["required_parameters"], []) + self.assertEqual(templates["customer_summary"]["required_parameters"], ["customer"]) + self.assertEqual(templates["product_summary"]["required_parameters"], ["product"]) + self.assertEqual(templates["staff_summary"]["required_parameters"], ["staff"]) + self.assertIn("raw_json", schema["sql_guardrails"]["forbidden_columns"]) + self.assertIn("raw_snapshots", schema["sql_guardrails"]["forbidden_tables"]) + + def test_execute_business_templates_cover_core_totals_rankings_lists_and_delivery_plans(self): + sync_kingdee_orders( + self.vault, + [ + { + "id": "1001", + "billno": "SO-001", + "bizdate": "2026-06-03", + "customer_number": "C01", + "customer_name": "Customer A", + "totalallamount": "100.00", + "settlecurrency_number": "CNY", + "billstatus": "C", + "orderstatus": "D", + "closestatus": "A", + "org_name": "Org A", + "dept_name": "Dept A", + "operator_operatorname": "徐丽", + "billentry": [ + { + "id": "L1", + "material_masterid_number": "P01", + "material_masterid_name": "Part 1", + "qty": "2", + "amountandtax": "60", + "project_name": "Project A", + "warehouse_name": "Warehouse A", + "orderdeliverentry": [ + { + "id": "D1", + "d_plandate": "2026-06-10", + "d_plandeliverydate": "2026-06-08", + "d_planqty": "2", + } + ], + }, + { + "id": "L2", + "material_masterid_number": "P02", + "material_masterid_name": "Part 2", + "qty": "1", + "amountandtax": "40", + "project_name": "Project B", + "warehouse_name": "Warehouse B", + }, + ], + }, + { + "id": "1002", + "billno": "SO-002", + "bizdate": "2026-06-04", + "customer_number": "C02", + "customer_name": "Customer B", + "totalallamount": "300.00", + "settlecurrency_number": "CNY", + "billstatus": "B", + "orderstatus": "C", + "closestatus": "A", + "org_name": "Org B", + "dept_name": "Dept B", + "operator_operatorname": "王五", + "h5a3_basedatafield2_operatorname": "徐丽", + "billentry": [ + { + "id": "L1", + "material_masterid_number": "P01", + "material_masterid_name": "Part 1", + "qty": "3", + "amountandtax": "300", + "project_name": "Project A", + "warehouse_name": "Warehouse A", + "orderdeliverentry": [ + { + "id": "D2", + "d_plandate": "2026-06-11", + "d_planqty": "3", + } + ], + } + ], + }, + ], + synced_at="2026-07-06T09:00:00Z", + ) + + count = execute_business_template(self.db_path, "order_count", start_date="2026-06-01", end_date="2026-07-01") + sales = execute_business_template(self.db_path, "sales_amount", year=2026, month=6) + customers = execute_business_template(self.db_path, "customer_count", year=2026, month=6) + products = execute_business_template(self.db_path, "product_count", year=2026, month=6) + quantity = execute_business_template(self.db_path, "quantity_sum", year=2026, month=6) + average = execute_business_template(self.db_path, "avg_order_amount", year=2026, month=6) + top_products = execute_business_template(self.db_path, "top_products", year=2026, month=6, limit=1) + staff = execute_business_template( + self.db_path, + "staff_summary", + year=2026, + month=6, + staff="徐丽", + staff_role_scope="all", + ) + delivery_qty = execute_business_template(self.db_path, "delivery_plan_quantity", year=2026, month=6) + status = execute_business_template(self.db_path, "billstatus_breakdown", year=2026, month=6) + + self.assertEqual(count["data"]["order_count"], 2) + self.assertEqual(sales["data"]["sales_amount"], 400.0) + self.assertEqual(customers["data"]["customer_count"], 2) + self.assertEqual(products["data"]["product_count"], 2) + self.assertEqual(quantity["data"]["quantity_sum"], 6.0) + self.assertEqual(average["data"]["avg_order_amount"], 200.0) + self.assertEqual(top_products["rows"][0]["product_id"], "P01") + self.assertEqual(top_products["rows"][0]["quantity"], 5.0) + self.assertEqual(staff["data"]["order_count"], 2) + self.assertEqual(staff["data"]["sales_amount"], 400.0) + self.assertEqual(delivery_qty["data"]["delivery_plan_quantity"], 5.0) + self.assertEqual( + [(row["status"], row["order_count"]) for row in status["rows"]], + [("B", 1), ("C", 1)], + ) + + def test_execute_business_template_rejects_unknown_or_missing_parameters(self): + with self.assertRaisesRegex(RuntimeError, "unknown business template"): + execute_business_template(self.db_path, "not_a_template", year=2026, month=6) + with self.assertRaisesRegex(RuntimeError, "requires"): + execute_business_template(self.db_path, "staff_summary", year=2026, month=6) + + def test_cli_business_schema_and_template_print_json(self): + sync_kingdee_orders( + self.vault, + [ + { + "id": "1001", + "billno": "SO-001", + "bizdate": "2026-06-03", + "customer_name": "Customer A", + "totalallamount": "100.00", + } + ], + synced_at="2026-07-06T09:00:00Z", + ) + from obsidian_wiki.cli import main + + schema_output = StringIO() + with redirect_stdout(schema_output): + schema_code = main(["business", "schema", "--json"]) + template_output = StringIO() + with redirect_stdout(template_output): + template_code = main( + [ + "business", + "template", + "order_count", + "--vault", + str(self.vault), + "--year", + "2026", + "--month", + "6", + "--json", + ] + ) + + self.assertEqual(schema_code, 0) + self.assertEqual(template_code, 0) + self.assertIn("order_count", {template["name"] for template in json.loads(schema_output.getvalue())["templates"]}) + self.assertEqual(json.loads(template_output.getvalue())["data"]["order_count"], 1) + def test_incremental_start_modifytime_uses_cursor_with_lookback_or_initial_days(self): initial = compute_incremental_start_modifytime( self.db_path,