feat(business): add schema-driven query templates

Expose a business schema catalog and explicit template runner for Kingdee sales-order facts. Update wiki-query guidance so the LLM routes operational questions through templates first and guarded SQL only when needed. Refresh user docs and ops aliases, and add coverage for schema metadata, template execution, and CLI JSON output.

Validation: python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests scripts; git diff --check
This commit is contained in:
Codex
2026-07-06 22:27:49 +00:00
parent b85c26e0ba
commit d3f3151218
8 changed files with 1327 additions and 39 deletions
+95
View File
@@ -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 <vault>/_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 <vault>/_meta/business-facts.sqlite)")