216 lines
8.0 KiB
Python
216 lines
8.0 KiB
Python
"""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 "暂无相关经营数据。"
|