184 lines
6.7 KiB
Python
184 lines
6.7 KiB
Python
"""AI 开单：自然语言文本 → 结构化订单行 → 匹配商品（批次 B2）。
|
||
|
||
链路：文本 --LLM--> [{name, barcode?, qty, unit?}] --匹配--> {matched, unmatched}
|
||
匹配优先级：barcode 精确 → 商品编码精确 → 品名精确 → 品名包含 → 相似度（difflib）
|
||
**无 LLM KEY 时抛 LlmUnavailable**（视图翻译为 400 明确错误，不 500）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from decimal import Decimal, InvalidOperation
|
||
from difflib import SequenceMatcher
|
||
|
||
from . import llm
|
||
|
||
|
||
class LlmUnavailable(Exception):
|
||
"""未配置 LLM（无 AI_API_KEY），AI 开单不可用。"""
|
||
|
||
|
||
PARSE_PROMPT = """你是进销存系统的录单助手。把下面这段人类写的订货文本，抽取成 JSON 数组。
|
||
|
||
规则：
|
||
- 每行一个对象：{"name": "商品名", "barcode": "条码或空串", "qty": 数量数字, "unit": "单位或空串"}
|
||
- 数量缺失时默认 1；"两箱"→2、"三瓶"→3 等中文数字要转成阿拉伯数字
|
||
- 只输出 JSON 数组，不要任何解释
|
||
|
||
文本：
|
||
{text}
|
||
"""
|
||
|
||
|
||
_CN_NUM = {"零": 0, "一": 1, "二": 2, "两": 2, "三": 3, "四": 4, "五": 5,
|
||
"六": 6, "七": 7, "八": 8, "九": 9, "十": 10}
|
||
|
||
|
||
def _cn_to_int(token: str) -> int | None:
|
||
"""把"三"/"十二"/"两"等中文数字转 int（够用即可，不追求完整语法）。"""
|
||
if token in _CN_NUM:
|
||
return _CN_NUM[token]
|
||
if token.startswith("十") and len(token) == 2 and token[1] in _CN_NUM:
|
||
return 10 + _CN_NUM[token[1]]
|
||
if len(token) == 2 and token[0] in _CN_NUM and token[1] == "十":
|
||
return _CN_NUM[token[0]] * 10
|
||
if len(token) == 3 and token[0] in _CN_NUM and token[1] == "十" and token[2] in _CN_NUM:
|
||
return _CN_NUM[token[0]] * 10 + _CN_NUM[token[2]]
|
||
return None
|
||
|
||
|
||
def fallback_parse(text: str) -> list:
|
||
"""无 LLM 时的规则兜底：按行/逗号切分，抓 "商品名 数字单位" 模式。
|
||
|
||
仅在显式要求（allow_rule_fallback=True）时使用——默认严格走 LLM，
|
||
保证"没有 KEY 就明确报不可用"的验收标准。
|
||
"""
|
||
items = []
|
||
for raw in re.split(r"[\n\r,，;；]+", text or ""):
|
||
line = raw.strip()
|
||
if not line:
|
||
continue
|
||
m = re.search(r"(\d+(?:\.\d+)?)\s*([^\s\d]*)", line)
|
||
qty = Decimal("1")
|
||
unit = ""
|
||
name = line
|
||
if m:
|
||
qty = Decimal(m.group(1))
|
||
unit = (m.group(2) or "").strip()
|
||
name = (line[:m.start()] + line[m.end():]).strip() or line
|
||
else:
|
||
# 中文数字：三箱 / 两瓶
|
||
m2 = re.search(r"([零一二两三四五六七八九十]+)\s*([^\s\d]*)", line)
|
||
if m2:
|
||
n = _cn_to_int(m2.group(1))
|
||
if n:
|
||
qty = Decimal(n)
|
||
unit = (m2.group(2) or "").strip()
|
||
name = (line[:m2.start()] + line[m2.end():]).strip() or line
|
||
# 去掉常见量词残留
|
||
name = re.sub(r"(箱|瓶|件|个|包|袋|提|盒|罐|桶|斤|公斤|kg|Kg|KG)$", "", name).strip()
|
||
if name:
|
||
items.append({"name": name, "barcode": "", "qty": float(qty), "unit": unit})
|
||
return items
|
||
|
||
|
||
def _norm_qty(value) -> Decimal:
|
||
try:
|
||
q = Decimal(str(value))
|
||
except (InvalidOperation, TypeError, ValueError):
|
||
q = Decimal("1")
|
||
return q if q > 0 else Decimal("1")
|
||
|
||
|
||
def extract_items(text: str, *, allow_rule_fallback: bool = False) -> list:
|
||
"""文本 → [{name, barcode, qty, unit}]。无 KEY 抛 LlmUnavailable。"""
|
||
if not text or not text.strip():
|
||
return []
|
||
|
||
if not llm.available():
|
||
if allow_rule_fallback:
|
||
return fallback_parse(text)
|
||
raise LlmUnavailable("未配置 AI 服务（AI_API_KEY），AI 录单暂不可用")
|
||
|
||
data = llm.extract_json(PARSE_PROMPT.format(text=text.strip()))
|
||
if not isinstance(data, list):
|
||
# LLM 偶发返回 {"items": [...]}
|
||
if isinstance(data, dict) and isinstance(data.get("items"), list):
|
||
data = data["items"]
|
||
else:
|
||
return []
|
||
|
||
items = []
|
||
for row in data:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
name = str(row.get("name") or "").strip()
|
||
if not name:
|
||
continue
|
||
items.append({
|
||
"name": name,
|
||
"barcode": str(row.get("barcode") or "").strip(),
|
||
"qty": float(_norm_qty(row.get("qty"))),
|
||
"unit": str(row.get("unit") or "").strip(),
|
||
})
|
||
return items
|
||
|
||
|
||
def match_products(tenant, items: list, *, unit_resolver=None) -> dict:
|
||
"""把抽取结果匹配到商品档案。
|
||
|
||
返回 {matched: [{...item, product_id, product_code, product_name, match_by,
|
||
score, unit_id?, unit_name?, price?}], unmatched: [...]}
|
||
"""
|
||
from apps.catalog.models import Product
|
||
|
||
products = list(Product.objects.filter(tenant=tenant, is_deleted=False))
|
||
by_barcode = {p.barcode: p for p in products if p.barcode}
|
||
by_code = {p.code.lower(): p for p in products}
|
||
by_name = {p.name: p for p in products}
|
||
|
||
matched, unmatched = [], []
|
||
for item in items:
|
||
hit, how, score = None, "", 0.0
|
||
|
||
if item.get("barcode") and item["barcode"] in by_barcode:
|
||
hit, how, score = by_barcode[item["barcode"]], "barcode", 100.0
|
||
elif item["name"].lower() in by_code:
|
||
hit, how, score = by_code[item["name"].lower()], "code", 100.0
|
||
elif item["name"] in by_name:
|
||
hit, how, score = by_name[item["name"]], "name_exact", 100.0
|
||
else:
|
||
# 包含匹配（长度优先，避免短名吃掉长名）
|
||
cands = [p for p in products if item["name"] and item["name"] in p.name]
|
||
if cands:
|
||
hit = sorted(cands, key=lambda p: len(p.name))[0]
|
||
how, score = "name_contains", 85.0
|
||
else:
|
||
# 相似度兜底
|
||
best, best_score = None, 0.0
|
||
for p in products:
|
||
r = SequenceMatcher(None, item["name"], p.name).ratio()
|
||
if r > best_score:
|
||
best, best_score = p, r
|
||
if best is not None and best_score >= 0.6:
|
||
hit, how, score = best, "fuzzy", round(best_score * 100, 1)
|
||
|
||
if hit is None:
|
||
unmatched.append(item)
|
||
continue
|
||
|
||
row = {
|
||
**item,
|
||
"product_id": hit.id,
|
||
"product_code": hit.code,
|
||
"product_name": hit.name,
|
||
"match_by": how,
|
||
"score": score,
|
||
"base_price": str(hit.sale_price),
|
||
}
|
||
if unit_resolver is not None:
|
||
row.update(unit_resolver(tenant, hit, item))
|
||
matched.append(row)
|
||
|
||
return {"matched": matched, "unmatched": unmatched}
|