diff --git a/backend/app/api/ai.py b/backend/app/api/ai.py index c1bdeeb..6f8e268 100644 --- a/backend/app/api/ai.py +++ b/backend/app/api/ai.py @@ -3,7 +3,7 @@ from sqlalchemy.orm import Session from backend.app.api.deps import require_permissions, require_roles from backend.app.db import get_db_session -from backend.app.schemas.ai import CorrectRecognizeResultRequest, RecognizeImageRequest +from backend.app.schemas.ai import CorrectRecognizeResultRequest, ParseOrderRequest, RecognizeImageRequest from backend.app.schemas.common import success_payload from backend.app.services.ai_service import ai_service @@ -29,3 +29,13 @@ def correct_recognize_result( _permission_user: dict = Depends(require_permissions("ai:correct")), ) -> dict: return success_payload(ai_service.correct_result(log_id, payload.model_dump(), session)) + + +@router.post("/parse-order") +def parse_order( + payload: ParseOrderRequest, + session: Session = Depends(get_db_session), + current_user: dict = Depends(require_roles("salesman", "admin")), + _permission_user: dict = Depends(require_permissions("ai:parse-order")), +) -> dict: + return success_payload(ai_service.parse_order(payload.model_dump(), session)) diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index dc8c17d..4439adf 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -15,6 +15,7 @@ from backend.app.services.reminder_service import reminder_service from backend.app.services.report_service import report_service from backend.app.services.supplier_service import supplier_service from backend.app.services.system_service import system_service +from backend.app.services.ai_service import ai_service # 统一依赖入口，后续如果切换到真实容器或数据库实现，只需要改这里。 @@ -94,3 +95,7 @@ def get_audit_service(): def get_system_service(): return system_service + + +def get_ai_service(): + return ai_service diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 78dbb31..c57844e 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -57,6 +57,12 @@ class Settings(BaseSettings): wechat_template_logistics_timeout: str = Field(default="", alias="WECHAT_TEMPLATE_LOGISTICS_TIMEOUT") wechat_template_arrears: str = Field(default="", alias="WECHAT_TEMPLATE_ARREARS") wechat_template_inactive_customer: str = Field(default="", alias="WECHAT_TEMPLATE_INACTIVE_CUSTOMER") + llm_parse_enabled: bool = Field(default=True, alias="LLM_PARSE_ENABLED") + llm_parse_model: str = Field(default="qwen-plus", alias="LLM_PARSE_MODEL") + llm_parse_api_url: str = Field( + default="https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions", + alias="LLM_PARSE_API_URL", + ) @property def cors_origins(self) -> list[str]: diff --git a/backend/app/schemas/ai.py b/backend/app/schemas/ai.py index e397caf..04b356d 100644 --- a/backend/app/schemas/ai.py +++ b/backend/app/schemas/ai.py @@ -1,4 +1,4 @@ -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator class RecognizeImageRequest(BaseModel): @@ -9,3 +9,17 @@ class RecognizeImageRequest(BaseModel): class CorrectRecognizeResultRequest(BaseModel): corrected_result: dict + + +class ParseOrderRequest(BaseModel): + input_type: str = Field(pattern="^(text|image)$") + text: str | None = None + image_url: str | None = None + + @model_validator(mode="after") + def validate_input(self): + if self.input_type == "text" and not (self.text or "").strip(): + raise ValueError("文本模式下 text 不能为空") + if self.input_type == "image" and not (self.image_url or "").strip(): + raise ValueError("图片模式下 image_url 不能为空") + return self diff --git a/backend/app/services/ai_service.py b/backend/app/services/ai_service.py index d189601..5964523 100644 --- a/backend/app/services/ai_service.py +++ b/backend/app/services/ai_service.py @@ -1,5 +1,7 @@ import json +import re from decimal import Decimal +from difflib import SequenceMatcher from urllib import error, parse, request from sqlalchemy.exc import SQLAlchemyError @@ -12,6 +14,345 @@ 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: + 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: + 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: + 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 行列表做订单版面分析。""" + + 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: + 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: + 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: + 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]: + 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: + """调用 dashscope qwen-plus 进行订单文本结构化解析。""" + + 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: + 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: + 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: + 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: + 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: provider_name = "base" @@ -292,5 +633,253 @@ class AIService: except json.JSONDecodeError: return value + # ------------------------------------------------------------------ + # 智能填单：核心解析入口 + # ------------------------------------------------------------------ + + def parse_order(self, payload: dict, session: Session | None = None) -> dict: + 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.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: + 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: + 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: + 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]: + if session is not None: + try: + from backend.app.services.product_service import product_service + result = product_service.list_products({}, session) + return result.get("list", []) + except Exception: + pass + return [ + { + "product_name": "演示产品A", + "product_id": 2001, + "specifications": [ + {"product_id": 2001, "specification": "10kg", "unit": "吨", "sale_price": 100, "cost_price": 60} + ], + }, + { + "product_name": "演示产品B", + "product_id": 2002, + "specifications": [ + {"product_id": 2002, "specification": "20kg", "unit": "吨", "sale_price": 180, "cost_price": 120} + ], + }, + ] + + def _match_products(self, items: list[dict], + product_groups: list[dict], + session: Session | None) -> list[dict]: + 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]: + 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: + 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]: + if session is not None: + 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: + pass + return [{"value": 1001, "label": "演示工厂A"}] + ai_service = AIService() diff --git a/backend/app/services/demo_store.py b/backend/app/services/demo_store.py index 25823db..3b9fde7 100644 --- a/backend/app/services/demo_store.py +++ b/backend/app/services/demo_store.py @@ -14,7 +14,7 @@ class DemoStore: "role_code": "salesman", "token": "demo-sales-token", "menus": [{"menu_name": "我的订单", "menu_path": "/orders"}], - "permissions": ["order:create", "order:list", "order:submit"], + "permissions": ["order:create", "order:list", "order:submit", "ai:parse-order"], }, ("admin01", "admin"): { "user_id": 99, diff --git a/docs/smart-order-entry-design.md b/docs/smart-order-entry-design.md new file mode 100644 index 0000000..449c437 --- /dev/null +++ b/docs/smart-order-entry-design.md @@ -0,0 +1,1788 @@ +# 智能填单功能方案设计 + +## 一、功能概述 + +### 1.1 背景 + +当前业务员在创建订单时需要手动填写 4 个信息分区（客户信息、订单信息、物流费用、产品明细），字段多且重复录入频繁。业务员日常接收订单的方式主要是微信聊天、电话沟通、拍照转发，信息散落在各种非结构化场景中。 + +### 1.2 目标 + +提供两种快速录入入口，让业务员像顺丰快递员一样"粘贴一下就能出单"： + +- **文本粘贴解析**：把微信聊天记录、电话记录等文本直接粘贴，系统自动识别客户信息和产品明细 +- **图片智能识别**：拍照或从相册选取订单图片（手写单、名片、聊天截图），系统 OCR 识别后自动填充 + +### 1.3 与现有系统的关系 + +本功能是现有下单流程（`OrderFormPage.vue` → `POST /api/orders`）的前置增强，不改变已有订单创建逻辑。解析结果最终仍通过现有的 `createOrder` 接口提交。 + +### 1.3 核心体验流程 + +``` +业务员打开新建订单页面 + ↓ +看到"智能填单"入口（文本粘贴 / 图片上传 两个 Tab） + ↓ +输入内容 → 点击"智能解析" + ↓ +系统返回结构化预览（逐字段展示，标注置信度） + ↓ +业务员核对、修正（可选） + ↓ +点击"确认填入" → 自动填充到表单对应字段 + ↓ +业务员补充缺失字段 → 提交订单 +``` + +--- + +## 二、系统架构 + +### 2.1 现有基础设施盘点 + +| 能力 | 现有位置 | 当前状态 | +|------|---------|---------| +| OSS 图片上传 | `file_service.py` / `storage_service.py` | 完整可用，前端有上传凭证流程 | +| OCR 识别 | `ai_service.py` → `AliyunOCRProvider` | 可提取文本行列表和置信度，但当前仅做通用文本提取，无订单场景专项处理 | +| 阿里云 AI 配置 | `config.py`（endpoint/key/region） | 已配置，可扩展 | +| 订单数据模型 | `schemas/orders.py` → `CreateOrderRequest` | 完整，新功能的输出需对齐此结构 | +| 产品库 | `product_repository.py` | 前端 `fetchProductOptions()` 可获取完整列表 | +| 客户库 | `customer_repository.py` | 前端 `fetchCustomerOptions()` 可获取完整列表 | + +### 2.2 现有 OCR 能力与不足 + +当前 `AliyunOCRProvider` 返回的数据结构： + +```python +raw_result = { + "provider": "aliyun_ocr", + "model": "default", + "image_url": "...", + "payload": { ... }, # 阿里云原始响应 +} +suggested_result = { + "recognized_text": "整段文本", + "line_list": ["第1行", "第2行", ...], # 逐行文本 + "customer_name": "第一行文本截取", +} +confidence = 0.92 # OCR 置信度 +``` + +**现有不足**：`line_list` 和 `confidence` 这两个关键数据在现有流程中被浪费了。`line_list` 包含了文本的行级结构信息（顺序、分行），可以用来做版面分析；`confidence` 反映了图片清晰度和识别质量，应该参与到最终解析置信度的计算中。 + +### 2.3 新增组件 + +``` +┌─────────────────────────────────────────────────────┐ +│ 前端 OrderFormPage.vue │ +│ ┌──────────────┐ ┌──────────────┐ │ +│ │ 文本粘贴 Tab │ │ 图片上传 Tab │ │ +│ └──────┬───────┘ └──────┬───────┘ │ +│ │ │ │ +│ ↓ ↓ │ +│ ┌─────────────────────────────┐ │ +│ │ POST /api/ai/parse-order │ │ +│ └─────────────┬───────────────┘ │ +│ │ │ +│ ┌─────────────↓───────────────┐ │ +│ │ 确认填充弹窗/预览层 │ │ +│ └─────────────┬───────────────┘ │ +│ │ 确认 │ +│ ┌─────────────↓───────────────┐ │ +│ │ 表单自动填充 │ │ +│ └─────────────────────────────┘ │ +└─────────────────────────────────────────────────────┘ + │ + │ HTTP + ↓ +┌─────────────────────────────────────────────────────┐ +│ 后端 API 层 │ +│ ┌─────────────────────────────┐ │ +│ │ POST /api/ai/parse-order │ │ +│ │ （api/ai.py 新增端点） │ │ +│ └─────────────┬───────────────┘ │ +│ │ │ +│ ┌─────────────↓───────────────┐ │ +│ │ AIService.parse_order() │ │ +│ │ （ai_service.py 新增方法）│ │ +│ └──┬──────────────┬───────────┘ │ +│ │ │ │ +│ │ 图片模式 │ 文本模式 │ +│ ↓ ↓ │ +│ ┌────────────┐ ┌────────┐ │ +│ │ OrderOCR │ │ │ │ +│ │ Parser │ │ │ │ +│ │ (版面分析) │ │ │ │ +│ │ ↓ │ │ │ │ +│ │ AliyunOCR │ │ │ │ +│ │ Provider │ │ │ │ +│ └─────┬──────┘ │ │ │ +│ │ │ │ │ +│ ↓ ↓ │ │ +│ ┌──────────────────────┐ │ │ +│ │ 正则提取层 │ │ │ +│ │ RuleExtractor │ │ │ +│ │ + OCR 预填充合并 │ │ │ +│ └──────────┬───────────┘ │ │ +│ ↓ │ │ +│ ┌──────────────────────┐ │ │ +│ │ LLM 结构化解析 │ │ │ +│ │ LLMOrderParser │ │ │ +│ │ (文本 prompt / │ │ │ +│ │ 图片 prompt) │ │ │ +│ └──────────┬───────────┘ │ │ +│ ↓ │ │ +│ ┌──────────────────────┐ │ │ +│ │ 合并 + 产品匹配 │ │ │ +│ │ + 综合置信度计算 │ │ │ +│ │ (含 OCR 置信度) │ │ │ +│ └──────────────────────┘ │ │ +│ │ │ +│ ┌─────────────────────────┐ │ +│ │ 统一结构化输出 │ │ +│ │ 对齐 CreateOrderRequest │ │ +│ └─────────────────────────┘ │ +└─────────────────────────────────────────────────────┘ +``` + +--- + +## 三、接口设计 + +### 3.1 智能解析接口 + +**POST** `/api/ai/parse-order` + +**权限**：`salesman` / `admin` 角色，`ai:recognize` 权限 + +**请求体**： + +```json +{ + "input_type": "text", + "text": "张三 13812345678 杭州市西湖区文一路100号\n要A产品 规格25kg 200吨 单价85\n备注：周五前要送到", + "image_url": null +} +``` + +```json +{ + "input_type": "image", + "text": null, + "image_url": "https://oss-cn-hangzhou.aliyuncs.com/bucket/uploads/xxx.jpg" +} +``` + +**响应体**： + +```json +{ + "code": 0, + "data": { + "parsed_order": { + "customer_name": "张三", + "customer_mobile": "13812345678", + "customer_address": "杭州市西湖区文一路100号", + "order_source": null, + "delivery_type": null, + "factory_id": null, + "remark": "周五前要送到", + "items": [ + { + "product_name": "A产品", + "specification": "25kg", + "unit": "吨", + "quantity": 200, + "sale_price": 85, + "cost_price": null + } + ], + "freight_total": null, + "tax_total": null, + "other_fee_total": null, + "commission_amount": null + }, + "raw_text": "张三 13812345678 杭州市西湖区文一路100号\n要A产品...", + "confidence": 0.88, + "ocr_confidence": 0.92, + "parse_source": "hybrid", + "product_matches": [ + { + "input_name": "A产品", + "candidates": [ + { "product_id": 101, "product_name": "精品A型", "match_score": 0.82 }, + { "product_id": 205, "product_name": "A系列标准品", "match_score": 0.75 } + ] + } + ], + "warnings": ["未识别到订单来源", "未识别到配送方式"] + } +} +``` + +> `ocr_confidence` 仅在图片输入模式下返回，文本模式下为 `null`。`confidence` 是综合置信度（OCR 置信度 × 字段完整度折扣），`ocr_confidence` 是纯 OCR 识别置信度，两者独立返回便于前端分别展示。 + +### 3.2 Schema 定义 + +**schemas/ai.py 新增**： + +```python +class ParseOrderRequest(BaseModel): + input_type: str = Field(pattern="^(text|image)$") + text: str | None = None + image_url: str | None = None + + @model_validator(mode="after") + def validate_input(self): + if self.input_type == "text" and not (self.text or "").strip(): + raise ValueError("文本模式下 text 不能为空") + if self.input_type == "image" and not (self.image_url or "").strip(): + raise ValueError("图片模式下 image_url 不能为空") + return self + + +class ParsedOrderItem(BaseModel): + product_name: str + specification: str | None = None + unit: str | None = None + quantity: float | None = None + sale_price: float | None = None + cost_price: float | None = None + remark: str | None = None + + +class ProductMatch(BaseModel): + input_name: str + candidates: list[dict] + + +class ParseOrderResponse(BaseModel): + parsed_order: dict + raw_text: str + confidence: float + ocr_confidence: float | None = None # 仅图片模式返回 + parse_source: str # "rule" | "llm" | "hybrid" + product_matches: list[ProductMatch] + warnings: list[str] +``` + +--- + +## 四、核心解析逻辑 + +### 4.0 文本预处理（TextPreprocessor） + +业务员粘贴的原始文本往往包含大量噪声——微信聊天的时间戳、用户名、表情符号、[图片] 标记等。这些噪声会干扰 LLM 解析，必须在进入解析管线之前清洗。 + +```python +import re + + +class TextPreprocessor: + """清洗粘贴文本中的噪声，保留有效订单信息。""" + + # 微信聊天时间戳：2024-03-15 14:30:22 或 [14:30] + 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: + """清洗原始文本，返回适合解析的干净文本。""" + if not text or not text.strip(): + return "" + + result = text + + # 1. 去除时间戳 + result = self.TIMESTAMP_RE.sub('', result) + result = self.BRACKET_TIME_RE.sub('', result) + + # 2. 去除无意义标记 + for marker in self.NOISE_MARKERS: + result = result.replace(marker, '') + + # 3. 去除微信用户名前缀（只在行首出现的短名称 + 冒号） + result = self.USERNAME_PREFIX_RE.sub('', result) + + # 4. 合并连续空行为单行 + result = re.sub(r'\n{3,}', '\n\n', result) + + # 5. 去除首尾空白 + result = result.strip() + + return result +``` + +此预处理器在所有解析路径之前调用，文本模式和图片模式都经过清洗后再进入后续流程。 + +### 4.1 解析流程 + +图片输入和文本输入走不同的前处理路径，OCR 不再是简单的"图片转文字"，而是作为独立的预处理层参与整个解析过程： + +``` +输入 + │ + ├─ input_type = "image" + │ ↓ + │ ┌─────────────────────────────────┐ + │ │ OCR 预处理层（OrderOCRParser） │ + │ │ │ + │ │ 1. 调用 AliyunOCRProvider │ + │ │ 获取 line_list + confidence │ + │ │ │ + │ │ 2. 版面分析：按行扫描，识别 │ + │ │ - 标题行 / 表头行 │ + │ │ - 地址区域 │ + │ │ - 表格数据区域（产品明细） │ + │ │ │ + │ │ 3. 区域级正则提取： │ + │ │ - 标题区域 → 客户姓名 │ + │ │ - 地址区域 → 客户地址 │ + │ │ - 表格区域 → 产品行解析 │ + │ │ │ + │ │ 4. 输出： │ + │ │ - raw_text（合并后文本） │ + │ │ - ocr_confidence（OCR置信度）│ + │ │ - layout_zones（版面分区） │ + │ └───────────┬─────────────────────┘ + │ ↓ + │ OCR 结构化中间结果 + │ { raw_text, ocr_confidence, layout_zones, + │ pre_filled: {customer_name, customer_mobile, ...} } + │ + ├─ input_type = "text" + │ ↓ + │ raw_text = text + │ ocr_confidence = null（不需要 OCR） + │ layout_zones = null + │ pre_filled = {}（无预填充） + │ + ↓ +正则提取层（毫秒级，确定性字段） + │ - 手机号：/1[3-9]\d{9}/ + │ - 数量+单位：/(\d+\.?\d*)\s*(吨|件|箱|包|个|米|kg)/ + │ - 价格：/单价\s*[:：]?\s*(\d+\.?\d*)/ + │ - 地址关键词提取 + │ - 合并 OCR 预填充的字段（如有） + │ + ↓ +规则提取结果（部分字段已有值） + │ + ↓ +LLM 结构化解析（处理模糊/缺失部分） + │ 调用 dashscope qwen-plus + │ prompt 中附带： + │ - raw_text + │ - 正则已提取的字段 + │ - OCR 版面分区提示（如 "第3-5行为产品表格区域"） + │ - 产品库名称列表 + │ + ↓ +合并结果：正则确定的字段 > LLM 推断的字段 + │ + ↓ +产品库模糊匹配（前端做） + │ + ↓ +置信度计算：综合 OCR 置信度 + 字段完整度 + 警告数 + │ + ↓ +返回结构化订单数据 +``` + +### 4.2 OCR 预处理层（OrderOCRParser） + +这是本次设计的核心新增。现有的 `AliyunOCRProvider` 是通用 OCR，不认识订单的结构。`OrderOCRParser` 在其上层做订单场景的版面分析，把散乱的文本行转化为有结构的中间数据。 + +```python +import re + + +class OrderOCRParser: + """基于 OCR 行列表做订单版面分析，提取结构化中间数据。 + + 核心思路：OCR 返回的 line_list 是有序的文本行列表。 + 不同行在图片中的空间位置隐含了语义分区—— + 第一行通常是标题/客户名，中间有地址行，下方可能是表格。 + 即使没有坐标信息，行的顺序和内容关键词也足以做分区判断。 + """ + + # --- 关键词规则：用于识别每行的语义角色 --- + + 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 行列表中提取结构化中间数据。 + + Returns: + { + "raw_text": "合并后的完整文本", + "ocr_confidence": 0.92, + "layout_zones": { + "header_lines": [0, 1], # 标题/客户区域的行索引 + "address_lines": [2], # 地址区域的行索引 + "table_lines": [3, 4, 5], # 产品表格区域的行索引 + "other_lines": [6], # 其他行 + }, + "pre_filled": { + "customer_name": "张三", + "customer_mobile": "13812345678", + "customer_address": "杭州市西湖区文一路100号", + }, + "table_rows": [ + {"raw": "A产品 25kg 200吨 85元", "fields": {...}}, + ], + } + """ + 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: + """将每行文本分类到不同的版面区域。""" + zones = { + "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: + """从各区域中提取具体字段。""" + result = {} + # 标题区域：提取客户姓名（通常是第一行非空文本） + 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 = [] + for i in zones.get("address_lines", []): + addr_parts.append(lines[i].strip()) + 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]: + """尝试解析表格区域的每一行为产品明细。""" + rows = [] + 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 = {"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 +``` + +### 4.2 正则提取层 + +```python +import re + +class RuleExtractor: + """基于规则的字段提取器，处理确定性高的字段。""" + + PHONE_RE = re.compile(r'1[3-9]\d{9}') + # 数量+单位：支持 200吨、200 吨、200.5kg 等 + QTY_UNIT_RE = re.compile(r'(\d+\.?\d*)\s*(吨|件|箱|包|个|米|kg|KG|kg|吨|公斤|斤|卷|组|套|台|条|根|片|块)') + # 单价 + PRICE_RE = re.compile(r'(?:单价|价格|报价)\s*[:：]?\s*(\d+\.?\d*)') + # 地址关键词 + ADDRESS_KEYWORDS = ['省', '市', '区', '县', '镇', '路', '街', '号', '楼', '室', '栋', '单元', '村', '大厦', '广场'] + + def extract(self, text: str) -> dict: + result = {} + + # 手机号 + phones = self.PHONE_RE.findall(text) + if phones: + result['customer_mobile'] = phones[0] + + # 数量（取第一个匹配作为整体数量，产品级数量在 LLM 层处理） + 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)) + + # 地址（基于关键词的启发式提取） + result['customer_address'] = self._extract_address(text) + + return result + + def _extract_address(self, text: str) -> str | 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 +``` + +### 4.3 LLM 结构化解析 + +LLM 的 prompt 需要根据输入来源做区分——图片模式下，OCR 已经做了版面分析和部分字段提取，prompt 应该把这些中间结果告诉 LLM，让它专注于处理 OCR 搞不定的部分，而不是从零开始解析。 + +```python +import json +from urllib import request as urllib_request + + +class LLMOrderParser: + """调用 dashscope qwen-plus 进行订单文本结构化解析。""" + + # 文本输入模式的 prompt：从零解析 + 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_names}""" + + # 图片/OCR 模式的 prompt：基于 OCR 已提取的信息补充解析 + 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_names}""" + + def parse(self, text: str, product_names: list[str], + api_key: str, api_url: str, + ocr_context: dict | None = None) -> dict: + """解析订单文本。 + + Args: + ocr_context: 图片模式下传入，包含 OCR 预处理层的中间结果。 + 文本模式下为 None。 + """ + 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_names='\n'.join(f'- {name}' for name in product_names[:50]), + ) + else: + system_prompt = self.TEXT_SYSTEM_PROMPT.format( + product_names='\n'.join(f'- {name}' for name in product_names[:50]), + ) + + 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，兼容 markdown 代码块包裹。""" + 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: + """安全调用 LLM，失败时返回 None（由上层退化为纯规则模式）。""" + try: + return self.parse(text, product_groups, api_key, api_url, ocr_context) + except Exception: + # LLM 调用失败（网络超时、API 报错、JSON 解析失败等） + # 不向上抛异常，返回 None 让调用方用规则提取结果兜底 + return None +``` + +### 4.4 AIService 新增方法 + +```python +# ai_service.py 新增 + +def parse_order(self, payload: dict, session: Session | None = None) -> dict: + 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 结构化解析（传入 OCR 上下文，失败时退化为纯规则模式） + product_groups = self._get_product_groups(session) + llm_parser = LLMOrderParser() + llm_result = llm_parser.safe_parse( + raw_text, + product_groups, + settings.aliyun_ai_access_key_id, + settings.llm_parse_api_url, + ocr_context=ocr_context, + ) + + # 5. 合并：LLM 成功时走完整路径，失败时退化为规则模式 + 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: + existing_customer = self.customer_repository.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 并用 OrderOCRParser 做版面分析。 + + Returns: + OrderOCRParser.parse() 的完整输出。 + """ + provider = self._build_provider() + raw_result, suggested_result, ocr_confidence = provider.recognize( + image_url, "order_parse", 0 + ) + # 从 suggested_result 中获取行列表 + line_list = suggested_result.get("line_list", []) + if not line_list: + # 如果 OCR 没返回行列表，按换行符拆分 + 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 推断的字段。""" + 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 _get_product_names(self, session: Session | None) -> list[str]: + if session is not None: + try: + products = self.product_repository.list_products(session) + return [p.product_name for p in products] + except Exception: + pass + return ["演示产品A", "演示产品B"] + +def _match_products(self, items: list[dict], + product_names: list[str], + session: Session | None) -> list[dict]: + """对解析出的每个产品名称做模糊匹配。""" + from difflib import SequenceMatcher + matches = [] + for item in items: + input_name = item.get("product_name", "") + if not input_name: + continue + scored = [ + (name, SequenceMatcher(None, input_name, name).ratio()) + for name in product_names + ] + scored.sort(key=lambda x: x[1], reverse=True) + matches.append({ + "input_name": input_name, + "candidates": [ + {"product_name": name, "match_score": round(score, 2)} + for name, score in scored[:3] if score > 0.3 + ], + }) + return matches + +def _validate_parsed_order(self, order: dict) -> list[str]: + warnings = [] + 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 置信度；纯文本输入时 OCR 置信度为 null， + 只基于字段完整度计算。 + """ + base = 1.0 + # OCR 置信度作为基础（图片模式） + 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 _build_fallback_result(self, rule_result: dict) -> dict: + """LLM 不可用时，用规则提取结果构建最小可用的解析结果。""" + items = [] + 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]: + """获取按名称分组的产品列表（含规格），复用 ProductService 的分组逻辑。""" + if session is not None: + try: + from backend.app.services.product_service import product_service + result = product_service.list_products({}, session) + return result.get("list", []) + except Exception: + pass + return [ + { + "product_name": "演示产品A", + "product_id": 2001, + "specifications": [ + {"product_id": 2001, "specification": "10kg", "unit": "吨", "sale_price": 100, "cost_price": 60} + ], + }, + { + "product_name": "演示产品B", + "product_id": 2002, + "specifications": [ + {"product_id": 2002, "specification": "20kg", "unit": "吨", "sale_price": 180, "cost_price": 120} + ], + }, + ] + +def _get_factory_options(self, session: Session | None) -> list[dict]: + """获取可用工厂列表，供前端在解析结果中选择。""" + if session is not None: + 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: + pass + return [{"value": 1001, "label": "演示工厂A"}] + +--- + +## 五、前端设计 + +### 5.1 交互设计 + +在 `OrderFormPage.vue` 表单顶部插入"智能填单"卡片，视觉上是一个独立面板，解析成功后收起并展示已解析字段概要。 + +``` +┌─────────────────────────────────────────────────┐ +│ 🤖 智能填单 │ +│ │ +│ [ 粘贴文本 ] [ 拍照/上传图片 ] │ +│ │ +│ ┌─────────────────────────────────────────┐ │ +│ │ │ │ +│ │ 请粘贴订单相关文本... │ │ +│ │ （微信聊天记录、电话记录等） │ │ +│ │ │ │ +│ └─────────────────────────────────────────┘ │ +│ │ +│ [ 🪄 智能解析 ] │ +│ │ +└─────────────────────────────────────────────────┘ +``` + +解析完成后变为确认预览模式： + +``` +┌─────────────────────────────────────────────────┐ +│ 🤖 智能填单 - 解析结果 置信度 88% │ +│ │ +│ ✅ 客户姓名: 张三 [编辑] │ +│ ✅ 客户手机: 13812345678 [编辑] │ +│ ✅ 客户地址: 杭州市西湖区文一路100号 [编辑] │ +│ ⚠️ 订单来源: 未识别 [填写] │ +│ │ +│ 产品明细： │ +│ ┌─────────────────────────────────────────┐ │ +│ │ A产品 / 25kg / 200吨 / 单价85 │ │ +│ │ 匹配建议: → 精品A型 (82%) │ │ +│ └─────────────────────────────────────────┘ │ +│ │ +│ ⚠️ 提示: 未识别到配送方式、未识别到工厂 │ +│ │ +│ [ 重新解析 ] [ 确认填入表单 ] │ +└─────────────────────────────────────────────────┘ +``` + +### 5.2 前端状态管理与数据流 + +"确认填入"是整个功能中最关键的环节——解析结果要正确映射到现有表单的状态结构中。以下是详细的数据流设计。 + +**新增响应式状态**（在 `