"""AI 经营问答（批次 B3）：白名单工具取数 → LLM 组织语言。 安全设计（计划要求"只读、不给 LLM 写权限"）： - 取数完全由本模块的**白名单工具**完成，LLM 只拿到聚合后的数字，不能生成查询； - 所有工具都是只读 ORM，且强制 `tenant=tenant`（越权查他租户不可能）； - 无 LLM KEY 时返回结构化数据 + `llm_enhanced=False`（前端可隐藏入口）。 """ from __future__ import annotations from datetime import date, timedelta from . import llm # 意图 → 工具名（关键词命中即取该工具数据） INTENT_KEYWORDS = [ ("sales", ["销售", "卖了", "营收", "营业额", "出货"]), ("receivable", ["应收", "欠款", "回款", "催收", "未收", "账款"]), ("inventory", ["库存", "存货", "积压", "周转", "缺货"]), ("purchase", ["采购", "进货", "买入"]), ("profit", ["利润", "毛利", "赚", "成本", "盈利"]), ("ranking", ["排行", "最好卖", "畅销", "大客户", "top", "TOP"]), ] def detect_intents(question: str) -> list: """按关键词命中意图；都没命中则给一个"全局概览"。""" q = question or "" hits = [name for name, kws in INTENT_KEYWORDS if any(k in q for k in kws)] return hits or ["overview"] # ------------------------------------------------------------ # 白名单工具（全部只读 + 租户隔离） # ------------------------------------------------------------ def tool_sales(tenant) -> dict: from apps.report.services import get_dashboard_summary s = get_dashboard_summary(tenant) return {"本月销售额": float(s["month_sales"]["amount"]), "本月销售单数": s["month_sales"]["count"]} def tool_purchase(tenant) -> dict: from apps.report.services import get_dashboard_summary s = get_dashboard_summary(tenant) return {"本月采购额": float(s["month_purchase"]["amount"]), "本月进货单数": s["month_purchase"]["count"]} def tool_inventory(tenant) -> dict: from apps.report.services import get_dashboard_summary s = get_dashboard_summary(tenant) return {"库存估值": float(s["inventory"]["valuation"]), "SKU 数": s["inventory"]["sku_count"], "库存件数": float(s["inventory"]["total_quantity"])} def tool_receivable(tenant) -> dict: from decimal import Decimal from apps.finance.models import Receivable from apps.finance.services import receivable_aging aging = receivable_aging(tenant) overdue_amount = Decimal("0") for b in aging["buckets"]: if b["bucket"] not in ("0-30",): overdue_amount += Decimal(str(b["amount"])) return { "应收未结合计": float(aging["total"]), "其中 30 天以上": float(overdue_amount), "账龄分桶": {b["bucket"]: float(b["amount"]) for b in aging["buckets"]}, } def tool_profit(tenant) -> dict: from apps.finance import services as fin_services period = fin_services.current_period(tenant) if period is None: return {"本月营业收入": 0.0, "本月营业成本": 0.0, "本月毛利": 0.0, "本月净利润": 0.0} data = fin_services.income_statement(tenant, period) return { "本月营业收入": float(data.get("total_revenue") or 0), "本月营业成本": float(data.get("cogs") or 0), "本月毛利": float(data.get("gross_profit") or 0), "本月净利润": float(data.get("net_profit") or 0), } def tool_ranking(tenant) -> dict: from apps.report.services import get_sales_rank today = date.today() start = today - timedelta(days=30) prods = get_sales_rank(tenant, start_date=start, rank_by="product", top_n=5) custs = get_sales_rank(tenant, start_date=start, rank_by="customer", top_n=5) return { "近30天商品排行": [ {"商品": p["product_name"], "销售额": float(p["total_amount"])} for p in prods ], "近30天客户排行": [ {"客户": c["customer_name"], "销售额": float(c["total_amount"])} for c in custs ], } def tool_risk(tenant) -> dict: from . import risk as ai_risk top = ai_risk.risk_ranking(tenant, top_n=5) return { "风险客户TOP5": [ {"客户": r["customer_name"], "风险分": r["score"], "档位": r["level"], "未结": r["outstanding"]} for r in top ] } TOOLS = { "sales": tool_sales, "purchase": tool_purchase, "inventory": tool_inventory, "receivable": tool_receivable, "profit": tool_profit, "ranking": tool_ranking, "risk": tool_risk, } def gather(tenant, question: str) -> dict: """按意图取数：返回 {intents, data}。'overview' 时给全景四项。""" intents = detect_intents(question) if intents == ["overview"]: intents = ["sales", "receivable", "inventory", "profit"] data = {} for name in intents: fn = TOOLS.get(name) if fn is None: continue try: data[name] = fn(tenant) except Exception as exc: # 单个工具失败不影响整体问答 data[name] = {"error": str(exc)} return {"intents": intents, "data": data} ASK_PROMPT = """你是经销商进销存系统的"老板参谋"。下面是系统实时取到的经营数据（JSON）， 请用中文回答老板的问题：先给结论，再给 1-2 条可执行建议。总长不超过 120 字。 问题：{question} 数据：{data} """ def answer(tenant, question: str) -> dict: """问答主入口：取数 →（有 LLM 则）组织语言。""" gathered = gather(tenant, question) payload = { "question": question, "intents": gathered["intents"], "data": gathered["data"], "llm_enhanced": False, "answer": None, } if llm.available(): import json text = llm.collect_advice( ASK_PROMPT.format(question=question, data=json.dumps(gathered["data"], ensure_ascii=False)) ) if text: payload["answer"] = text payload["llm_enhanced"] = True if not payload["answer"]: payload["answer"] = _stat_answer(gathered) return payload def _stat_answer(gathered: dict) -> str: """无 LLM 时的统计口径回答（结构化文本，保证功能不空转）。""" parts = [] d = gathered["data"] if "sales" in d and "error" not in d["sales"]: parts.append(f"本月销售额 ¥{d['sales'].get('本月销售额', 0):,.2f}" f"（{d['sales'].get('本月销售单数', 0)} 单）") if "purchase" in d and "error" not in d["purchase"]: parts.append(f"本月采购额 ¥{d['purchase'].get('本月采购额', 0):,.2f}") if "receivable" in d and "error" not in d["receivable"]: parts.append(f"应收未结 ¥{d['receivable'].get('应收未结合计', 0):,.2f}" f"，其中 30 天以上 ¥{d['receivable'].get('其中 30 天以上', 0):,.2f}") if "inventory" in d and "error" not in d["inventory"]: parts.append(f"库存估值 ¥{d['inventory'].get('库存估值', 0):,.2f}" f"（{d['inventory'].get('SKU 数', 0)} 个 SKU）") if "profit" in d and "error" not in d["profit"]: parts.append(f"本月净利润 ¥{d['profit'].get('本月净利润', 0):,.2f}") if "ranking" in d and "error" not in d["ranking"]: top = (d["ranking"].get("近30天商品排行") or [{}])[0] if top.get("商品"): parts.append(f"近 30 天最畅销：{top['商品']}（¥{top.get('销售额', 0):,.2f}）") if "risk" in d and "error" not in d["risk"]: risk_top = (d["risk"].get("风险客户TOP5") or [{}])[0] if risk_top.get("客户"): parts.append(f"应收风险最高：{risk_top['客户']}" f"（{risk_top.get('风险分', 0):.0f} 分 / {risk_top.get('档位')}）") return "；".join(parts) + "。" if parts else "暂无相关经营数据。"