1260 lines
48 KiB
Python
1260 lines
48 KiB
Python
"""AI 智能服务模块
|
||
|
||
提供两大核心能力：
|
||
1. OCR 图片识别：将图片中的文字提取为结构化数据，支持阿里云 OCR 和 Mock 两种 provider。
|
||
2. 智能填单：通过文本预处理、规则提取、LLM 结构化解析、产品库模糊匹配等多层管线，
|
||
从粘贴文本或图片中自动识别订单信息并填充到表单。
|
||
|
||
架构分层：
|
||
- TextPreprocessor：清洗粘贴文本中的时间戳、表情标记等噪声。
|
||
- RuleExtractor：基于正则的确定性字段提取（手机号、数量、价格、地址）。
|
||
- OrderOCRParser：基于 OCR 行列表做订单版面分析和字段提取。
|
||
- LLMOrderParser：调用通义千问 qwen-plus 做订单文本结构化解析。
|
||
- BaseOCRProvider / AliyunOCRProvider / MockOCRProvider：OCR 提供者抽象与实现。
|
||
- AIService：业务入口，串联上述组件完成图片识别和智能填单流程。
|
||
|
||
被调用方：ai 路由（图片识别、识别结果修正、智能填单接口）。
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
from decimal import Decimal
|
||
from difflib import SequenceMatcher
|
||
from urllib import error, parse, request
|
||
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.core.config import get_settings
|
||
from backend.app.core.error_codes import ErrorCode
|
||
from backend.app.core.exceptions import AppException
|
||
from backend.app.repositories.ai_repository import AIRepository
|
||
from backend.app.services.audit_service import audit_service
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 智能填单：文本预处理
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TextPreprocessor:
|
||
"""清洗粘贴文本中的噪声，保留有效订单信息。
|
||
|
||
处理内容包括：移除时间戳、括号时间标记、表情/媒体标记、用户名前缀等，
|
||
并合并多余空行。无外部依赖，纯文本处理。
|
||
"""
|
||
|
||
TIMESTAMP_RE = re.compile(r'\d{4}[-/]\d{1,2}[-/]\d{1,2}\s+\d{1,2}:\d{2}(:\d{2})?')
|
||
BRACKET_TIME_RE = re.compile(r'\[\d{1,2}:\d{2}(:\d{2})?\]')
|
||
USERNAME_PREFIX_RE = re.compile(r'^[^:\s]{1,10}[：:]\s*', re.MULTILINE)
|
||
NOISE_MARKERS = [
|
||
'[图片]', '[表情]', '[语音]', '[视频]', '[文件]', '[链接]',
|
||
'[红包]', '[转账]', '[位置]', '[名片]', '—— ——',
|
||
]
|
||
|
||
def preprocess(self, text: str) -> str:
|
||
"""清洗文本，移除噪声内容并合并多余空行。
|
||
|
||
参数：
|
||
text: 原始粘贴文本。
|
||
|
||
返回：
|
||
清洗后的纯文本，空输入返回空字符串。
|
||
"""
|
||
if not text or not text.strip():
|
||
return ""
|
||
result = text
|
||
result = self.TIMESTAMP_RE.sub('', result)
|
||
result = self.BRACKET_TIME_RE.sub('', result)
|
||
for marker in self.NOISE_MARKERS:
|
||
result = result.replace(marker, '')
|
||
result = self.USERNAME_PREFIX_RE.sub('', result)
|
||
result = re.sub(r'\n{3,}', '\n\n', result)
|
||
return result.strip()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 智能填单：规则提取层
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class RuleExtractor:
|
||
"""基于规则的字段提取器，处理确定性高的字段。
|
||
|
||
通过正则表达式提取手机号、数量+单位、单价，并基于地址关键词密度识别地址段。
|
||
适用于文本预处理后的清洗文本。
|
||
"""
|
||
|
||
PHONE_RE = re.compile(r'1[3-9]\d{9}')
|
||
QTY_UNIT_RE = re.compile(
|
||
r'(\d+\.?\d*)\s*(吨|件|箱|包|个|米|kg|KG|公斤|斤|卷|组|套|台|条|根|片|块)'
|
||
)
|
||
PRICE_RE = re.compile(r'(?:单价|价格|报价)\s*[:：]?\s*(\d+\.?\d*)')
|
||
ADDRESS_KEYWORDS = [
|
||
'省', '市', '区', '县', '镇', '路', '街', '号',
|
||
'楼', '室', '栋', '单元', '村', '大厦', '广场',
|
||
]
|
||
|
||
def extract(self, text: str) -> dict:
|
||
"""从文本中提取确定性字段。
|
||
|
||
参数：
|
||
text: 清洗后的文本。
|
||
|
||
返回：
|
||
字典，包含 customer_mobile、_raw_quantity、_raw_unit、_raw_price、customer_address 等字段。
|
||
"""
|
||
result: dict = {}
|
||
phones = self.PHONE_RE.findall(text)
|
||
if phones:
|
||
result['customer_mobile'] = phones[0]
|
||
qty_match = self.QTY_UNIT_RE.search(text)
|
||
if qty_match:
|
||
result['_raw_quantity'] = float(qty_match.group(1))
|
||
result['_raw_unit'] = qty_match.group(2)
|
||
price_match = self.PRICE_RE.search(text)
|
||
if price_match:
|
||
result['_raw_price'] = float(price_match.group(1))
|
||
addr = self._extract_address(text)
|
||
if addr:
|
||
result['customer_address'] = addr
|
||
return result
|
||
|
||
def _extract_address(self, text: str) -> str | None:
|
||
"""基于地址关键词密度提取地址段。
|
||
|
||
逐行扫描，关键词命中数 >= 2 的连续行视为地址段，取最长段。
|
||
|
||
参数：
|
||
text: 清洗后的文本。
|
||
|
||
返回：
|
||
拼接后的地址字符串，未识别到时返回 None。
|
||
"""
|
||
lines = text.split('\n')
|
||
best: list[str] = []
|
||
current: list[str] = []
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
continue
|
||
addr_hits = sum(1 for kw in self.ADDRESS_KEYWORDS if kw in stripped)
|
||
if addr_hits >= 2:
|
||
current.append(stripped)
|
||
else:
|
||
if len(current) > len(best):
|
||
best = list(current)
|
||
current = []
|
||
if len(current) > len(best):
|
||
best = current
|
||
return ''.join(best) if best else None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 智能填单：OCR 版面分析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class OrderOCRParser:
|
||
"""基于 OCR 行列表做订单版面分析。
|
||
|
||
将 OCR 返回的文本行按区域分类（头部、地址、表格、其他），
|
||
从各区域中提取客户信息和产品明细。
|
||
"""
|
||
|
||
PHONE_RE = re.compile(r'1[3-9]\d{9}')
|
||
ADDRESS_KEYWORDS = {
|
||
'省', '市', '区', '县', '镇', '路', '街', '号',
|
||
'楼', '室', '栋', '单元', '村', '大厦', '广场',
|
||
'弄', '巷', '苑', '园', '城',
|
||
}
|
||
QTY_RE = re.compile(
|
||
r'(\d+\.?\d*)\s*(吨|件|箱|包|个|米|kg|KG|公斤|斤|卷|组|套|台|条|根|片|块)'
|
||
)
|
||
PRICE_RE = re.compile(r'(?:单价|价格|报价|￥|¥)\s*[:：]?\s*(\d+\.?\d*)')
|
||
TABLE_HEADER_KEYWORDS = {'产品', '品名', '名称', '规格', '数量', '单价', '金额', '合计'}
|
||
|
||
def parse(self, line_list: list[str], ocr_confidence: float) -> dict:
|
||
"""解析 OCR 行列表，输出结构化的版面分析结果。
|
||
|
||
参数：
|
||
line_list: OCR 返回的文本行列表。
|
||
ocr_confidence: OCR 识别置信度。
|
||
|
||
返回：
|
||
包含 raw_text、ocr_confidence、layout_zones、pre_filled、table_rows 的字典。
|
||
"""
|
||
raw_text = '\n'.join(line_list)
|
||
zones = self._classify_lines(line_list)
|
||
pre_filled = self._extract_from_zones(line_list, zones)
|
||
table_rows = self._parse_table_lines(line_list, zones.get('table_lines', []))
|
||
return {
|
||
"raw_text": raw_text,
|
||
"ocr_confidence": ocr_confidence,
|
||
"layout_zones": zones,
|
||
"pre_filled": pre_filled,
|
||
"table_rows": table_rows,
|
||
}
|
||
|
||
def _classify_lines(self, lines: list[str]) -> dict:
|
||
"""将 OCR 文本行按区域类型分类。
|
||
|
||
分类逻辑：前 3 行为头部、含地址关键词的连续行归为地址区、
|
||
遇到表格表头后连续的产品行归为表格区，其余为其他。
|
||
|
||
参数：
|
||
lines: OCR 文本行列表。
|
||
|
||
返回：
|
||
包含 header_lines、address_lines、table_lines、other_lines 的分类索引字典。
|
||
"""
|
||
zones: dict[str, list[int]] = {
|
||
"header_lines": [], "address_lines": [],
|
||
"table_lines": [], "other_lines": [],
|
||
}
|
||
table_started = False
|
||
for i, line in enumerate(lines):
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
continue
|
||
if any(kw in stripped for kw in self.TABLE_HEADER_KEYWORDS):
|
||
table_started = True
|
||
zones["table_lines"].append(i)
|
||
continue
|
||
if table_started:
|
||
if self.QTY_RE.search(stripped) or self.PRICE_RE.search(stripped):
|
||
zones["table_lines"].append(i)
|
||
continue
|
||
table_started = False
|
||
addr_count = sum(1 for kw in self.ADDRESS_KEYWORDS if kw in stripped)
|
||
if addr_count >= 2:
|
||
zones["address_lines"].append(i)
|
||
continue
|
||
if i < 3:
|
||
zones["header_lines"].append(i)
|
||
else:
|
||
zones["other_lines"].append(i)
|
||
return zones
|
||
|
||
def _extract_from_zones(self, lines: list[str], zones: dict) -> dict:
|
||
"""从已分类的区域中提取客户基本信息（姓名、手机、地址）。
|
||
|
||
参数：
|
||
lines: OCR 文本行列表。
|
||
zones: 区域分类结果字典。
|
||
|
||
返回：
|
||
包含 customer_mobile、customer_name、customer_address 的预填字段字典。
|
||
"""
|
||
result: dict = {}
|
||
for i in zones.get("header_lines", []):
|
||
text = lines[i].strip()
|
||
if not text:
|
||
continue
|
||
phone_match = self.PHONE_RE.search(text)
|
||
if phone_match:
|
||
result["customer_mobile"] = phone_match.group(0)
|
||
name_part = self.PHONE_RE.sub('', text).strip()
|
||
if name_part and len(name_part) <= 20 and not any(
|
||
kw in name_part for kw in self.ADDRESS_KEYWORDS
|
||
):
|
||
result.setdefault("customer_name", name_part)
|
||
addr_parts = [lines[i].strip() for i in zones.get("address_lines", [])]
|
||
if addr_parts:
|
||
result["customer_address"] = ''.join(addr_parts)
|
||
return result
|
||
|
||
def _parse_table_lines(self, lines: list[str], table_indices: list[int]) -> list[dict]:
|
||
"""解析表格区域中的产品行，提取数量、单价和产品名称。
|
||
|
||
参数：
|
||
lines: OCR 文本行列表。
|
||
table_indices: 表格区域的行索引列表。
|
||
|
||
返回：
|
||
产品行列表，每个元素包含 raw（原始行）和 fields（提取的字段）字典。
|
||
"""
|
||
rows: list[dict] = []
|
||
for i in table_indices:
|
||
line = lines[i].strip()
|
||
if not line:
|
||
continue
|
||
if any(kw in line for kw in self.TABLE_HEADER_KEYWORDS) and not self.QTY_RE.search(line):
|
||
continue
|
||
row: dict = {"raw": line, "fields": {}}
|
||
qty_match = self.QTY_RE.search(line)
|
||
if qty_match:
|
||
row["fields"]["quantity"] = float(qty_match.group(1))
|
||
row["fields"]["unit"] = qty_match.group(2)
|
||
price_match = self.PRICE_RE.search(line)
|
||
if price_match:
|
||
row["fields"]["sale_price"] = float(price_match.group(1))
|
||
name_part = self.QTY_RE.sub('', line)
|
||
name_part = self.PRICE_RE.sub('', name_part)
|
||
name_part = re.sub(r'\d+\.?\d*', '', name_part).strip()
|
||
if name_part:
|
||
row["fields"]["product_name"] = name_part
|
||
if row["fields"]:
|
||
rows.append(row)
|
||
return rows
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 智能填单：LLM 结构化解析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class LLMOrderParser:
|
||
"""调用通义千问 qwen-plus 进行订单文本结构化解析。
|
||
|
||
支持纯文本和 OCR 图片两种输入模式，通过 system prompt 引导 LLM 输出标准 JSON 格式的订单信息。
|
||
内置产品库提示词（product_hints），帮助 LLM 匹配产品名称和规格。
|
||
"""
|
||
|
||
TEXT_SYSTEM_PROMPT = """你是订单信息解析助手。从用户提供的文本中提取订单信息。
|
||
|
||
严格按以下 JSON 格式输出，不要输出其他内容：
|
||
{
|
||
"customer_name": "客户姓名",
|
||
"customer_mobile": "手机号",
|
||
"customer_address": "地址",
|
||
"order_source": "订单来源（如能识别）",
|
||
"delivery_type": "配送方式（如能识别）",
|
||
"remark": "备注",
|
||
"items": [
|
||
{
|
||
"product_name": "产品名称",
|
||
"specification": "规格",
|
||
"unit": "单位",
|
||
"quantity": 数量数字,
|
||
"sale_price": 单价数字
|
||
}
|
||
]
|
||
}
|
||
|
||
规则：
|
||
- 手机号必须是 11 位数字，以 1 开头
|
||
- 数量和价格必须是数字（不是字符串）
|
||
- 无法识别的字段填 null，不要编造
|
||
- 如果文本中有多个产品，每个产品一个 items 条目
|
||
|
||
可选的产品库（名称 + 规格）：
|
||
{product_hints}"""
|
||
|
||
IMAGE_SYSTEM_PROMPT = """你是订单信息解析助手。OCR 系统已经从图片中提取了文本并做了初步分析。
|
||
你需要基于 OCR 的结果，补充和完善订单信息。
|
||
|
||
OCR 已提取的信息：
|
||
- OCR 置信度：{ocr_confidence}
|
||
- 已识别的客户姓名：{customer_name}
|
||
- 已识别的客户手机：{customer_mobile}
|
||
- 已识别的客户地址：{customer_address}
|
||
- 产品表格区域识别到的原始行：
|
||
{table_rows}
|
||
|
||
你需要完成：
|
||
1. 验证 OCR 提取的字段是否合理
|
||
2. 从原始文本中补充 OCR 未提取到的字段
|
||
3. 解析产品明细（如果 OCR 表格区域数据可用，优先使用）
|
||
4. 识别订单来源、配送方式等附加信息
|
||
|
||
严格按以下 JSON 格式输出：
|
||
{
|
||
"customer_name": "客户姓名",
|
||
"customer_mobile": "手机号",
|
||
"customer_address": "地址",
|
||
"order_source": "订单来源",
|
||
"delivery_type": "配送方式",
|
||
"remark": "备注",
|
||
"items": [
|
||
{
|
||
"product_name": "产品名称",
|
||
"specification": "规格",
|
||
"unit": "单位",
|
||
"quantity": 数量数字,
|
||
"sale_price": 单价数字
|
||
}
|
||
]
|
||
}
|
||
|
||
规则：
|
||
- 无法识别的字段填 null，不要编造
|
||
- 如果 OCR 已提取的字段看起来正确，直接沿用
|
||
|
||
可选的产品库（名称 + 规格）：
|
||
{product_hints}"""
|
||
|
||
def _build_product_hints(self, product_groups: list[dict]) -> str:
|
||
"""构建产品库提示词，供 LLM 在解析时参考。
|
||
|
||
参数：
|
||
product_groups: 产品分组列表，每组包含 product_name 和 specifications。
|
||
|
||
返回：
|
||
格式化的产品提示文本，最多 30 个产品，每个最多 5 个规格。
|
||
"""
|
||
hints: list[str] = []
|
||
for group in product_groups[:30]:
|
||
specs = ', '.join(
|
||
f"{s['specification']}({s['unit']})"
|
||
for s in group.get("specifications", [])[:5]
|
||
)
|
||
hints.append(f"- {group['product_name']}: {specs}")
|
||
return '\n'.join(hints) or "（产品库为空）"
|
||
|
||
def parse(self, text: str, product_groups: list[dict],
|
||
api_key: str, api_url: str,
|
||
ocr_context: dict | None = None) -> dict:
|
||
"""调用 LLM 解析订单文本，返回结构化订单数据。
|
||
|
||
参数：
|
||
text: 待解析的订单文本。
|
||
product_groups: 产品库数据，用于构建提示词。
|
||
api_key: LLM API 访问密钥。
|
||
api_url: LLM API 地址。
|
||
ocr_context: 可选的 OCR 上下文，图片模式时提供版面分析结果。
|
||
|
||
返回：
|
||
包含 customer_name、customer_mobile、items 等字段的订单字典。
|
||
|
||
异常：
|
||
LLM 调用失败或返回格式异常时抛出 AppException。
|
||
"""
|
||
product_hints = self._build_product_hints(product_groups)
|
||
if ocr_context:
|
||
system_prompt = self.IMAGE_SYSTEM_PROMPT.format(
|
||
ocr_confidence=ocr_context.get("ocr_confidence", "N/A"),
|
||
customer_name=ocr_context.get("pre_filled", {}).get("customer_name", "未识别"),
|
||
customer_mobile=ocr_context.get("pre_filled", {}).get("customer_mobile", "未识别"),
|
||
customer_address=ocr_context.get("pre_filled", {}).get("customer_address", "未识别"),
|
||
table_rows='\n'.join(
|
||
f" - {row['raw']}" for row in ocr_context.get("table_rows", [])
|
||
) or " 无",
|
||
product_hints=product_hints,
|
||
)
|
||
else:
|
||
system_prompt = self.TEXT_SYSTEM_PROMPT.format(product_hints=product_hints)
|
||
|
||
payload = json.dumps({
|
||
"model": "qwen-plus",
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": f"请解析以下订单内容：\n\n{text}"},
|
||
],
|
||
"temperature": 0.1,
|
||
"max_tokens": 1024,
|
||
}).encode("utf-8")
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {api_key}",
|
||
}
|
||
req = urllib_request.Request(
|
||
url=api_url, data=payload, headers=headers, method="POST"
|
||
)
|
||
with urllib_request.urlopen(req, timeout=30) as resp:
|
||
result = json.loads(resp.read().decode("utf-8"))
|
||
|
||
content = result["choices"][0]["message"]["content"]
|
||
return self._extract_json(content)
|
||
|
||
def _extract_json(self, text: str) -> dict:
|
||
"""从 LLM 响应文本中提取 JSON 内容。
|
||
|
||
处理 LLM 可能包裹在 markdown 代码块中的情况。
|
||
|
||
参数：
|
||
text: LLM 原始响应文本。
|
||
|
||
返回：
|
||
解析后的字典。
|
||
"""
|
||
text = text.strip()
|
||
if text.startswith("```"):
|
||
text = text.split("\n", 1)[1]
|
||
text = text.rsplit("```", 1)[0]
|
||
return json.loads(text.strip())
|
||
|
||
def safe_parse(self, text: str, product_groups: list[dict],
|
||
api_key: str, api_url: str,
|
||
ocr_context: dict | None = None) -> dict | None:
|
||
"""安全版解析入口，异常时返回 None 而非抛出异常。
|
||
|
||
参数：
|
||
text: 待解析的订单文本。
|
||
product_groups: 产品库数据。
|
||
api_key: LLM API 访问密钥。
|
||
api_url: LLM API 地址。
|
||
ocr_context: 可选的 OCR 上下文。
|
||
|
||
返回：
|
||
解析后的订单字典，失败时返回 None。
|
||
"""
|
||
try:
|
||
return self.parse(text, product_groups, api_key, api_url, ocr_context)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# 避免和后面 urllib_request 冲突，这里在文件顶部已经 import 了
|
||
urllib_request = request
|
||
|
||
|
||
class BaseOCRProvider:
|
||
"""OCR 提供者基类，定义统一的识别接口。
|
||
|
||
子类需实现 recognize 方法，返回 (raw_result, suggested_result, confidence) 三元组。
|
||
"""
|
||
provider_name = "base"
|
||
|
||
def recognize(self, image_url: str, biz_type: str, biz_id: int) -> tuple[dict, dict, float]:
|
||
"""识别图片中的文字内容。
|
||
|
||
参数：
|
||
image_url: 图片 URL 地址。
|
||
biz_type: 业务类型标识。
|
||
biz_id: 业务记录 ID。
|
||
|
||
返回：
|
||
三元组 (原始识别结果, 建议结果, 置信度)。
|
||
"""
|
||
raise NotImplementedError
|
||
|
||
|
||
class MockOCRProvider(BaseOCRProvider):
|
||
"""模拟 OCR 提供者，用于开发调试和测试环境。
|
||
|
||
不调用真实 OCR 服务，直接返回固定格式的模拟结果，置信度 0.92。
|
||
"""
|
||
provider_name = "mock_aliyun_adapter"
|
||
|
||
def recognize(self, image_url: str, biz_type: str, biz_id: int) -> tuple[dict, dict, float]:
|
||
"""返回模拟的 OCR 识别结果。
|
||
|
||
参数：
|
||
image_url: 图片 URL 地址。
|
||
biz_type: 业务类型标识。
|
||
biz_id: 业务记录 ID。
|
||
|
||
返回：
|
||
三元组 (模拟原始结果, 模拟建议结果, 0.92)。
|
||
"""
|
||
# 模拟 OCR 识别出的文本行，用于开发调试时走通完整解析流程
|
||
sample_lines = [
|
||
"张三 13800138000",
|
||
"广东省深圳市南山区科技园路88号",
|
||
"产品名称：工业级密封圈",
|
||
"规格：DN50 数量：100件 单价：15.5",
|
||
"合计金额：1550.00",
|
||
"备注：尽快发货",
|
||
]
|
||
suggested_result = {
|
||
"image_name": image_url.rsplit("/", 1)[-1],
|
||
"biz_type": biz_type,
|
||
"biz_id": biz_id,
|
||
"line_list": sample_lines,
|
||
"recognized_text": "\n".join(sample_lines),
|
||
}
|
||
raw_result = {
|
||
"provider": self.provider_name,
|
||
"image_url": image_url,
|
||
"fields": suggested_result,
|
||
}
|
||
return raw_result, suggested_result, 0.92
|
||
|
||
|
||
class AliyunOCRProvider(BaseOCRProvider):
|
||
"""阿里云 OCR 提供者，通过官方 SDK 调用阿里云文字识别服务。
|
||
|
||
依赖 settings 中的 aliyun_ai_access_key_id、aliyun_ai_access_key_secret、
|
||
aliyun_ai_region 等配置。使用 alibabacloud_ocr_api20210707 SDK。
|
||
"""
|
||
provider_name = "aliyun_ocr"
|
||
|
||
ENDPOINT = "ocr-api.cn-hangzhou.aliyuncs.com"
|
||
|
||
def __init__(self, settings) -> None:
|
||
self.settings = settings
|
||
|
||
def recognize(self, image_url: str, biz_type: str, biz_id: int) -> tuple[dict, dict, float]:
|
||
"""调用阿里云 OCR 识别图片中的文字。
|
||
|
||
参数：
|
||
image_url: 图片 URL 地址。
|
||
biz_type: 业务类型标识。
|
||
biz_id: 业务记录 ID。
|
||
|
||
返回：
|
||
三元组 (原始 API 响应, 结构化建议结果, 置信度)。
|
||
"""
|
||
payload = self._call_aliyun(image_url)
|
||
raw_result = {
|
||
"provider": self.provider_name,
|
||
"region": self.settings.aliyun_ai_region,
|
||
"image_url": image_url,
|
||
"payload": payload,
|
||
}
|
||
suggested_result = self._build_suggested_result(payload, image_url, biz_type, biz_id)
|
||
confidence = self._extract_confidence(payload)
|
||
return raw_result, suggested_result, confidence
|
||
|
||
def _get_client(self):
|
||
"""构建阿里云 OCR SDK 客户端。"""
|
||
from alibabacloud_ocr_api20210707.client import Client as OcrClient
|
||
from alibabacloud_tea_openapi.models import Config
|
||
|
||
config = Config(
|
||
access_key_id=self.settings.aliyun_ai_access_key_id,
|
||
access_key_secret=self.settings.aliyun_ai_access_key_secret,
|
||
endpoint=self.ENDPOINT,
|
||
)
|
||
return OcrClient(config)
|
||
|
||
def _call_aliyun(self, image_url: str) -> dict:
|
||
"""通过官方 SDK 调用阿里云 OCR 接口。
|
||
|
||
先从 URL 下载图片字节，再通过 SDK 发送识别请求。
|
||
|
||
参数：
|
||
image_url: 图片 URL 地址。
|
||
|
||
返回：
|
||
阿里云 OCR SDK 的响应字典。
|
||
|
||
异常：
|
||
下载图片失败或 SDK 调用失败时抛出 AppException。
|
||
"""
|
||
from alibabacloud_ocr_api20210707.models import RecognizeGeneralRequest
|
||
from urllib import request as urllib_request, error as urllib_error
|
||
from urllib.parse import urlparse, quote, urlunparse
|
||
|
||
# 对 URL 中的非 ASCII 字符（如中文文件名）进行编码，避免 ASCII 编码异常
|
||
parsed = urlparse(image_url)
|
||
safe_url = urlunparse(parsed._replace(path=quote(parsed.path)))
|
||
|
||
try:
|
||
with urllib_request.urlopen(safe_url, timeout=15) as resp:
|
||
image_bytes = resp.read()
|
||
except (urllib_error.HTTPError, urllib_error.URLError) as exc:
|
||
raise AppException(
|
||
code=ErrorCode.THIRD_PARTY_FAILED,
|
||
message=f"下载 OCR 图片失败：{exc}",
|
||
status_code=400,
|
||
) from exc
|
||
|
||
try:
|
||
client = self._get_client()
|
||
ocr_request = RecognizeGeneralRequest(body=image_bytes)
|
||
response = client.recognize_general(ocr_request)
|
||
except Exception as exc:
|
||
raise AppException(
|
||
code=ErrorCode.THIRD_PARTY_FAILED,
|
||
message=f"阿里云 OCR SDK 调用失败：{exc}",
|
||
status_code=400,
|
||
) from exc
|
||
|
||
body = response.body
|
||
if hasattr(body, "to_map"):
|
||
payload = body.to_map()
|
||
elif isinstance(body, dict):
|
||
payload = body
|
||
else:
|
||
payload = json.loads(str(body)) if body else {}
|
||
|
||
# SDK 返回 Data 为 JSON 字符串，解析为 dict 便于后续统一提取
|
||
if isinstance(payload.get("Data"), str):
|
||
try:
|
||
payload["Data"] = json.loads(payload["Data"])
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
|
||
return payload
|
||
|
||
def _build_suggested_result(self, payload: dict, image_url: str, biz_type: str, biz_id: int) -> dict:
|
||
"""将阿里云 OCR 原始响应转换为统一的结构化建议结果。
|
||
|
||
从响应中递归提取文本行，拼接为 recognized_text 和 line_list。
|
||
|
||
参数：
|
||
payload: 阿里云 OCR API 响应字典。
|
||
image_url: 图片 URL 地址。
|
||
biz_type: 业务类型标识。
|
||
biz_id: 业务记录 ID。
|
||
|
||
返回：
|
||
包含 image_name、biz_type、recognized_text、line_list 等字段的建议结果字典。
|
||
"""
|
||
# 不强依赖某个固定响应字段，先把常见 OCR 结果平铺为统一结构，便于前端先联调。
|
||
texts = self._collect_texts(payload)
|
||
result = {
|
||
"image_name": image_url.rsplit("/", 1)[-1],
|
||
"biz_type": biz_type,
|
||
"biz_id": biz_id,
|
||
"recognized_text": "\n".join(texts).strip(),
|
||
"line_list": texts,
|
||
}
|
||
if texts:
|
||
result["customer_name"] = texts[0][:50]
|
||
return result
|
||
|
||
def _collect_texts(self, value) -> list[str]:
|
||
"""从嵌套的字典/列表结构中递归提取所有文本值并去重。
|
||
|
||
识别常见 OCR 响应字段名（text、content、word、words 等），
|
||
跳过空值和重复文本。
|
||
|
||
参数：
|
||
value: OCR 响应中的任意嵌套结构。
|
||
|
||
返回：
|
||
去重后的文本列表，保持原始出现顺序。
|
||
"""
|
||
texts: list[str] = []
|
||
if isinstance(value, dict):
|
||
for key, item in value.items():
|
||
lowered = str(key).lower()
|
||
if lowered in {"text", "content", "word", "words", "value", "label", "title"} and isinstance(item, str):
|
||
stripped = item.strip()
|
||
if stripped:
|
||
texts.append(stripped)
|
||
else:
|
||
texts.extend(self._collect_texts(item))
|
||
elif isinstance(value, list):
|
||
for item in value:
|
||
texts.extend(self._collect_texts(item))
|
||
elif isinstance(value, str):
|
||
stripped = value.strip()
|
||
if stripped:
|
||
texts.append(stripped)
|
||
deduplicated: list[str] = []
|
||
seen: set[str] = set()
|
||
for text in texts:
|
||
if text not in seen:
|
||
seen.add(text)
|
||
deduplicated.append(text)
|
||
return deduplicated
|
||
|
||
def _extract_confidence(self, payload: dict) -> float:
|
||
"""从 OCR 响应中提取置信度分数。
|
||
|
||
依次尝试 confidence、score、probability 字段，未找到时默认返回 0.9。
|
||
|
||
参数：
|
||
payload: OCR API 响应字典。
|
||
|
||
返回：
|
||
置信度浮点数（0.0 ~ 1.0）。
|
||
"""
|
||
for key in ("confidence", "score", "probability"):
|
||
value = payload.get(key)
|
||
if isinstance(value, (int, float)):
|
||
return float(value)
|
||
if isinstance(value, str):
|
||
try:
|
||
return float(value)
|
||
except ValueError:
|
||
continue
|
||
return 0.9
|
||
|
||
|
||
class AIService:
|
||
"""AI 智能业务服务入口。
|
||
|
||
整合 OCR 识别、规则提取、LLM 解析、产品库匹配等能力，
|
||
提供图片识别（recognize_image）、识别结果修正（correct_result）、
|
||
智能填单（parse_order）三大对外接口。
|
||
|
||
依赖：
|
||
- AIRepository：AI 识别日志的存储。
|
||
- get_settings：系统配置，包含 API 密钥和 provider 选择。
|
||
- AuditService：操作审计日志记录。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self.repository = AIRepository()
|
||
self.settings = get_settings()
|
||
|
||
def recognize_image(self, payload: dict, session: Session | None = None) -> dict:
|
||
"""调用 OCR 识别图片中的文字并保存识别日志。
|
||
|
||
参数：
|
||
payload: 请求数据字典，必含 image_url、biz_type，可选 biz_id。
|
||
session: 数据库会话，不可为 None。
|
||
|
||
返回：
|
||
包含 log_id、raw_result、confidence、suggested_result 的字典。
|
||
|
||
被调用方：ai 路由（图片识别接口）。
|
||
"""
|
||
image_url = (payload.get("image_url") or "").strip()
|
||
biz_type = (payload.get("biz_type") or "").strip()
|
||
biz_id = payload.get("biz_id")
|
||
|
||
if not image_url:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="图片地址不能为空", status_code=400)
|
||
if not biz_type:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="业务类型不能为空", status_code=400)
|
||
if image_url.startswith("mock://fail"):
|
||
raise AppException(code=ErrorCode.THIRD_PARTY_FAILED, message="AI 识别失败", status_code=400)
|
||
|
||
provider = self._build_provider()
|
||
raw_result, suggested_result, confidence = provider.recognize(image_url, biz_type, int(biz_id or 0))
|
||
|
||
if session is not None:
|
||
try:
|
||
# 识别原始响应和建议结果都落库，后面人工修正和问题回溯都依赖这份快照。
|
||
log = self.repository.create_log(
|
||
session,
|
||
{
|
||
"biz_type": biz_type,
|
||
"biz_id": biz_id,
|
||
"image_url": image_url,
|
||
"raw_result": json.dumps(raw_result, ensure_ascii=False),
|
||
"confidence": Decimal(str(confidence)),
|
||
"corrected_result": None,
|
||
"created_by": None,
|
||
},
|
||
)
|
||
audit_service.write_log(
|
||
session,
|
||
{
|
||
"operate_type": "ai_recognize",
|
||
"biz_type": biz_type,
|
||
"biz_id": biz_id,
|
||
"before_value": None,
|
||
"after_value": {
|
||
"log_id": log.id,
|
||
"provider": raw_result.get("provider"),
|
||
"confidence": confidence,
|
||
"suggested_result": suggested_result,
|
||
},
|
||
"remark": f"AI识别图片 {image_url}",
|
||
},
|
||
)
|
||
session.commit()
|
||
return {
|
||
"log_id": log.id,
|
||
"raw_result": raw_result,
|
||
"confidence": confidence,
|
||
"suggested_result": suggested_result,
|
||
}
|
||
except AppException:
|
||
session.rollback()
|
||
raise
|
||
except SQLAlchemyError as exc:
|
||
session.rollback()
|
||
raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc
|
||
|
||
raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库连接不可用", status_code=500)
|
||
|
||
def correct_result(self, log_id: int, payload: dict, session: Session | None = None) -> dict:
|
||
"""人工修正 AI 识别结果并保存。
|
||
|
||
参数：
|
||
log_id: AI 识别日志 ID。
|
||
payload: 修正数据字典，必含 corrected_result（dict）。
|
||
session: 数据库会话，不可为 None。
|
||
|
||
返回：
|
||
包含 log_id 和 corrected 标志的字典。
|
||
|
||
被调用方：ai 路由（识别结果修正接口）。
|
||
"""
|
||
corrected_result = payload.get("corrected_result")
|
||
if not isinstance(corrected_result, dict) or not corrected_result:
|
||
raise AppException(code=ErrorCode.PARAM_ERROR, message="修正结果不能为空", status_code=400)
|
||
|
||
if session is not None:
|
||
try:
|
||
log = self.repository.get_log(session, log_id)
|
||
if log is None:
|
||
raise AppException(code=ErrorCode.NOT_FOUND, message="识别记录不存在", status_code=404)
|
||
before_value = self._safe_load_json(log.corrected_result)
|
||
log.corrected_result = json.dumps(corrected_result, ensure_ascii=False)
|
||
session.add(log)
|
||
audit_service.write_log(
|
||
session,
|
||
{
|
||
"operate_type": "ai_correct",
|
||
"biz_type": log.biz_type,
|
||
"biz_id": log.biz_id,
|
||
"before_value": before_value,
|
||
"after_value": corrected_result,
|
||
"remark": f"修正AI识别结果 {log_id}",
|
||
},
|
||
)
|
||
session.commit()
|
||
return {"log_id": log.id, "corrected": True}
|
||
except AppException:
|
||
session.rollback()
|
||
raise
|
||
except SQLAlchemyError as exc:
|
||
session.rollback()
|
||
raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库异常", status_code=500) from exc
|
||
|
||
raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库连接不可用", status_code=500)
|
||
|
||
def _build_provider(self) -> BaseOCRProvider:
|
||
"""根据系统配置构建对应的 OCR 提供者实例。
|
||
|
||
当 ai_provider 配置为 'aliyun' 时返回 AliyunOCRProvider，否则返回 MockOCRProvider。
|
||
|
||
返回：
|
||
OCR 提供者实例。
|
||
"""
|
||
provider_name = self.settings.ai_provider.strip().lower()
|
||
# provider 选择集中在这里，后续扩展腾讯云或本地 OCR 时不需要动主流程。
|
||
if provider_name == "aliyun":
|
||
return AliyunOCRProvider(self.settings)
|
||
return MockOCRProvider()
|
||
|
||
def _safe_load_json(self, value: str | None):
|
||
"""安全解析 JSON 字符串，异常时返回原始值或 None。
|
||
|
||
参数：
|
||
value: JSON 字符串或 None。
|
||
|
||
返回：
|
||
解析后的对象，解析失败返回原始字符串，空值返回 None。
|
||
"""
|
||
if not value:
|
||
return None
|
||
try:
|
||
return json.loads(value)
|
||
except json.JSONDecodeError:
|
||
return value
|
||
|
||
# ------------------------------------------------------------------
|
||
# 智能填单：核心解析入口
|
||
# ------------------------------------------------------------------
|
||
|
||
def parse_order(self, payload: dict, session: Session | None = None) -> dict:
|
||
"""智能填单核心入口：从文本或图片中解析订单信息。
|
||
|
||
处理流程：
|
||
1. 获取原始文本（图片模式先调 OCR）
|
||
2. 文本预处理（去噪声）
|
||
3. 规则提取（正则提取手机号、数量、价格、地址）
|
||
4. LLM 结构化解析
|
||
5. 规则结果与 LLM 结果合并
|
||
6. 客户库预匹配
|
||
7. 产品库模糊匹配
|
||
8. 校验并计算置信度
|
||
|
||
参数：
|
||
payload: 请求数据字典，必含 input_type（'text' 或 'image'），
|
||
text 模式需提供 text 字段，image 模式需提供 image_url 字段。
|
||
session: 数据库会话，为 None 时使用 mock 模式。
|
||
|
||
返回：
|
||
包含 parsed_order、raw_text、confidence、parse_source、product_matches、
|
||
factory_options、warnings 的字典。
|
||
|
||
被调用方：ai 路由（智能填单接口）。
|
||
"""
|
||
input_type = payload["input_type"]
|
||
|
||
# 1. 获取原始文本 + OCR 中间结果
|
||
ocr_context = None
|
||
if input_type == "image":
|
||
ocr_context = self._ocr_and_parse_image(payload["image_url"])
|
||
raw_text = ocr_context["raw_text"]
|
||
else:
|
||
raw_text = payload["text"].strip()
|
||
|
||
# 2. 文本预处理
|
||
preprocessor = TextPreprocessor()
|
||
raw_text = preprocessor.preprocess(raw_text)
|
||
if not raw_text:
|
||
raise AppException(
|
||
code=ErrorCode.PARAM_ERROR,
|
||
message="无法提取到文本内容",
|
||
status_code=400,
|
||
)
|
||
|
||
# 3. 正则提取
|
||
rule_extractor = RuleExtractor()
|
||
rule_result = rule_extractor.extract(raw_text)
|
||
if ocr_context and ocr_context.get("pre_filled"):
|
||
for key, value in ocr_context["pre_filled"].items():
|
||
if value and not rule_result.get(key):
|
||
rule_result[key] = value
|
||
|
||
# 4. LLM 结构化解析
|
||
settings = self.settings
|
||
product_groups = self._get_product_groups(session)
|
||
llm_parser = LLMOrderParser()
|
||
llm_result = llm_parser.safe_parse(
|
||
raw_text,
|
||
product_groups,
|
||
settings.llm_parse_api_key or settings.aliyun_ai_access_key_id,
|
||
settings.llm_parse_api_url,
|
||
ocr_context=ocr_context,
|
||
)
|
||
|
||
# 5. 合并
|
||
if llm_result:
|
||
merged = self._merge_results(rule_result, llm_result)
|
||
parse_source = "hybrid"
|
||
else:
|
||
merged = self._build_fallback_result(rule_result)
|
||
parse_source = "rule"
|
||
|
||
# 6. 客户库预匹配
|
||
if session is not None and merged.get("customer_mobile"):
|
||
try:
|
||
from backend.app.repositories.customer_repository import CustomerRepository
|
||
customer_repo = CustomerRepository()
|
||
existing_customer = customer_repo.find_by_mobile(
|
||
session, merged["customer_mobile"]
|
||
)
|
||
if existing_customer:
|
||
merged["customer_id"] = existing_customer.id
|
||
if existing_customer.address and not merged.get("customer_address"):
|
||
merged["customer_address"] = existing_customer.address
|
||
except Exception:
|
||
pass
|
||
|
||
# 7. 产品库模糊匹配
|
||
product_matches = self._match_products(
|
||
merged.get("items", []), product_groups, session
|
||
)
|
||
|
||
# 8. 校验 + 置信度
|
||
warnings = self._validate_parsed_order(merged)
|
||
if parse_source == "rule":
|
||
warnings.append("LLM 服务不可用，仅使用规则提取，部分字段可能不完整")
|
||
ocr_conf = ocr_context["ocr_confidence"] if ocr_context else None
|
||
confidence = self._calc_confidence(merged, warnings, ocr_confidence=ocr_conf)
|
||
|
||
return {
|
||
"parsed_order": merged,
|
||
"raw_text": raw_text,
|
||
"confidence": confidence,
|
||
"ocr_confidence": ocr_conf,
|
||
"parse_source": parse_source,
|
||
"is_mock": session is None,
|
||
"product_matches": product_matches,
|
||
"factory_options": self._get_factory_options(session),
|
||
"warnings": warnings,
|
||
}
|
||
|
||
def _ocr_and_parse_image(self, image_url: str) -> dict:
|
||
"""调用 OCR 识别图片并进行版面分析。
|
||
|
||
参数：
|
||
image_url: 图片 URL 地址。
|
||
|
||
返回：
|
||
OrderOCRParser.parse() 的返回结果，包含 raw_text、pre_filled、table_rows 等。
|
||
"""
|
||
provider = self._build_provider()
|
||
raw_result, suggested_result, ocr_confidence = provider.recognize(
|
||
image_url, "order_parse", 0
|
||
)
|
||
line_list = suggested_result.get("line_list", [])
|
||
if not line_list:
|
||
text = suggested_result.get("recognized_text", "")
|
||
line_list = [line for line in text.split("\n") if line.strip()]
|
||
ocr_parser = OrderOCRParser()
|
||
return ocr_parser.parse(line_list, ocr_confidence)
|
||
|
||
def _merge_results(self, rule_result: dict, llm_result: dict) -> dict:
|
||
"""合并规则提取结果和 LLM 解析结果。
|
||
|
||
以 LLM 结果为基础，规则提取的手机号优先使用，地址取较长者。
|
||
|
||
参数：
|
||
rule_result: 规则提取结果。
|
||
llm_result: LLM 解析结果。
|
||
|
||
返回：
|
||
合并后的订单字典。
|
||
"""
|
||
merged = dict(llm_result)
|
||
if rule_result.get("customer_mobile"):
|
||
merged["customer_mobile"] = rule_result["customer_mobile"]
|
||
rule_addr = rule_result.get("customer_address", "")
|
||
llm_addr = merged.get("customer_address", "")
|
||
if rule_addr and len(rule_addr) > len(llm_addr or ""):
|
||
merged["customer_address"] = rule_addr
|
||
return merged
|
||
|
||
def _build_fallback_result(self, rule_result: dict) -> dict:
|
||
"""当 LLM 不可用时，仅基于规则提取结果构建订单数据。
|
||
|
||
参数：
|
||
rule_result: 规则提取结果。
|
||
|
||
返回：
|
||
包含基础订单字段的字典，产品明细可能不完整。
|
||
"""
|
||
items: list[dict] = []
|
||
if rule_result.get("_raw_quantity"):
|
||
items.append({
|
||
"product_name": "",
|
||
"specification": "",
|
||
"unit": rule_result.get("_raw_unit", ""),
|
||
"quantity": rule_result["_raw_quantity"],
|
||
"sale_price": rule_result.get("_raw_price", 0),
|
||
"cost_price": None,
|
||
})
|
||
return {
|
||
"customer_name": rule_result.get("customer_name"),
|
||
"customer_mobile": rule_result.get("customer_mobile"),
|
||
"customer_address": rule_result.get("customer_address"),
|
||
"order_source": None,
|
||
"delivery_type": None,
|
||
"factory_id": None,
|
||
"remark": None,
|
||
"items": items,
|
||
}
|
||
|
||
def _get_product_groups(self, session: Session | None) -> list[dict]:
|
||
"""获取产品库数据，用于构建 LLM 提示词和模糊匹配。
|
||
|
||
参数：
|
||
session: 数据库会话，为 None 时返回空列表。
|
||
|
||
返回：
|
||
产品分组列表。
|
||
"""
|
||
if session is None:
|
||
return []
|
||
try:
|
||
from backend.app.services.product_service import product_service
|
||
result = product_service.list_products({}, session)
|
||
return result.get("list", [])
|
||
except Exception:
|
||
return []
|
||
|
||
def _match_products(self, items: list[dict],
|
||
product_groups: list[dict],
|
||
session: Session | None) -> list[dict]:
|
||
"""对解析出的产品明细进行库内模糊匹配。
|
||
|
||
使用 SequenceMatcher 计算名称（权重 0.6）和规格（权重 0.4）的相似度，
|
||
返回每个输入产品的前 3 个候选匹配（阈值 0.3）。
|
||
|
||
参数：
|
||
items: 解析出的产品明细列表。
|
||
product_groups: 产品库分组数据。
|
||
session: 数据库会话。
|
||
|
||
返回：
|
||
匹配结果列表，每个元素包含 input_name、input_specification、candidates。
|
||
"""
|
||
flat_products: list[dict] = []
|
||
for group in product_groups:
|
||
for spec in group.get("specifications", []):
|
||
flat_products.append({
|
||
"product_id": spec["product_id"],
|
||
"product_name": group["product_name"],
|
||
"specification": spec["specification"],
|
||
"unit": spec["unit"],
|
||
"sale_price": spec.get("sale_price", 0),
|
||
"cost_price": spec.get("cost_price", 0),
|
||
})
|
||
matches: list[dict] = []
|
||
for item in items:
|
||
input_name = item.get("product_name", "")
|
||
input_spec = item.get("specification", "")
|
||
if not input_name:
|
||
continue
|
||
scored: list[tuple] = []
|
||
for prod in flat_products:
|
||
name_score = SequenceMatcher(None, input_name, prod["product_name"]).ratio()
|
||
spec_score = SequenceMatcher(None, input_spec, prod["specification"]).ratio() if input_spec else 0
|
||
combined = name_score * 0.6 + spec_score * 0.4
|
||
scored.append((prod, combined))
|
||
scored.sort(key=lambda x: x[1], reverse=True)
|
||
matches.append({
|
||
"input_name": input_name,
|
||
"input_specification": input_spec,
|
||
"candidates": [
|
||
{
|
||
"product_id": prod["product_id"],
|
||
"product_name": prod["product_name"],
|
||
"specification": prod["specification"],
|
||
"unit": prod["unit"],
|
||
"sale_price": prod["sale_price"],
|
||
"cost_price": prod["cost_price"],
|
||
"match_score": round(score, 2),
|
||
}
|
||
for prod, score in scored[:3] if score > 0.3
|
||
],
|
||
})
|
||
return matches
|
||
|
||
def _validate_parsed_order(self, order: dict) -> list[str]:
|
||
"""校验解析后的订单数据完整性，生成警告列表。
|
||
|
||
检查项：客户姓名、手机号、产品明细、产品名称、数量。
|
||
|
||
参数：
|
||
order: 解析后的订单字典。
|
||
|
||
返回：
|
||
警告信息列表，无警告时返回空列表。
|
||
"""
|
||
warnings: list[str] = []
|
||
if not order.get("customer_name"):
|
||
warnings.append("未识别到客户姓名")
|
||
if not order.get("customer_mobile"):
|
||
warnings.append("未识别到客户手机号")
|
||
if not order.get("items"):
|
||
warnings.append("未识别到产品明细")
|
||
for item in order.get("items", []):
|
||
if not item.get("product_name"):
|
||
warnings.append("存在产品名称为空的明细行")
|
||
if not item.get("quantity"):
|
||
warnings.append(f"产品 {item.get('product_name', '?')} 未识别到数量")
|
||
return warnings
|
||
|
||
def _calc_confidence(self, order: dict, warnings: list[str],
|
||
ocr_confidence: float | None = None) -> float:
|
||
"""计算综合置信度分数。
|
||
|
||
计算规则：以 OCR 置信度（默认 1.0）为基础，每个警告扣 0.1，
|
||
每个缺少单价的产品扣 0.05，最低不低于 0.1。
|
||
|
||
参数：
|
||
order: 解析后的订单字典。
|
||
warnings: 警告列表。
|
||
ocr_confidence: OCR 置信度，可选。
|
||
|
||
返回：
|
||
综合置信度（0.1 ~ 1.0）。
|
||
"""
|
||
base = 1.0
|
||
if ocr_confidence is not None:
|
||
base = ocr_confidence
|
||
base -= len(warnings) * 0.1
|
||
if order.get("items"):
|
||
for item in order["items"]:
|
||
if not item.get("sale_price"):
|
||
base -= 0.05
|
||
return round(max(base, 0.1), 2)
|
||
|
||
def _get_factory_options(self, session: Session | None) -> list[dict]:
|
||
"""获取工厂（供应商）下拉选项列表，供前端智能填单表单使用。
|
||
|
||
参数：
|
||
session: 数据库会话，为 None 时返回空列表。
|
||
|
||
返回：
|
||
包含 value（supplier_id）和 label（supplier_name）的字典列表。
|
||
"""
|
||
if session is None:
|
||
return []
|
||
try:
|
||
from backend.app.services.supplier_service import supplier_service
|
||
result = supplier_service.list_suppliers(
|
||
session, {"supplier_type": "factory"}
|
||
)
|
||
return [
|
||
{"value": item["supplier_id"], "label": item["supplier_name"]}
|
||
for item in result.get("list", [])
|
||
]
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
ai_service = AIService()
|