obsidian-wiki/obsidian_wiki/business_facts.py

"""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 re
import sqlite3
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

from obsidian_wiki.file_assets import clean_config_value
from obsidian_wiki.kingdee_api import fetch_kingdee_sales_orders, load_kingdee_config, token_cache_path_for_config

BUSINESS_DB_RELATIVE_PATH = Path("_meta") / "business-facts.sqlite"
KINGDEE_SYNC_STATE_SOURCE = "kingdee_sales_orders_modifytime"

ORDER_COLUMNS = (
    "order_id",
    "order_no",
    "order_date",
    "customer_id",
    "customer_name",
    "total_amount",
    "currency",
    "status",
    "updated_at",
    "billstatus",
    "orderstatus",
    "closestatus",
    "org_id",
    "org_number",
    "org_name",
    "dept_id",
    "dept_number",
    "dept_name",
    "operator_id",
    "operator_number",
    "operator_name",
    "total_net_amount",
    "total_tax_amount",
    "total_all_amount",
    "receive_address",
    "comment",
    "raw_json",
)

ORDER_ITEM_COLUMNS = (
    "order_id",
    "line_id",
    "product_id",
    "product_name",
    "quantity",
    "amount",
    "seq",
    "material_model",
    "unit_number",
    "unit_name",
    "price",
    "price_and_tax",
    "amount_and_tax",
    "tax_amount",
    "delivery_date",
    "project_number",
    "project_name",
    "warehouse_number",
    "warehouse_name",
    "raw_json",
)

ORDER_DELIVERY_COLUMNS = (
    "order_id",
    "line_id",
    "delivery_id",
    "plan_date",
    "plan_delivery_date",
    "plan_qty",
    "plan_base_qty",
    "plan_unit_number",
    "plan_unit_name",
    "receive_address",
    "receive_place_number",
    "receive_place_name",
    "remark",
    "transport_lead_time",
    "raw_json",
)

ORDER_STAFF_COLUMNS = (
    "order_id",
    "role",
    "staff_id",
    "staff_number",
    "staff_name",
)

STAFF_ROLE_FIELDS = (
    ("primary_sales", "operator_id", "operator_operatornumber", "operator_operatorname"),
    ("sales_2", "h5a3_basedatafield2_id", "h5a3_basedatafield2_operatornumber", "h5a3_basedatafield2_operatorname"),
    ("sales_3", "h5a3_basedatafield3_id", "h5a3_basedatafield3_operatornumber", "h5a3_basedatafield3_operatorname"),
    ("sales_4", "h5a3_basedatafield5_id", "h5a3_basedatafield5_operatornumber", "h5a3_basedatafield5_operatorname"),
    ("followup", "h5a3_basedatafield4_id", "h5a3_basedatafield4_operatornumber", "h5a3_basedatafield4_operatorname"),
)

ALL_STAFF_ROLES = tuple(role for role, _id_field, _number_field, _name_field in STAFF_ROLE_FIELDS)
PRIMARY_STAFF_ROLES = ("primary_sales",)

ALLOWED_SQL_TABLES = {"orders", "order_items", "order_deliveries", "order_staff", "customers", "products", "sync_runs"}
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")


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,
                billstatus TEXT,
                orderstatus TEXT,
                closestatus TEXT,
                org_id TEXT,
                org_number TEXT,
                org_name TEXT,
                dept_id TEXT,
                dept_number TEXT,
                dept_name TEXT,
                operator_id TEXT,
                operator_number TEXT,
                operator_name TEXT,
                total_net_amount REAL NOT NULL DEFAULT 0,
                total_tax_amount REAL NOT NULL DEFAULT 0,
                total_all_amount REAL NOT NULL DEFAULT 0,
                receive_address TEXT,
                comment TEXT,
                raw_json TEXT,
                raw_hash TEXT,
                created_at TEXT NOT NULL,
                synced_at TEXT NOT NULL
            );

            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,
                seq INTEGER,
                material_model TEXT,
                unit_number TEXT,
                unit_name TEXT,
                price REAL,
                price_and_tax REAL,
                amount_and_tax REAL,
                tax_amount REAL,
                delivery_date TEXT,
                project_number TEXT,
                project_name TEXT,
                warehouse_number TEXT,
                warehouse_name TEXT,
                raw_json TEXT,
                PRIMARY KEY (order_id, line_id),
                FOREIGN KEY (order_id) REFERENCES orders(order_id)
            );

            CREATE TABLE IF NOT EXISTS order_deliveries (
                order_id TEXT NOT NULL,
                line_id TEXT NOT NULL,
                delivery_id TEXT NOT NULL,
                plan_date TEXT,
                plan_delivery_date TEXT,
                plan_qty REAL,
                plan_base_qty REAL,
                plan_unit_number TEXT,
                plan_unit_name TEXT,
                receive_address TEXT,
                receive_place_number TEXT,
                receive_place_name TEXT,
                remark TEXT,
                transport_lead_time INTEGER,
                raw_json TEXT,
                PRIMARY KEY (order_id, line_id, delivery_id),
                FOREIGN KEY (order_id, line_id) REFERENCES order_items(order_id, line_id)
            );

            CREATE TABLE IF NOT EXISTS order_staff (
                order_id TEXT NOT NULL,
                role TEXT NOT NULL,
                staff_id TEXT,
                staff_number TEXT,
                staff_name TEXT,
                PRIMARY KEY (order_id, role),
                FOREIGN KEY (order_id) REFERENCES orders(order_id)
            );

            CREATE TABLE IF NOT EXISTS sync_state (
                source TEXT PRIMARY KEY,
                cursor_value TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );

            CREATE INDEX IF NOT EXISTS idx_orders_order_date ON orders(order_date);
            CREATE INDEX IF NOT EXISTS idx_orders_customer_name ON orders(customer_name);
            CREATE INDEX IF NOT EXISTS idx_order_items_product_id ON order_items(product_id);
            CREATE INDEX IF NOT EXISTS idx_order_deliveries_plan_date ON order_deliveries(plan_date);
            CREATE INDEX IF NOT EXISTS idx_order_staff_name_role ON order_staff(staff_name, role);
            CREATE INDEX IF NOT EXISTS idx_order_staff_number_role ON order_staff(staff_number, role);
            """
        )
        _ensure_columns(
            conn,
            "orders",
            {
                "billstatus": "TEXT",
                "orderstatus": "TEXT",
                "closestatus": "TEXT",
                "org_id": "TEXT",
                "org_number": "TEXT",
                "org_name": "TEXT",
                "dept_id": "TEXT",
                "dept_number": "TEXT",
                "dept_name": "TEXT",
                "operator_id": "TEXT",
                "operator_number": "TEXT",
                "operator_name": "TEXT",
                "total_net_amount": "REAL NOT NULL DEFAULT 0",
                "total_tax_amount": "REAL NOT NULL DEFAULT 0",
                "total_all_amount": "REAL NOT NULL DEFAULT 0",
                "receive_address": "TEXT",
                "comment": "TEXT",
                "raw_json": "TEXT",
            },
        )
        _ensure_columns(
            conn,
            "order_items",
            {
                "seq": "INTEGER",
                "material_model": "TEXT",
                "unit_number": "TEXT",
                "unit_name": "TEXT",
                "price": "REAL",
                "price_and_tax": "REAL",
                "amount_and_tax": "REAL",
                "tax_amount": "REAL",
                "delivery_date": "TEXT",
                "project_number": "TEXT",
                "project_name": "TEXT",
                "warehouse_number": "TEXT",
                "warehouse_name": "TEXT",
                "raw_json": "TEXT",
            },
        )


def _ensure_columns(conn: sqlite3.Connection, table: str, columns: dict[str, str]) -> None:
    existing = {row["name"] for row in conn.execute(f"PRAGMA table_info({table})").fetchall()}
    for name, column_type in columns.items():
        if name not in existing:
            conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {column_type}")


def load_kingdee_payload_from_json(path: str | Path) -> tuple[list[dict[str, Any]], Any]:
    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,
    config_path: str | Path | None = None,
    config_overrides: dict[str, str] | None = None,
    filters: dict[str, Any] | None = None,
    page_size: int | None = None,
    max_pages: int = 100,
) -> tuple[list[dict[str, Any]], Any]:
    config = load_kingdee_config(config_path)
    config.update(config_overrides or {})
    if api_url:
        config["KINGDEE_SALORDER_API_URL"] = api_url
    if token:
        config["KINGDEE_ACCESS_TOKEN"] = token
    safe_filters = dict(filters or {})
    if since and not safe_filters.get("start_modifytime"):
        safe_filters["start_modifytime"] = since
    return fetch_kingdee_sales_orders(
        config,
        filters=safe_filters,
        token_cache_path=token_cache_path_for_config(config_path),
        page_size=page_size,
        max_pages=max_pages,
    )


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", "rows", "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 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; when internal_parties/internal_party_ids are configured, templates exclude internal company transfer orders by default. Set include_internal_orders=true only for intercompany-flow audits. 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.",
            "intercompany_dedupe": "For multi-company architectures, pass internal_parties name substrings and/or internal_party_ids. Templates exclude an order only when both the sales organization and ordering customer match configured internal parties, so direct external customer orders to a factory are still counted.",
        },
        "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",
            "internal_parties",
            "internal_party_ids",
            "include_internal_orders",
        ],
    }


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 _coerce_string_list(value: Any) -> tuple[str, ...]:
    if value in (None, ""):
        return ()
    if isinstance(value, str):
        values = re.split(r"[,，\n]", value)
    elif isinstance(value, (list, tuple, set)):
        values = []
        for item in value:
            values.extend(re.split(r"[,，\n]", str(item)))
    else:
        values = [value]
    return tuple(str(item).strip() for item in values if str(item).strip())


def _truthy(value: Any) -> bool:
    if isinstance(value, bool):
        return value
    return str(value).strip().lower() in {"1", "true", "yes", "y", "on"}


def _internal_party_match_sql(
    *,
    name_fields: tuple[str, ...],
    id_fields: tuple[str, ...],
    party_terms: tuple[str, ...],
    party_ids: tuple[str, ...],
) -> tuple[str, list[Any]]:
    clauses: list[str] = []
    args: list[Any] = []
    for term in party_terms:
        for field in name_fields:
            clauses.append(f"{field} LIKE ?")
            args.append(f"%{term}%")
    for party_id in party_ids:
        for field in id_fields:
            clauses.append(f"{field} = ?")
            args.append(party_id)
    return "(" + " OR ".join(clauses) + ")", args


def _internal_transfer_filter_sql(params: dict[str, Any], alias: str = "o") -> tuple[list[str], list[Any]]:
    if _truthy(params.get("include_internal_orders")):
        return [], []
    party_terms = _coerce_string_list(params.get("internal_parties"))
    party_ids = _coerce_string_list(params.get("internal_party_ids"))
    if not party_terms and not party_ids:
        return [], []

    org_clause, org_args = _internal_party_match_sql(
        name_fields=(f"{alias}.org_name",),
        id_fields=(f"{alias}.org_id", f"{alias}.org_number"),
        party_terms=party_terms,
        party_ids=party_ids,
    )
    customer_clause, customer_args = _internal_party_match_sql(
        name_fields=(f"{alias}.customer_name",),
        id_fields=(f"{alias}.customer_id",),
        party_terms=party_terms,
        party_ids=party_ids,
    )
    return [f"NOT ({org_clause} AND {customer_clause})"], [*org_args, *customer_args]


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])
    internal_clauses, internal_args = _internal_transfer_filter_sql(params, alias=alias)
    clauses.extend(internal_clauses)
    args.extend(internal_args)
    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]],
    *,
    source_payload: Any | None = None,
    synced_at: str | None = None,
    db_path: Path | None = None,
    cursor_from: str | None = None,
    cursor_to: 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()
    cursor_to = cursor_to or synced_at

    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, cursor_to, len(records), upserted, run_id),
            )
            conn.execute(
                """
                INSERT INTO sync_state (source, cursor_value, updated_at)
                VALUES (?, ?, ?)
                ON CONFLICT(source) DO UPDATE SET
                    cursor_value=excluded.cursor_value,
                    updated_at=excluded.updated_at
                """,
                (KINGDEE_SYNC_STATE_SOURCE, cursor_to, synced_at),
            )
        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,
        "cursor_from": cursor_from,
        "cursor_to": cursor_to,
    }


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", "billno", "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", "bizdate", "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_number", "customer_id", "customerId", "FCustomerId") or _first(customer, "number", "id"))
    customer_name = _text(_first(record, "customer_name", "customerName", "FCustomerName") or _first(customer, "name"))
    items = _items(record)
    total_amount_value = _first(record, "total_amount", "totalallamount", "totalamount", "amount", "amount_total", "FAllAmount", "FTotalAmount")
    total_all_amount = _float(total_amount_value)
    explicit_total_all_amount = _first(record, "totalallamount", "total_all_amount")
    return {
        "order_id": order_id,
        "order_no": order_no,
        "order_date": order_date,
        "customer_id": customer_id,
        "customer_name": customer_name,
        "total_amount": total_all_amount,
        "currency": _text(_first(record, "settlecurrency_number", "currency_number", "currency", "FCurrency")),
        "status": _status_text(record),
        "updated_at": _date_time_text(_first(record, "updated_at", "modifytime", "updatedAt", "FModifyDate", "modify_date")),
        "billstatus": _text(record.get("billstatus")),
        "orderstatus": _text(record.get("orderstatus")),
        "closestatus": _text(record.get("closestatus")),
        "org_id": _text(record.get("org_id")),
        "org_number": _text(record.get("org_number")),
        "org_name": _text(record.get("org_name")),
        "dept_id": _text(record.get("dept_id")),
        "dept_number": _text(record.get("dept_number")),
        "dept_name": _text(record.get("dept_name")),
        "operator_id": _text(record.get("operator_id")),
        "operator_number": _text(_first(record, "operator_operatornumber", "operator_number")),
        "operator_name": _text(_first(record, "operator_operatorname", "operator_name")),
        "total_net_amount": _float(_first(record, "totalamount", "total_net_amount")),
        "total_tax_amount": _float(_first(record, "totaltaxamount", "total_tax_amount")),
        "total_all_amount": _float(explicit_total_all_amount) if explicit_total_all_amount is not None else total_all_amount,
        "receive_address": _text(_first(record, "receiveaddress", "receive_address")),
        "comment": _text(record.get("comment")),
        "raw_json": _json_dumps(record),
        "staff": _staff_roles(record),
        "items": items,
    }


def query_business_question(
    vault_path: str | Path,
    question: str,
    *,
    db_path: Path | None = None,
    now: str | None = None,
    all_staff: bool = False,
    fallback_sql: str | None = None,
    business_params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    db_path = db_path or business_db_path_for_vault(vault_path)
    init_db(db_path)
    params = dict(business_params or {})
    q = question.strip()
    lower = q.lower()
    staff_name = _extract_staff_name(q)
    if staff_name:
        include_all_staff = all_staff or _question_requests_all_staff(q)
        start, end, range_label = parse_date_range(q, now=now)
        if "amount" in lower or "sales" in lower or "销售额" in q or "金额" in q:
            return staff_sales_amount(
                db_path,
                staff_name,
                start,
                end,
                range_label=range_label,
                all_staff=include_all_staff,
                now=now,
                params=params,
            )
        if "多少" in q or "几单" in q or "接了" in q or "count" in lower:
            return staff_order_count(
                db_path,
                staff_name,
                start,
                end,
                range_label=range_label,
                all_staff=include_all_staff,
                now=now,
                params=params,
            )
        if "订单" in q or "order" in lower:
            return staff_order_list(
                db_path,
                staff_name,
                start,
                end,
                range_label=range_label,
                all_staff=include_all_staff,
                now=now,
                params=params,
            )

    try:
        year, month = parse_month(q, now=now)
    except RuntimeError:
        if fallback_sql:
            return execute_business_sql(db_path, fallback_sql)
        return no_matching_template(q, db_path, now=now)

    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, params=params)
    if "amount" in lower or "sales" in lower or "销售额" in q or "金额" in q:
        return monthly_sales_amount(db_path, year, month, now=now, params=params)
    if "trend" in lower or "趋势" in q:
        return customer_order_trend(db_path, q, year, month, now=now, params=params)
    return monthly_order_count(db_path, year, month, now=now, params=params)


def monthly_order_count(
    db_path: Path,
    year: int,
    month: int,
    *,
    now: str | None = None,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    start, end = month_bounds(year, month)
    where, args = _order_where(params or {}, start, end)
    with connect(db_path) as conn:
        row = conn.execute(
            f"SELECT COUNT(DISTINCT o.order_no) AS value FROM orders o WHERE {where}",
            args,
        ).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,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    start, end = month_bounds(year, month)
    where, args = _order_where(params or {}, start, end)
    with connect(db_path) as conn:
        row = conn.execute(
            f"SELECT COALESCE(SUM(o.total_amount), 0) AS value FROM orders o WHERE {where}",
            args,
        ).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,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    start, end = month_bounds(year, month)
    where, args = _order_where(params or {}, start, end)
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT COALESCE(NULLIF(o.customer_name, ''), NULLIF(o.customer_id, ''), 'Unknown') AS customer,
                   COUNT(DISTINCT o.order_no) AS order_count,
                   COALESCE(SUM(o.total_amount), 0) AS sales_amount
              FROM orders o
             WHERE {where}
             GROUP BY customer
             ORDER BY sales_amount DESC, order_count DESC, customer ASC
             LIMIT ?
            """.format(where=where),
            (*args, 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,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    customer = _extract_customer_name(question)
    start, end = month_bounds(year, month)
    where, args = _order_where(params or {}, start, end)
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT substr(o.order_date, 1, 7) AS month,
                   COUNT(DISTINCT o.order_no) AS order_count,
                   COALESCE(SUM(o.total_amount), 0) AS sales_amount
              FROM orders o
             WHERE {where}
               AND (? = '' OR customer_name LIKE ? OR customer_id LIKE ?)
             GROUP BY substr(order_date, 1, 7)
             ORDER BY month ASC
            """.format(where=where),
            (*args, 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 staff_order_count(
    db_path: Path,
    staff_name: str,
    start: str,
    end: str,
    *,
    range_label: str,
    all_staff: bool = False,
    now: str | None = None,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    roles = ALL_STAFF_ROLES if all_staff else PRIMARY_STAFF_ROLES
    where, args = _order_where(params or {}, start, end)
    with connect(db_path) as conn:
        row = conn.execute(
            f"""
            SELECT COUNT(DISTINCT o.order_no) AS value
              FROM orders o
              JOIN order_staff s ON s.order_id = o.order_id
             WHERE {where}
               AND ({_staff_role_placeholders(roles)})
               AND (s.staff_name LIKE ? OR s.staff_number LIKE ?)
            """,
            (*args, *roles, f"%{staff_name}%", f"%{staff_name}%"),
        ).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": "staff_order_count",
        "data": {
            "staff_name": staff_name,
            "staff_roles": list(roles),
            "start_date": start,
            "end_date_exclusive": end,
            "range_label": range_label,
            "order_count": count,
        },
        "last_sync": last_sync,
        "is_stale": is_stale,
        "answer": _answer(
            f"Result: {staff_name} has {count} order(s) for {range_label}.",
            f"Scope: roles {', '.join(roles)}, deduplicated by sales order number, order date from {start} to {end_exclusive_label(end)}.",
            last_sync,
            is_stale=is_stale,
        ),
    }


def staff_sales_amount(
    db_path: Path,
    staff_name: str,
    start: str,
    end: str,
    *,
    range_label: str,
    all_staff: bool = False,
    now: str | None = None,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    roles = ALL_STAFF_ROLES if all_staff else PRIMARY_STAFF_ROLES
    where, args = _order_where(params or {}, start, end)
    with connect(db_path) as conn:
        row = conn.execute(
            f"""
            SELECT COALESCE(SUM(total_amount), 0) AS value
              FROM (
                    SELECT DISTINCT o.order_id, o.total_amount
                      FROM orders o
                      JOIN order_staff s ON s.order_id = o.order_id
                     WHERE {where}
                       AND ({_staff_role_placeholders(roles)})
                       AND (s.staff_name LIKE ? OR s.staff_number LIKE ?)
                   )
            """,
            (*args, *roles, f"%{staff_name}%", f"%{staff_name}%"),
        ).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": "staff_sales_amount",
        "data": {
            "staff_name": staff_name,
            "staff_roles": list(roles),
            "start_date": start,
            "end_date_exclusive": end,
            "range_label": range_label,
            "sales_amount": amount,
        },
        "last_sync": last_sync,
        "is_stale": is_stale,
        "answer": _answer(
            f"Result: {staff_name} sales amount for {range_label} is {amount:.2f}.",
            f"Scope: roles {', '.join(roles)}, sum of distinct order total_amount, order date from {start} to {end_exclusive_label(end)}.",
            last_sync,
            is_stale=is_stale,
        ),
    }


def staff_order_list(
    db_path: Path,
    staff_name: str,
    start: str,
    end: str,
    *,
    range_label: str,
    all_staff: bool = False,
    now: str | None = None,
    limit: int = 100,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    roles = ALL_STAFF_ROLES if all_staff else PRIMARY_STAFF_ROLES
    where, args = _order_where(params or {}, 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, o.total_net_amount, o.total_tax_amount,
                   o.currency, o.billstatus, o.orderstatus, o.org_name
              FROM orders o
              JOIN order_staff s ON s.order_id = o.order_id
             WHERE {where}
               AND ({_staff_role_placeholders(roles)})
               AND (s.staff_name LIKE ? OR s.staff_number LIKE ?)
             ORDER BY o.order_date DESC, o.order_no ASC
             LIMIT ?
            """,
            (*args, *roles, f"%{staff_name}%", f"%{staff_name}%", limit),
        ).fetchall()
    orders = [
        {
            "order_id": row["order_id"],
            "order_no": row["order_no"],
            "order_date": row["order_date"],
            "customer_name": row["customer_name"],
            "total_amount": float(row["total_amount"] or 0),
            "total_net_amount": float(row["total_net_amount"] or 0),
            "total_tax_amount": float(row["total_tax_amount"] or 0),
            "currency": row["currency"],
            "billstatus": row["billstatus"],
            "orderstatus": row["orderstatus"],
            "org_name": row["org_name"],
        }
        for row in rows
    ]
    last_sync = last_successful_sync(db_path)
    is_stale = sync_is_stale(last_sync, now=now)
    return {
        "template": "staff_order_list",
        "data": {
            "staff_name": staff_name,
            "staff_roles": list(roles),
            "start_date": start,
            "end_date_exclusive": end,
            "range_label": range_label,
            "orders": orders,
        },
        "last_sync": last_sync,
        "is_stale": is_stale,
        "answer": _answer(
            f"Result: {staff_name} has {len(orders)} listed order(s) for {range_label}.",
            f"Scope: roles {', '.join(roles)}, order date from {start} to {end_exclusive_label(end)}, limit {limit}.",
            last_sync,
            is_stale=is_stale,
        ),
    }


def no_matching_template(question: str, db_path: Path, *, now: str | None = None) -> dict[str, Any]:
    last_sync = last_successful_sync(db_path)
    is_stale = sync_is_stale(last_sync, now=now)
    return {
        "template": "no_matching_template",
        "data": {"question": question},
        "last_sync": last_sync,
        "is_stale": is_stale,
        "answer": _answer(
            "Result: no matching fixed business query template.",
            "Scope: provide a guarded fallback SQL query to run against structured business tables.",
            last_sync,
            is_stale=is_stale,
        ),
    }


def execute_business_sql(db_path: Path, sql: str, *, limit: int = SQL_DEFAULT_LIMIT) -> dict[str, Any]:
    init_db(db_path)
    statement = _validate_business_sql(sql)
    safe_limit = max(1, min(int(limit or SQL_DEFAULT_LIMIT), SQL_MAX_LIMIT))
    executed_sql = f"SELECT * FROM ({statement}) LIMIT ?"
    with connect(db_path) as conn:
        conn.set_authorizer(_business_sql_authorizer)
        try:
            cursor = conn.execute(executed_sql, (safe_limit,))
            rows = [dict(row) for row in cursor.fetchall()]
        finally:
            conn.set_authorizer(None)
    columns = list(rows[0].keys()) if rows else []
    return {
        "template": "sql_fallback",
        "sql": statement,
        "limit": safe_limit,
        "columns": columns,
        "rows": rows,
        "row_count": len(rows),
        "answer": f"Result: SQL fallback returned {len(rows)} row(s).",
    }


def _validate_business_sql(sql: str) -> str:
    statement = sql.strip()
    if not statement:
        raise RuntimeError("SQL is empty")
    if statement.endswith(";"):
        statement = statement[:-1].strip()
    if ";" in statement:
        raise RuntimeError("SQL fallback accepts one statement only")
    first = re.match(r"^\s*([A-Za-z]+)", statement)
    keyword = first.group(1).upper() if first else ""
    if keyword == "PRAGMA":
        raise RuntimeError("PRAGMA is not allowed in SQL fallback")
    if keyword not in {"SELECT", "WITH"}:
        raise RuntimeError("SQL fallback only SELECT or WITH ... SELECT statements")
    if re.search(r"\braw_json\b", statement, re.IGNORECASE):
        raise RuntimeError("SQL fallback cannot read raw_json")
    if re.search(r"\braw_snapshots\b", statement, re.IGNORECASE):
        raise RuntimeError("SQL fallback cannot read raw_snapshots")
    forbidden = r"\b(INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|ATTACH|DETACH|VACUUM|REINDEX|REPLACE)\b"
    if re.search(forbidden, statement, re.IGNORECASE):
        raise RuntimeError("SQL fallback only SELECT or WITH ... SELECT statements")
    return statement


def _business_sql_authorizer(action: int, arg1: str | None, arg2: str | None, _db: str | None, _trigger: str | None) -> int:
    if action == sqlite3.SQLITE_READ:
        table = str(arg1 or "")
        column = str(arg2 or "")
        if table == "raw_snapshots":
            return sqlite3.SQLITE_DENY
        if column in FORBIDDEN_SQL_COLUMNS:
            return sqlite3.SQLITE_DENY
        if table and table not in ALLOWED_SQL_TABLES:
            return sqlite3.SQLITE_DENY
        return sqlite3.SQLITE_OK
    if action in {sqlite3.SQLITE_SELECT, sqlite3.SQLITE_FUNCTION}:
        return sqlite3.SQLITE_OK
    return sqlite3.SQLITE_DENY


def _staff_role_placeholders(roles: tuple[str, ...]) -> str:
    return " OR ".join("s.role = ?" for _role in roles)


def parse_date_range(text: str, *, now: str | None = None) -> tuple[str, str, str]:
    today = _today(now)
    if "最近一周" in text or "近一周" in text or "最近7天" in text or "近7天" in text:
        start = today - timedelta(days=6)
        end = today + timedelta(days=1)
        return start.isoformat(), end.isoformat(), "最近一周"
    if "今天" in text:
        start = today
        end = today + timedelta(days=1)
        return start.isoformat(), end.isoformat(), "今天"
    if "昨天" in text:
        start = today - timedelta(days=1)
        end = today
        return start.isoformat(), end.isoformat(), "昨天"
    if "本周" in text or "这周" in text:
        start = today - timedelta(days=today.weekday())
        end = start + timedelta(days=7)
        return start.isoformat(), end.isoformat(), "本周"
    if "上周" in text:
        end = today - timedelta(days=today.weekday())
        start = end - timedelta(days=7)
        return start.isoformat(), end.isoformat(), "上周"
    if "本月" in text or "这个月" in text:
        start = today.replace(day=1)
        end = _add_month(start)
        return start.isoformat(), end.isoformat(), "本月"
    try:
        year, month = parse_month(text, now=now)
        start, end = month_bounds(year, month)
        return start, end, f"{year}-{month:02d}"
    except RuntimeError:
        start = today - timedelta(days=6)
        end = today + timedelta(days=1)
        return start.isoformat(), end.isoformat(), "最近一周"


def _today(now: str | None = None) -> date:
    raw = now or utc_now()
    return datetime.fromisoformat(raw.replace("Z", "+00:00")).date()


def _add_month(value: date) -> date:
    if value.month == 12:
        return date(value.year + 1, 1, 1)
    return date(value.year, value.month + 1, 1)


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"(?<!\d)(\d{1,2})\s*月", text)
        if not month_match:
            raise RuntimeError("question must include a month, such as 2026-06 or 2026年6月")
        year = int((now or utc_now())[:4])
        month = int(month_match.group(1))
    if month < 1 or month > 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 compute_incremental_start_modifytime(
    db_path: Path,
    *,
    now: str | None = None,
    lookback_hours: int = 48,
    initial_days: int = 7,
) -> str:
    init_db(db_path)
    cursor = _sync_state_cursor(db_path) or last_successful_sync(db_path)
    if cursor:
        start = _parse_datetime(cursor) - timedelta(hours=lookback_hours)
    else:
        start = _parse_datetime(now or utc_now()) - timedelta(days=initial_days)
    return _format_kingdee_datetime(start)


def _sync_state_cursor(db_path: Path) -> str:
    with connect(db_path) as conn:
        row = conn.execute(
            "SELECT cursor_value FROM sync_state WHERE source=?",
            (KINGDEE_SYNC_STATE_SOURCE,),
        ).fetchone()
    return str(row["cursor_value"] 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),
        )
    order_columns_sql = ", ".join(ORDER_COLUMNS)
    order_placeholders = ", ".join("?" for _ in (*ORDER_COLUMNS, "raw_hash", "created_at", "synced_at"))
    order_updates = ",\n            ".join(f"{column}=excluded.{column}" for column in ORDER_COLUMNS if column != "order_id")
    conn.execute(
        f"""
        INSERT INTO orders ({order_columns_sql}, raw_hash, created_at, synced_at)
        VALUES ({order_placeholders})
        ON CONFLICT(order_id) DO UPDATE SET
            {order_updates},
            raw_hash=excluded.raw_hash,
            synced_at=excluded.synced_at
        """,
        tuple(order[column] for column in ORDER_COLUMNS) + (raw_hash, synced_at, synced_at),
    )
    conn.execute("DELETE FROM order_staff WHERE order_id=?", (order["order_id"],))
    staff_columns_sql = ", ".join(ORDER_STAFF_COLUMNS)
    staff_placeholders = ", ".join("?" for _ in ORDER_STAFF_COLUMNS)
    for staff in order["staff"]:
        staff_row = {"order_id": order["order_id"], **staff}
        conn.execute(
            f"""
            INSERT INTO order_staff ({staff_columns_sql})
            VALUES ({staff_placeholders})
            """,
            tuple(staff_row[column] for column in ORDER_STAFF_COLUMNS),
        )
    conn.execute("DELETE FROM order_deliveries WHERE order_id=?", (order["order_id"],))
    conn.execute("DELETE FROM order_items WHERE order_id=?", (order["order_id"],))
    item_columns_sql = ", ".join(ORDER_ITEM_COLUMNS)
    item_placeholders = ", ".join("?" for _ in ORDER_ITEM_COLUMNS)
    delivery_columns_sql = ", ".join(ORDER_DELIVERY_COLUMNS)
    delivery_placeholders = ", ".join("?" for _ in ORDER_DELIVERY_COLUMNS)
    for item in order["items"]:
        if item["product_id"]:
            conn.execute(
                """
                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),
            )
        item_row = {"order_id": order["order_id"], **item}
        conn.execute(
            f"""
            INSERT INTO order_items ({item_columns_sql})
            VALUES ({item_placeholders})
            """,
            tuple(item_row[column] for column in ORDER_ITEM_COLUMNS),
        )
        for delivery in item["deliveries"]:
            delivery_row = {"order_id": order["order_id"], "line_id": item["line_id"], **delivery}
            conn.execute(
                f"""
                INSERT INTO order_deliveries ({delivery_columns_sql})
                VALUES ({delivery_placeholders})
                """,
                tuple(delivery_row[column] for column in ORDER_DELIVERY_COLUMNS),
            )


def _staff_roles(record: dict[str, Any]) -> list[dict[str, str]]:
    staff: list[dict[str, str]] = []
    for role, id_field, number_field, name_field in STAFF_ROLE_FIELDS:
        staff_id = _text(record.get(id_field))
        staff_number = _text(record.get(number_field))
        staff_name = _text(record.get(name_field))
        if not staff_id and not staff_number and not staff_name:
            continue
        staff.append(
            {
                "role": role,
                "staff_id": staff_id,
                "staff_number": staff_number,
                "staff_name": staff_name,
            }
        )
    return staff


def _items(record: dict[str, Any]) -> list[dict[str, Any]]:
    raw_items = None
    for key in ("items", "billentry", "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, "material_masterid_number", "product_id", "productId", "material_id", "FMaterialId") or _first(product, "number", "id"))
        product_name = _text(_first(raw, "material_masterid_name", "product_name", "productName", "material_name", "FMaterialName") or _first(product, "name"))
        line_id = _text(_first(raw, "line_id", "id", "line_no", "entry_id", "FEntryID", "seq")) or str(index)
        items.append(
            {
                "line_id": line_id,
                "product_id": product_id,
                "product_name": product_name,
                "quantity": _float(_first(raw, "quantity", "qty", "FQty")),
                "amount": _float(_first(raw, "amount", "line_amount", "FAmount", "amountandtax")),
                "seq": _int(_first(raw, "seq", "line_no")),
                "material_model": _text(_first(raw, "material_masterid_modelnum", "material_model", "model")),
                "unit_number": _text(_first(raw, "unit_number", "unitNumber", "FUnitNumber")),
                "unit_name": _text(_first(raw, "unit_name", "unitName", "FUnitName")),
                "price": _float(_first(raw, "price", "FPrice")),
                "price_and_tax": _float(_first(raw, "priceandtax", "price_and_tax")),
                "amount_and_tax": _float(_first(raw, "amountandtax", "amount_and_tax")),
                "tax_amount": _float(_first(raw, "taxamount", "tax_amount")),
                "delivery_date": _date_text(_first(raw, "deliverydate", "delivery_date")),
                "project_number": _text(_first(raw, "project_number", "projectNumber")),
                "project_name": _text(_first(raw, "project_name", "projectName")),
                "warehouse_number": _text(_first(raw, "warehouse_number", "warehouseNumber")),
                "warehouse_name": _text(_first(raw, "warehouse_name", "warehouseName")),
                "raw_json": _json_dumps(raw),
                "deliveries": _deliveries(raw),
            }
        )
    return items


def _deliveries(item: dict[str, Any]) -> list[dict[str, Any]]:
    raw_deliveries = None
    for key in ("orderdeliverentry", "deliveries", "delivery_entries", "plan_entries"):
        if isinstance(item.get(key), list):
            raw_deliveries = item[key]
            break
    if raw_deliveries is None:
        raw_deliveries = []
    deliveries = []
    for index, raw in enumerate(raw_deliveries, start=1):
        if not isinstance(raw, dict):
            continue
        deliveries.append(
            {
                "delivery_id": _text(_first(raw, "id", "delivery_id", "entry_id")) or str(index),
                "plan_date": _date_text(_first(raw, "d_plandate", "plan_date")),
                "plan_delivery_date": _date_text(_first(raw, "d_plandeliverydate", "plan_delivery_date")),
                "plan_qty": _float(_first(raw, "d_planqty", "plan_qty")),
                "plan_base_qty": _float(_first(raw, "d_planbaseqty", "plan_base_qty")),
                "plan_unit_number": _text(_first(raw, "d_planunit_number", "plan_unit_number")),
                "plan_unit_name": _text(_first(raw, "d_planunit_name", "plan_unit_name")),
                "receive_address": _text(_first(raw, "d_receiveaddress", "receive_address")),
                "receive_place_number": _text(_first(raw, "d_receiveaddressf7_number", "receive_place_number")),
                "receive_place_name": _text(_first(raw, "d_receiveaddressf7_name", "receive_place_name")),
                "remark": _text(_first(raw, "d_remark", "remark")),
                "transport_lead_time": _int(_first(raw, "d_transportleadtime", "transport_lead_time")),
                "raw_json": _json_dumps(raw),
            }
        )
    return deliveries


def _first(record: dict[str, Any], *keys: str) -> Any:
    for key in keys:
        if key in record and record[key] not in (None, ""):
            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 _parse_datetime(value: str) -> datetime:
    text = _text(value)
    if not text:
        return datetime.now(timezone.utc).replace(microsecond=0)
    normalized = text.replace("Z", "+00:00")
    if " " in normalized and "T" not in normalized:
        normalized = normalized.replace(" ", "T", 1)
    parsed = datetime.fromisoformat(normalized)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc).replace(tzinfo=None, microsecond=0)


def _format_kingdee_datetime(value: datetime) -> str:
    return value.replace(microsecond=0).strftime("%Y-%m-%d %H:%M:%S")


def _float(value: Any) -> float:
    if value in (None, ""):
        return 0.0
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def _int(value: Any) -> int | None:
    if value in (None, ""):
        return None
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _json_dumps(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str)


def _extract_customer_name(question: str) -> str:
    match = re.search(r"customer\s+(.+?)\s+trend", question, re.IGNORECASE)
    return match.group(1).strip() if match else ""


def _extract_staff_name(question: str) -> str:
    if not any(token in question for token in ("销售员", "销售额", "订单", "接了", "多少单")) and not re.search(
        r"\border(s)?\b|\bsales\b|\bstaff\b", question, re.IGNORECASE
    ):
        return ""
    text = question.strip()
    text = re.sub(r"(20\d{2})\s*[-/年.]\s*\d{1,2}\s*月?", " ", text)
    remove_patterns = (
        "最近一周",
        "近一周",
        "最近7天",
        "近7天",
        "今天",
        "昨天",
        "本周",
        "这周",
        "上周",
        "本月",
        "这个月",
        "接了多少单",
        "接多少单",
        "多少单",
        "几单",
        "销售额",
        "金额",
        "订单列表",
        "订单",
        "全部角色",
        "所有销售角色",
        "所有角色",
        "含销售员2",
        "含跟单",
        "包括跟单",
        "包含跟单",
        "主销售员",
        "销售员",
    )
    for pattern in remove_patterns:
        text = text.replace(pattern, " ")
    text = re.sub(r"\b(order|orders|sales|amount|count|staff|all roles|primary)\b", " ", text, flags=re.IGNORECASE)
    text = re.sub(r"[\s,，。:：?？]+", " ", text).strip()
    candidate = text.split()[0] if text else ""
    if candidate.startswith(("按", "列出", "统计", "查询")):
        return ""
    if re.fullmatch(r"[\u4e00-\u9fff]+", candidate) and len(candidate) > 4:
        return ""
    return candidate


def _question_requests_all_staff(question: str) -> bool:
    return any(
        token in question
        for token in ("全部角色", "所有销售角色", "所有角色", "含销售员2", "含跟单", "包括跟单", "包含跟单")
    )


def _status_text(record: dict[str, Any]) -> str:
    parts = []
    for key in ("billstatus", "orderstatus", "closestatus", "status", "document_status", "FDocumentStatus"):
        value = _text(record.get(key))
        if value:
            parts.append(f"{key}={value}")
    return ";".join(parts)