843 lines
39 KiB
Python
843 lines
39 KiB
Python
# ============================================================
|
||
# 多模态输入管线框架（moduleId: domain-multimodal, 可重生 ✅）
|
||
# 矩阵 101（方向 Y）：语音/图片/文件多模态与低置信确认
|
||
# - Extractor 协议：raw（文本/字节/文件名）→ ExtractionCandidate 列表
|
||
# （结构化候选：{kind, value, confidence(0-1), source, fields}）
|
||
# - ExtractorRegistry：按 kind 注册/查询；未注册提取器显式报错（不编造）
|
||
# - 内置 stub 提取器：file_import（复用 importers 的 detect/preview/validate
|
||
# 思路输出候选+置信度）、text_order（「订单 单号 x 数量 n」模板抽取）、
|
||
# image_meta（图片/附件元数据：文件名/大小/类型/路径 → 订单/物料/日期线索
|
||
# 候选；无 OCR/VLM 能力时置信 ≤0.6 且标注 requiresVlm，走低置信确认门禁）
|
||
# - 低置信确认门禁（ingest_candidates / stage_multimodal_confirm）：
|
||
# 候选置信度 < 阈值（默认 0.7，可配）→ harness P2 确认卡，确认前绝不
|
||
# 写世界状态；置信度 ≥ 阈值且字段完整 → 直通既有导入路径。
|
||
# - 审计：multimodal.extract / multimodal.stage / multimodal.apply
|
||
# （确认卡批准/驳回由 harness 通用通道写 import.commit(.reject)，stage
|
||
# 审计携带 confirmId 关联——复用既有 P2 权力矩阵，见 stage_multimodal_confirm）
|
||
# 真实 ASR/VLM/OCR：实现 Extractor 协议并经 ExtractorRegistry.register()
|
||
# （replace=True 可覆盖同名 stub）接入；未注册时抛 ExtractorNotRegisteredError
|
||
# （fail closed，不编造识别结果）。真实 OCR/VLM 属外部模型服务扩展，本模块 stub
|
||
# 只做文件名/元数据级结构化候选，绝不编造图片识别内容。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import os
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Protocol, runtime_checkable
|
||
|
||
World = dict[str, Any]
|
||
|
||
# 低置信确认阈值（环境变量可覆盖；ingest 亦可按请求覆盖）
|
||
DEFAULT_CONFIDENCE_THRESHOLD = 0.7
|
||
|
||
# 文本订单 stub 模板：口语/短文本订单语句的结构化槽位（确定性模板，非 LLM）
|
||
_TEXT_PATTERNS: dict[str, re.Pattern[str]] = {
|
||
"orderNo": re.compile(
|
||
r"(?:订单号|单号|工单号|订单编号|订单号是|单号是)\s*[:：]?\s*"
|
||
r"([A-Za-z0-9][A-Za-z0-9\-_/]{1,31})"),
|
||
"quantity": re.compile(r"(?:数量|quantity|个数)\s*[:：]?\s*(\d+(?:\.\d+)?)"),
|
||
"deliveryDate": re.compile(
|
||
r"(?:交期|交货日期|交货期|计划交期|dueDate|截止)\s*[:：]?\s*"
|
||
r"(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?)"),
|
||
"customerName": re.compile(
|
||
r"(?:客户名称|客户名|客户|customer)\s*[:：]?\s*([\u4e00-\u9fa5A-Za-z0-9]{1,20})"),
|
||
"productCode": re.compile(
|
||
r"(?:产品编码|产品料号|物料编码|料号|productCode|产品)\s*[:：]?\s*"
|
||
r"([A-Za-z0-9][A-Za-z0-9\-_]{1,31})"),
|
||
}
|
||
_TEXT_REQUIRED_FIELDS = ("orderNo", "quantity")
|
||
_TEXT_OPTIONAL_FIELDS = ("deliveryDate", "customerName", "productCode")
|
||
|
||
# 图片/附件元数据 stub 模板（矩阵 101 扩展：方向 JJ）
|
||
# 真实 OCR/VLM 需外部模型服务；stub 仅从文件名/大小/类型/路径提取结构化线索。
|
||
# 置信度封顶 0.6（IMAGE_META_MAX_CONFIDENCE）且标注 requiresVlm=True——低于
|
||
# 默认 0.7 阈值，必然走既有低置信确认门禁，确认前不写世界状态。
|
||
IMAGE_META_MAX_CONFIDENCE = 0.6
|
||
_IMAGE_ORDER_PATTERNS: tuple[re.Pattern[str], ...] = (
|
||
# 中文前缀：订单/订单号/工单/工单号 + 可选分隔 + 单号（数字或字母数字）
|
||
re.compile(
|
||
r"(?:订单|订单号|工单|工单号|订单编号|order|orderno)"
|
||
r"\s*[-_:：]?\s*([A-Za-z0-9][A-Za-z0-9\-_/]{1,31})"),
|
||
# 标准前缀：SO/PO/WO/MO + 可选分隔 + 至少 3 位数字（前后非字母数字，避免把
|
||
# report/consolidated 等单词里的子串误判为单号）
|
||
re.compile(
|
||
r"(?<![A-Za-z0-9])(?:so|po|wo|mo)[-_]?\s*(\d{3,})", re.IGNORECASE),
|
||
)
|
||
_IMAGE_MATERIAL_PATTERN = re.compile(
|
||
r"(?:物料|物料编码|物料代码|料号|(?<![a-z])bom(?![a-z]))"
|
||
r"\s*[-_:：]?\s*([A-Za-z0-9][A-Za-z0-9\-_/]{1,31})", re.IGNORECASE)
|
||
# 日期线索（优先级：完整日期 > 年月 > 紧凑 8 位；与订单号捕获区间重叠时视为
|
||
# 订单号而非日期——避免把订单数字误报为日期，守住「不编造」）
|
||
_IMAGE_DATE_FULL = re.compile(r"(\d{4})[-_/.年](\d{1,2})[-_/.月](\d{1,2})日?")
|
||
_IMAGE_DATE_MONTH = re.compile(r"(\d{4})[-_/.年](\d{1,2})月?")
|
||
_IMAGE_DATE_COMPACT = re.compile(r"(?<![0-9])(\d{4})(\d{2})(\d{2})(?![0-9])")
|
||
_IMAGE_ATTACH_KIND_RULES: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||
("order", re.compile(
|
||
r"订单|工单|(?<![a-z])(?:order|orderno|so|po|wo|mo)(?![a-z])", re.IGNORECASE)),
|
||
("material", re.compile(
|
||
r"物料|料号|(?<![a-z])(?:bom|material)(?![a-z])", re.IGNORECASE)),
|
||
("drawing", re.compile(
|
||
r"图纸|工艺|(?<![a-z])(?:drawing|drg|dwg)(?![a-z])", re.IGNORECASE)),
|
||
("quote", re.compile(r"报价|(?<![a-z])quote(?![a-z])", re.IGNORECASE)),
|
||
("image", re.compile(
|
||
r"照片|图片|扫描|(?<![a-z])(?:scan|photo|img|image)(?![a-z])", re.IGNORECASE)),
|
||
)
|
||
_MIME_BY_EXT: dict[str, str] = {
|
||
".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
|
||
".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp",
|
||
".svg": "image/svg+xml", ".tif": "image/tiff", ".tiff": "image/tiff",
|
||
".heic": "image/heic", ".heif": "image/heif",
|
||
".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
".xls": "application/vnd.ms-excel", ".csv": "text/csv", ".pdf": "application/pdf",
|
||
".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||
".doc": "application/msword",
|
||
}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ExtractionCandidate:
|
||
"""一条结构化提取候选（矩阵 101：ASR/VLM/文件解析的输出契约）。
|
||
|
||
kind: 候选类型（file_import / text_order / 扩展注册的 kind 如 asr、vlm）
|
||
value: 结构化载荷（file_import → importers preview 结果；text_order → 订单字段）
|
||
confidence: 置信度 0-1（< 阈值时须经确认才能写世界状态）
|
||
source: 来源标识（STUB_FILE / STUB_TEXT / 扩展来源名）
|
||
fields: 字段级置信明细（每字段原始值/置信度/完整性）
|
||
complete: 字段完整性（False 时拒绝写入，不编造缺失内容）
|
||
"""
|
||
|
||
kind: str
|
||
value: Any
|
||
confidence: float
|
||
source: str
|
||
fields: dict[str, Any] = field(default_factory=dict)
|
||
complete: bool = True
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"kind": self.kind,
|
||
"value": self.value,
|
||
"confidence": round(float(self.confidence), 4),
|
||
"source": self.source,
|
||
"fields": self.fields,
|
||
"complete": bool(self.complete),
|
||
}
|
||
|
||
|
||
class ExtractorNotRegisteredError(KeyError):
|
||
"""请求的多模态提取器未注册：显式报错，绝不编造真实 ASR/VLM 结果。"""
|
||
|
||
def __init__(self, kind: str, *, registered: tuple[str, ...] = ()) -> None:
|
||
self.kind = kind
|
||
self.registered = tuple(sorted(registered))
|
||
super().__init__(
|
||
f"未注册的多模态提取器：{kind!r}（已注册：{', '.join(self.registered) or '无'}）。"
|
||
"真实 ASR/VLM 需实现 Extractor 协议并经 ExtractorRegistry.register() 接入；"
|
||
"stub 框架不编造未接入服务的识别结果。"
|
||
)
|
||
|
||
|
||
@runtime_checkable
|
||
class Extractor(Protocol):
|
||
"""多模态提取器协议：raw（文本/字节/文件名）→ 结构化候选列表。"""
|
||
|
||
kind: str
|
||
source: str
|
||
|
||
def extract(self, raw: Any) -> list[ExtractionCandidate]:
|
||
...
|
||
|
||
|
||
class ExtractorRegistry:
|
||
"""按 kind 注册/查询提取器；未注册 kind 显式报错。"""
|
||
|
||
def __init__(self) -> None:
|
||
self._extractors: dict[str, Extractor] = {}
|
||
|
||
def register(self, extractor: Extractor, *, replace: bool = False) -> None:
|
||
"""注册一个提取器（真实 ASR/VLM 扩展入口）。
|
||
|
||
replace=True 可覆盖内置 stub（例如接入真实 ASR 后替换同名 stub）。
|
||
"""
|
||
kind = getattr(extractor, "kind", None)
|
||
if not kind or not isinstance(kind, str) or not kind.strip():
|
||
raise TypeError("extractor 必须暴露非空字符串 kind")
|
||
if kind in self._extractors and not replace:
|
||
raise ValueError(f"提取器已注册：{kind!r}（replace=True 可覆盖）")
|
||
self._extractors[kind.strip()] = extractor
|
||
|
||
def get(self, kind: str) -> Extractor:
|
||
try:
|
||
return self._extractors[kind]
|
||
except KeyError:
|
||
raise ExtractorNotRegisteredError(kind, registered=tuple(self._extractors)) from None
|
||
|
||
def list_kinds(self) -> list[str]:
|
||
return sorted(self._extractors)
|
||
|
||
def extract(self, kind: str, raw: Any) -> list[ExtractionCandidate]:
|
||
return self.get(kind).extract(raw)
|
||
|
||
|
||
class DxfDrawingExtractor:
|
||
"""工程图纸 DXF 提取器：标题栏/位置号/技术说明 → 结构化候选。
|
||
|
||
解析由 server.aps_domain.dxf_drawing 承担（块属性标题栏 + 中文标签兜底 +
|
||
文件名线索），候选置信度 = 物料候选置信度（标题栏图号/名称交叉校验后
|
||
通常 0.9；仅文件名线索时 0.5）。完整 = 物料编码与名称均已识别；
|
||
材质/数量/工序等图纸未承载字段保持 missing（不编造）。
|
||
写入正式主数据走 /api/dxf/stage → master.*.upsert 的 P2 确认卡，
|
||
不经 importers 批次（candidates_to_batches 对本 kind 显式 fail closed）。
|
||
"""
|
||
|
||
kind = "dxf_drawing"
|
||
source = "DXF_PARSE"
|
||
|
||
def extract(self, raw: Any) -> list[ExtractionCandidate]:
|
||
from server.aps_domain.dxf_drawing import (
|
||
drawing_to_master_candidates,
|
||
parse_dxf,
|
||
parse_dxf_bytes,
|
||
)
|
||
if isinstance(raw, dict) and raw.get("data") is not None:
|
||
parsed = parse_dxf_bytes(str(raw.get("filename") or "upload.dxf"),
|
||
bytes(raw["data"]))
|
||
elif isinstance(raw, dict) and raw.get("path"):
|
||
parsed = parse_dxf(str(raw["path"]))
|
||
elif isinstance(raw, str) and raw.strip():
|
||
parsed = parse_dxf(raw.strip())
|
||
else:
|
||
raise ValueError("dxf_drawing 提取器需要 {filename, data(bytes)}"
|
||
" 或 {path} 或路径字符串输入")
|
||
cands = drawing_to_master_candidates(parsed)
|
||
material = cands["material"]
|
||
tb_fields = (parsed.get("titleBlock") or {}).get("fields") or {}
|
||
value: dict[str, Any] = {
|
||
"drawingId": parsed["drawingId"],
|
||
"filename": parsed["filename"],
|
||
"dxfVersion": parsed["dxfVersion"],
|
||
"material": material,
|
||
"positionCount": len(parsed.get("positionNumbers") or []),
|
||
"noteCount": len(parsed.get("technicalNotes") or []),
|
||
"warnings": parsed.get("warnings") or [],
|
||
"candidates": cands,
|
||
}
|
||
complete = not material.get("missingFields")
|
||
fields = {
|
||
"titleBlock": {k: {"value": v.get("value"),
|
||
"confidence": v.get("confidence"),
|
||
"source": v.get("source")}
|
||
for k, v in tb_fields.items()},
|
||
"positionCount": value["positionCount"],
|
||
"bomComplete": bool(cands["bom"].get("complete")),
|
||
"routingComplete": bool(cands["routing"].get("complete")),
|
||
"missing": material.get("missingFields") or [],
|
||
"complete": complete,
|
||
"requiresConfirm": True,
|
||
}
|
||
return [ExtractionCandidate(
|
||
kind="dxf_drawing", value=value,
|
||
confidence=float(material.get("confidence") or 0.0),
|
||
source="DXF_PARSE", fields=fields, complete=complete,
|
||
)]
|
||
|
||
|
||
def default_registry(*, world: World | None = None) -> ExtractorRegistry:
|
||
"""内置 stub 注册表（file_import + text_order + image_meta + dxf_drawing）。
|
||
|
||
world：file_import 校验/识别用世界（None 时按当前加载的世界状态取）。
|
||
真实 ASR/VLM/OCR 扩展：default_registry().register(MyExtractor(), replace=True)。
|
||
"""
|
||
registry = ExtractorRegistry()
|
||
registry.register(FileImportExtractor(world=world))
|
||
registry.register(TextOrderExtractor())
|
||
registry.register(ImageMetaExtractor())
|
||
registry.register(DxfDrawingExtractor())
|
||
return registry
|
||
|
||
|
||
# ---------------- stub 提取器 ----------------
|
||
|
||
def _live_world() -> World:
|
||
from server.state.store import get_store
|
||
return get_store().data
|
||
|
||
|
||
def _coerce_file_raw(raw: Any) -> tuple[str, bytes]:
|
||
"""file_import 输入归一化：{filename, data(bytes)} 或 (filename, data)。"""
|
||
if isinstance(raw, tuple) and len(raw) == 2:
|
||
filename, data = raw
|
||
elif isinstance(raw, dict):
|
||
filename = str(raw.get("filename") or "upload")
|
||
data = raw.get("data")
|
||
else:
|
||
raise ValueError("file_import 提取器需要 {filename, data(bytes)} 或 (filename, data) 输入")
|
||
if not isinstance(data, (bytes, bytearray)):
|
||
raise TypeError("file_import 提取器需要字节内容 data")
|
||
return filename, bytes(data)
|
||
|
||
|
||
def _batch_confidence(batch: dict[str, Any]) -> float:
|
||
ok = int(batch.get("okCount") or 0)
|
||
err = int(batch.get("errorCount") or 0)
|
||
if ok + err <= 0:
|
||
return 0.0
|
||
return round(ok / (ok + err), 4)
|
||
|
||
|
||
class FileImportExtractor:
|
||
"""文件导入 stub：复用 importers 的 detect_kind/validate_batch/preview_file。
|
||
|
||
置信度 = 有效行占比（错误行不含置信）；字段完整 = canCommit 且有效行 > 0。
|
||
真实 OCR/VLM 表格识别可作为扩展注册（kind=file_import 覆盖或新 kind）。
|
||
"""
|
||
|
||
kind = "file_import"
|
||
source = "STUB_FILE"
|
||
|
||
def __init__(self, *, world: World | None = None) -> None:
|
||
self._world = world
|
||
|
||
def extract(self, raw: Any) -> list[ExtractionCandidate]:
|
||
from server.aps_domain.importers import preview_file
|
||
filename, data = _coerce_file_raw(raw)
|
||
world = self._world if self._world is not None else _live_world()
|
||
preview = preview_file(filename, data, world)
|
||
total = int(preview.get("totalOk") or 0) + int(preview.get("totalErrors") or 0)
|
||
if total <= 0:
|
||
confidence, complete = 0.0, False
|
||
else:
|
||
confidence = round(int(preview.get("totalOk") or 0) / total, 4)
|
||
complete = bool(preview.get("canCommit")) and int(preview.get("totalOk") or 0) > 0
|
||
batches: dict[str, Any] = {}
|
||
for b in preview.get("batches") or []:
|
||
kind = str(b.get("kind") or "unknown")
|
||
batches[kind] = {
|
||
"sheet": b.get("sheet"),
|
||
"okCount": int(b.get("okCount") or 0),
|
||
"errorCount": int(b.get("errorCount") or 0),
|
||
"confidence": _batch_confidence(b),
|
||
}
|
||
fields = {
|
||
"batches": batches,
|
||
"canCommit": bool(preview.get("canCommit")),
|
||
"totalOk": int(preview.get("totalOk") or 0),
|
||
"totalErrors": int(preview.get("totalErrors") or 0),
|
||
"complete": complete,
|
||
}
|
||
return [ExtractionCandidate(
|
||
kind="file_import", value=preview, confidence=confidence,
|
||
source="STUB_FILE", fields=fields, complete=complete,
|
||
)]
|
||
|
||
|
||
def _parse_quantity(v: Any) -> int | None:
|
||
try:
|
||
return int(float(str(v).strip()))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
class TextOrderExtractor:
|
||
"""文本订单 stub：「订单 单号 x 数量 n」模板抽取（确定性模板）。
|
||
|
||
必填槽位：订单号 + 数量（完整）；可选槽位：交期/客户/产品。
|
||
置信度：完整模板 0.6 起步，补齐可选槽位递增（交期 +0.2、客户 +0.1、
|
||
产品 +0.1，封顶 1.0）；槽位缺失（不完整）0.3——低置信必须经确认。
|
||
未命中任何模板槽位 → 返回空（不编造）。
|
||
"""
|
||
|
||
kind = "text_order"
|
||
source = "STUB_TEXT"
|
||
|
||
def extract(self, raw: Any) -> list[ExtractionCandidate]:
|
||
if isinstance(raw, str):
|
||
text = raw
|
||
elif isinstance(raw, dict):
|
||
text = str(raw.get("text") or "")
|
||
else:
|
||
text = str(raw or "")
|
||
text = text.strip()
|
||
if not text:
|
||
return []
|
||
matched: dict[str, str] = {}
|
||
for slot, pattern in _TEXT_PATTERNS.items():
|
||
m = pattern.search(text)
|
||
if m:
|
||
matched[slot] = m.group(1).strip()
|
||
# 宽松兜底：订单 <code>（ASCII 字母开头，避免把纯数字数量误识别为单号）
|
||
if "orderNo" not in matched:
|
||
m = re.search(r"订单\s+([A-Za-z][A-Za-z0-9\-_/]{1,31})", text)
|
||
if m:
|
||
matched["orderNo"] = m.group(1).strip()
|
||
if not matched:
|
||
return [] # 无任何模板命中：不编造
|
||
complete = all(f in matched for f in _TEXT_REQUIRED_FIELDS)
|
||
missing = [f for f in _TEXT_REQUIRED_FIELDS if f not in matched] + [
|
||
f for f in _TEXT_OPTIONAL_FIELDS if f not in matched]
|
||
if not complete:
|
||
confidence = 0.3
|
||
else:
|
||
confidence = 0.6 + (0.2 if "deliveryDate" in matched else 0.0) \
|
||
+ (0.1 if "customerName" in matched else 0.0) \
|
||
+ (0.1 if "productCode" in matched else 0.0)
|
||
qty = _parse_quantity(matched.get("quantity"))
|
||
value: dict[str, Any] = {
|
||
"orderNo": matched.get("orderNo", ""),
|
||
"quantity": qty if qty is not None else matched.get("quantity", ""),
|
||
"deliveryDate": matched.get("deliveryDate", ""),
|
||
"customerName": matched.get("customerName", ""),
|
||
"productCode": matched.get("productCode", ""),
|
||
}
|
||
fields = {
|
||
"matched": matched,
|
||
"required": list(_TEXT_REQUIRED_FIELDS),
|
||
"optional": list(_TEXT_OPTIONAL_FIELDS),
|
||
"missing": missing,
|
||
"complete": complete,
|
||
}
|
||
return [ExtractionCandidate(
|
||
kind="text_order", value=value, confidence=round(min(confidence, 1.0), 4),
|
||
source="STUB_TEXT", fields=fields, complete=complete,
|
||
)]
|
||
|
||
|
||
def _coerce_image_meta_raw(raw: Any) -> tuple[str, str | None, int | None, str | None]:
|
||
"""image_meta 输入归一化：raw → (name, path, size, mime)。
|
||
|
||
接受文件名/路径字符串或字典 {path|name|filename|size|mime|data_base64|...}。
|
||
无 path/name 时显式报错（不编造）；path 存在时 size 优先取真实文件大小。
|
||
"""
|
||
if isinstance(raw, str):
|
||
name, path = os.path.basename(raw), raw
|
||
size, mime = None, None
|
||
elif isinstance(raw, dict):
|
||
path = str(raw.get("path") or "").strip() or None
|
||
name = str(raw.get("name") or raw.get("filename") or "").strip()
|
||
if not name and path:
|
||
name = os.path.basename(path)
|
||
size = raw.get("size")
|
||
mime = str(raw.get("mime") or "").strip() or None
|
||
else:
|
||
raise ValueError(
|
||
"image_meta 提取器需要 文件名/路径 字符串或 {path|name|size|mime} 字典输入")
|
||
if not name:
|
||
raise ValueError("image_meta 提取器需要 path 或 name（本地图片/附件路径或文件名）")
|
||
if size is not None:
|
||
try:
|
||
size = int(size)
|
||
except (TypeError, ValueError):
|
||
size = None
|
||
if path and os.path.isfile(path):
|
||
try:
|
||
size = os.path.getsize(path) # 本地路径：取真实大小
|
||
except OSError:
|
||
pass
|
||
if not mime:
|
||
mime = _MIME_BY_EXT.get(os.path.splitext(name)[1].lower())
|
||
return name, path, size, mime
|
||
|
||
|
||
def _image_date_value(name: str, order_span: tuple[int, int] | None) -> str:
|
||
"""从文件名提取日期线索（完整日期 > 年月 > 紧凑 8 位）。
|
||
|
||
order_span：订单号捕获区间；日期若与订单号重叠（如 SO-20260801 的 8 位数字
|
||
已被识别为单号），不再重复识别为日期——防止把订单数字误报为日期（不编造）。
|
||
"""
|
||
def overlaps(m: re.Match[str]) -> bool:
|
||
return (order_span is not None
|
||
and m.start(1) < order_span[1] and order_span[0] < m.end(1))
|
||
m = _IMAGE_DATE_FULL.search(name)
|
||
if m and not overlaps(m):
|
||
return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}"
|
||
m = _IMAGE_DATE_MONTH.search(name)
|
||
if m and not overlaps(m):
|
||
return f"{m.group(1)}-{int(m.group(2)):02d}"
|
||
m = _IMAGE_DATE_COMPACT.search(name)
|
||
if m and not overlaps(m):
|
||
return f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
|
||
return ""
|
||
|
||
|
||
def _image_attach_kind(stem: str) -> str:
|
||
"""附件类型线索（确定性关键词，首个命中优先；无命中 → attachment）。"""
|
||
for kind, pattern in _IMAGE_ATTACH_KIND_RULES:
|
||
if pattern.search(stem):
|
||
return kind
|
||
return "attachment"
|
||
|
||
|
||
class ImageMetaExtractor:
|
||
"""图片/附件元数据 stub（矩阵 101 扩展：方向 JJ）。
|
||
|
||
输入：本地图片/附件路径或文件名（raw={path|name|size|mime|...}，或字符串）。
|
||
输出：结构化候选——从文件名提取订单/物料/日期线索（订单-102285668.xlsx →
|
||
orderNo=102285668；物料BOM-2026-08.xlsx → materialCode=BOM-2026-08 +
|
||
date=2026-08；IMG_20260802.png → date=2026-08-02），并携带大小/类型/路径等
|
||
元数据。
|
||
|
||
不编造：
|
||
- 无 OCR/VLM 能力时置信度封顶 0.6（IMAGE_META_MAX_CONFIDENCE）并标注
|
||
requiresVlm=True，必然低于默认 0.7 阈值 → 走既有低置信确认门禁；
|
||
- 文件名无任何订单/物料/日期线索 → 返回空列表（绝不编造识别内容）；
|
||
- 与订单号重叠的 8 位数字不被重复识别为日期（如 SO-20260801 只认单号）。
|
||
|
||
真实 OCR/VLM：实现 Extractor 协议并经 ExtractorRegistry.register()
|
||
（replace=True 覆盖本 stub）后，ExtractorRegistry.extract("image_meta", raw)
|
||
即切换为真实识别结果。
|
||
"""
|
||
|
||
kind = "image_meta"
|
||
source = "STUB_IMAGE_META"
|
||
|
||
def extract(self, raw: Any) -> list[ExtractionCandidate]:
|
||
name, path, size, mime = _coerce_image_meta_raw(raw)
|
||
stem = name.lower()
|
||
matched: dict[str, str] = {}
|
||
order_span: tuple[int, int] | None = None
|
||
for pattern in _IMAGE_ORDER_PATTERNS:
|
||
m = pattern.search(name)
|
||
if m:
|
||
matched["orderNo"] = m.group(1).strip()
|
||
order_span = (m.start(1), m.end(1))
|
||
break
|
||
m = _IMAGE_MATERIAL_PATTERN.search(name)
|
||
if m:
|
||
matched["materialCode"] = m.group(1).strip()
|
||
date = _image_date_value(name, order_span)
|
||
if date:
|
||
matched["date"] = date
|
||
if not matched:
|
||
return [] # 无任何文件名线索：不编造
|
||
attach_kind = _image_attach_kind(stem)
|
||
confidence = (0.2 + 0.1 * len(matched)
|
||
+ (0.1 if attach_kind != "attachment" else 0.0)
|
||
+ (0.1 if size is not None else 0.0))
|
||
confidence = round(min(confidence, IMAGE_META_MAX_CONFIDENCE), 4)
|
||
value: dict[str, Any] = {
|
||
"filename": name,
|
||
"path": path or "",
|
||
"size": size,
|
||
"mime": mime,
|
||
"attachmentKind": attach_kind,
|
||
"orderNo": matched.get("orderNo", ""),
|
||
"materialCode": matched.get("materialCode", ""),
|
||
"date": matched.get("date", ""),
|
||
"requiresVlm": True, # stub 无真实识别能力
|
||
"patterns": sorted(matched),
|
||
}
|
||
fields: dict[str, Any] = {
|
||
"matched": matched,
|
||
"missing": [f for f in ("orderNo", "materialCode", "date") if f not in matched],
|
||
"complete": True,
|
||
"requiresVlm": True,
|
||
"filename": name, "path": path or "", "mime": mime, "size": size,
|
||
}
|
||
return [ExtractionCandidate(
|
||
kind="image_meta", value=value, confidence=confidence,
|
||
source="STUB_IMAGE_META", fields=fields, complete=True,
|
||
)]
|
||
|
||
|
||
# ---------------- 入库归一化与确认门禁 ----------------
|
||
|
||
def _coerce_candidate(c: Any) -> ExtractionCandidate:
|
||
"""候选入参归一化：接受 ExtractionCandidate 或 JSON dict（extract 输出回传）。"""
|
||
if isinstance(c, ExtractionCandidate):
|
||
return c
|
||
if isinstance(c, dict):
|
||
try:
|
||
return ExtractionCandidate(
|
||
kind=str(c["kind"]),
|
||
value=c.get("value"),
|
||
confidence=float(c["confidence"]),
|
||
source=str(c.get("source") or "UNKNOWN"),
|
||
fields=c.get("fields") or {},
|
||
complete=bool(c.get("complete", False)),
|
||
)
|
||
except (KeyError, TypeError, ValueError) as exc:
|
||
raise ValueError(f"候选结构非法：{exc}") from exc
|
||
raise TypeError(f"候选必须是 ExtractionCandidate 或 dict，得到 {type(c).__name__}")
|
||
|
||
|
||
def _threshold(threshold: float | None) -> float:
|
||
if threshold is not None:
|
||
try:
|
||
threshold = float(threshold)
|
||
except (TypeError, ValueError) as exc:
|
||
raise ValueError("置信度阈值必须是数字") from exc
|
||
if not (0.0 <= threshold <= 1.0):
|
||
raise ValueError("置信度阈值必须在 0-1 之间")
|
||
return threshold
|
||
try:
|
||
return float(os.environ.get("APS_MULTIMODAL_CONFIDENCE_THRESHOLD")
|
||
or DEFAULT_CONFIDENCE_THRESHOLD)
|
||
except ValueError:
|
||
return DEFAULT_CONFIDENCE_THRESHOLD
|
||
|
||
|
||
def _text_order_batch(value: dict[str, Any]) -> dict[str, Any]:
|
||
"""文本订单候选 → 既有 orders 导入批次（缺交期按当日入柔性订单池，缺字段 fail closed）。"""
|
||
qty = _parse_quantity(value.get("quantity"))
|
||
if qty is None or qty <= 0:
|
||
raise ValueError("文本订单候选缺少有效数量，拒绝写入世界状态（fail closed，不编造）")
|
||
from server.timeutil import today0
|
||
row: dict[str, Any] = {
|
||
"orderNo": str(value.get("orderNo") or "").strip(),
|
||
"quantity": qty,
|
||
"deliveryDate": str(value.get("deliveryDate") or "").strip() or today0().strftime("%Y-%m-%d"),
|
||
"customerName": str(value.get("customerName") or "").strip(),
|
||
}
|
||
product_code = str(value.get("productCode") or "").strip()
|
||
if product_code:
|
||
row["productCode"] = product_code
|
||
return {"kind": "orders", "sheet": "multimodal-text", "okRows": [row]}
|
||
|
||
|
||
def candidates_to_batches(candidates: list[Any]) -> list[dict[str, Any]]:
|
||
"""多模态候选 → 既有 importers 批次（确认卡冻结参数 / 直通入库共用）。
|
||
|
||
支持可入库的 kind：file_import / text_order / image_meta。
|
||
image_meta → attachments 元数据登记批次（文件名/大小/类型/订单/物料/日期线索；
|
||
真实附件入库接线属外部 VLM 扩展，apply_import_commit 尚无 attachments 分支时
|
||
确认后不写域表——元数据完整保留在确认卡参数与审计中，绝不落虚假业务行）。
|
||
其余扩展 kind 拒绝入库（扩展提取器只输出识别结果，入库语义需显式接线）。
|
||
"""
|
||
batches: list[dict[str, Any]] = []
|
||
for c in candidates:
|
||
cand = _coerce_candidate(c)
|
||
if cand.kind == "file_import":
|
||
for b in (cand.value or {}).get("batches") or []:
|
||
if not b.get("okRows"):
|
||
continue
|
||
batches.append({
|
||
"kind": b.get("kind"), "sheet": b.get("sheet"),
|
||
"okRows": b.get("okRows") or [],
|
||
})
|
||
elif cand.kind == "text_order":
|
||
batches.append(_text_order_batch(cand.value or {}))
|
||
elif cand.kind == "image_meta":
|
||
v = cand.value or {}
|
||
batches.append({
|
||
"kind": "attachments",
|
||
"sheet": "image-meta",
|
||
"okRows": [{
|
||
"filename": v.get("filename") or "",
|
||
"path": v.get("path") or "",
|
||
"size": v.get("size"),
|
||
"mime": v.get("mime"),
|
||
"attachmentKind": v.get("attachmentKind"),
|
||
"orderNo": v.get("orderNo") or "",
|
||
"materialCode": v.get("materialCode") or "",
|
||
"date": v.get("date") or "",
|
||
"requiresVlm": bool(v.get("requiresVlm")),
|
||
}],
|
||
})
|
||
else:
|
||
hint = ("；dxf_drawing 候选请走 /api/dxf/stage 的 master.*.upsert P2 确认卡"
|
||
if cand.kind == "dxf_drawing" else "")
|
||
raise ValueError(
|
||
f"不支持直接入库的多模态候选 kind：{cand.kind!r}"
|
||
"（直通/确认只支持 file_import、text_order、image_meta）" + hint)
|
||
if not batches:
|
||
raise ValueError("候选归一化后没有可入库的批次（确认/直通均无可写内容）")
|
||
return batches
|
||
|
||
|
||
def _value_preview(c: ExtractionCandidate) -> str:
|
||
if c.kind == "file_import":
|
||
v = c.value or {}
|
||
return (f"文件 {v.get('filename', '')}：有效 {v.get('totalOk', 0)} 行"
|
||
f" / 错误 {v.get('totalErrors', 0)} 行")
|
||
if c.kind == "text_order":
|
||
v = c.value or {}
|
||
parts = [f"{k}={v.get(k) or '—'}" for k in ("orderNo", "quantity", "deliveryDate",
|
||
"customerName", "productCode")]
|
||
return "，".join(parts)
|
||
if c.kind == "image_meta":
|
||
v = c.value or {}
|
||
clues = "，".join(p for p in ("orderNo", "materialCode", "date") if v.get(p))
|
||
clues = clues or "无文件名线索（需 VLM 识别）"
|
||
return (f"附件 {v.get('filename') or '?'}（{v.get('attachmentKind') or '?'} · "
|
||
f"{v.get('mime') or '未知类型'} · {v.get('size') or '大小未知'} 字节）：{clues}")
|
||
return f"{c.kind} 候选（字段 {len(c.fields)} 项）"
|
||
|
||
|
||
def confirmation_for_multimodal(candidates: list[Any], threshold: float) -> tuple[str, list[str]]:
|
||
"""低置信确认卡内容：候选明细/置信度/来源（确认前绝不写世界状态）。"""
|
||
cands = [_coerce_candidate(c) for c in candidates]
|
||
title = "多模态候选确认 · 低置信入库"
|
||
lines = [f"置信度阈值 {threshold}：以下 {len(cands)} 条候选需人工确认后才会写入世界状态"]
|
||
for i, c in enumerate(cands, 1):
|
||
conf = f"{max(0.0, min(1.0, float(c.confidence))) * 100:.0f}%"
|
||
lines.append(f"{i}. [{c.kind} · 来源 {c.source} · 置信 {conf} · 完整 {c.complete}] {_value_preview(c)}")
|
||
lines.append("批准后经既有导入路径写入主干（P2，自动建档可回滚）；驳回不产生任何写入")
|
||
return title, lines
|
||
|
||
|
||
def stage_multimodal_confirm(world: World, next_id, candidates: list[Any], *,
|
||
session_id: str = "web", actor: str = "multimodal",
|
||
threshold: float | None = None) -> dict[str, Any]:
|
||
"""低置信候选 → P2 确认卡（复用既有 P2 动作 import.commit）。
|
||
|
||
说明：harness._POWER_MAP 是封闭表（server/agent_core/harness.py，本轮写范围外），
|
||
未登记动作默认按最高 P3（双人确认）处理，不符合 P2 单确认语义；因此低置信卡
|
||
复用既有 P2 动作 import.commit，确认执行走既有 execute_confirmed 分支（建档+审计+落盘）。
|
||
出卡仅登记意图与候选明细，确认前绝不写世界状态。
|
||
"""
|
||
from server.agent_core import harness
|
||
from server.agent_core.audit import write_audit
|
||
threshold = _threshold(threshold)
|
||
cands = [_coerce_candidate(c) for c in candidates]
|
||
if not cands:
|
||
raise ValueError("没有可确认的多模态候选")
|
||
low = [c for c in cands if not c.complete or float(c.confidence) < threshold]
|
||
if not low:
|
||
raise ValueError("没有低置信候选需要确认（置信度均 ≥ 阈值且字段完整）")
|
||
batches = candidates_to_batches(cands) # 冻结进确认卡参数（确认后原样执行）
|
||
title, lines = confirmation_for_multimodal(cands, threshold)
|
||
block = harness.stage_confirmation(
|
||
session_id or "web",
|
||
"import.commit",
|
||
{"filename": "multimodal-input", "batches": batches,
|
||
"multimodalMeta": {"threshold": threshold, "candidateCount": len(cands)}},
|
||
title=title, summary_lines=lines,
|
||
)
|
||
confirm_id = str(block.props["confirmId"])
|
||
write_audit(world, next_id, actor=actor, category="GATE", action="multimodal.stage",
|
||
target={"type": "MULTIMODAL", "id": confirm_id}, power="P2",
|
||
rationale={
|
||
"confirmId": confirm_id,
|
||
"threshold": threshold,
|
||
"action": "import.commit", # 复用既有 P2 权力矩阵动作
|
||
"candidates": [
|
||
{"kind": c.kind, "confidence": round(float(c.confidence), 4),
|
||
"source": c.source, "complete": bool(c.complete)} for c in cands
|
||
],
|
||
},
|
||
result="PENDING")
|
||
return {
|
||
"status": "staged",
|
||
"message": f"{title} 已进入 P2 确认队列（低置信候选，确认前未写世界状态）。",
|
||
"confirmId": confirm_id,
|
||
"block": block.model_dump(),
|
||
"threshold": threshold,
|
||
"candidates": [c.to_dict() for c in cands],
|
||
}
|
||
|
||
|
||
def apply_multimodal_candidates(world: World, next_id, candidates: list[Any], *,
|
||
actor: str = "multimodal",
|
||
threshold: float | None = None,
|
||
confirm_id: str | None = None) -> dict[str, Any]:
|
||
"""高置信/已确认候选 → 世界状态（直通既有导入路径 apply_import_commit）。
|
||
|
||
fail closed：任一候选不完整/置信度越界/文本候选缺数量 → 抛 ValueError，
|
||
不做任何写入（先全量校验再入库，无部分写入）。
|
||
"""
|
||
from server.agent_core.audit import write_audit
|
||
threshold = _threshold(threshold)
|
||
cands = [_coerce_candidate(c) for c in candidates]
|
||
incomplete = [c for c in cands if not c.complete]
|
||
if incomplete:
|
||
raise ValueError(
|
||
"候选字段不完整，拒绝写入世界状态："
|
||
+ ", ".join(f"{c.kind} 缺失 {c.fields.get('missing')}" if isinstance(c.fields, dict)
|
||
else c.kind for c in incomplete))
|
||
for c in cands:
|
||
conf = float(c.confidence)
|
||
if not (0.0 <= conf <= 1.0):
|
||
raise ValueError(f"候选置信度越界：{c.kind} {conf!r}")
|
||
batches = candidates_to_batches(cands) # 文本候选缺数量在此 fail closed
|
||
from server.aps_domain.importers import apply_import_commit
|
||
applied = apply_import_commit(world, next_id, batches)
|
||
write_audit(world, next_id, actor=actor, category="WORLD_WRITE", action="multimodal.apply",
|
||
target={"type": "MULTIMODAL", "id": ",".join(c.kind for c in cands) or "unknown"},
|
||
power="P2",
|
||
rationale={
|
||
"summary": applied["summary"],
|
||
"total": applied["total"],
|
||
"sources": [c.source for c in cands],
|
||
"threshold": threshold,
|
||
"confirmId": confirm_id,
|
||
})
|
||
return applied
|
||
|
||
|
||
def ingest_candidates(world: World, next_id, candidates: list[Any], *,
|
||
session_id: str = "web", actor: str = "multimodal",
|
||
threshold: float | None = None) -> dict[str, Any]:
|
||
"""多模态入库门禁（矩阵 101 核心）：
|
||
|
||
1. 字段不完整 → 显式拒绝（不编造、不出卡、不写世界状态）
|
||
2. 全部 置信度 ≥ 阈值 且字段完整 → 直通既有导入路径（apply_multimodal_candidates）
|
||
3. 任一候选低置信 → P2 确认卡（stage_multimodal_confirm；确认前绝不写世界状态，
|
||
批准后经 execute_confirmed 写，驳回不写）
|
||
"""
|
||
threshold = _threshold(threshold)
|
||
cands = [_coerce_candidate(c) for c in candidates]
|
||
if not cands:
|
||
raise ValueError("没有可入库的多模态候选")
|
||
incomplete = [c for c in cands if not c.complete]
|
||
if incomplete:
|
||
return {
|
||
"status": "incomplete",
|
||
"message": "候选字段不完整，拒绝写入（不编造缺失内容；请补充后重试）",
|
||
"threshold": threshold,
|
||
"candidates": [c.to_dict() for c in cands],
|
||
"missing": [
|
||
{"kind": c.kind,
|
||
"missingFields": c.fields.get("missing") if isinstance(c.fields, dict) else []}
|
||
for c in incomplete
|
||
],
|
||
}
|
||
low = [c for c in cands if float(c.confidence) < threshold]
|
||
if not low:
|
||
applied = apply_multimodal_candidates(world, next_id, cands, actor=actor,
|
||
threshold=threshold)
|
||
return {
|
||
"status": "applied",
|
||
"message": "多模态候选高置信直通导入完成（未出确认卡）。",
|
||
"summary": applied["summary"],
|
||
"total": applied["total"],
|
||
"threshold": threshold,
|
||
"candidates": [c.to_dict() for c in cands],
|
||
}
|
||
return stage_multimodal_confirm(world, next_id, cands, session_id=session_id,
|
||
actor=actor, threshold=threshold)
|
||
|
||
|
||
# ---------------- 网关辅助 ----------------
|
||
|
||
def build_raw(kind: str, *, text: str | None = None, filename: str | None = None,
|
||
data_base64: str | None = None) -> Any:
|
||
"""把网关请求字段归一化为提取器 raw 输入。
|
||
|
||
文件类：base64 → bytes（{filename, data}）；文本类：原样文本；
|
||
扩展 kind：透传字段字典（由已注册的扩展提取器自行解释）。
|
||
"""
|
||
if kind == "file_import":
|
||
if not data_base64:
|
||
raise ValueError("文件类提取需要 base64 文件内容（dataBase64）")
|
||
try:
|
||
data = base64.b64decode(data_base64, validate=True)
|
||
except (ValueError, TypeError) as exc:
|
||
raise ValueError("dataBase64 不是合法的 base64 内容") from exc
|
||
return {"filename": filename or "upload", "data": data}
|
||
if kind == "image_meta":
|
||
name = str(filename or "").strip()
|
||
if not name:
|
||
raise ValueError("图片/附件元数据提取需要 filename（本地路径或文件名）")
|
||
raw: dict[str, Any] = {"filename": name}
|
||
if data_base64:
|
||
raw["data_base64"] = data_base64 # 供真实 VLM 解码识别（stub 不使用）
|
||
return raw
|
||
if kind == "text_order":
|
||
text = str(text or "").strip()
|
||
if not text:
|
||
raise ValueError("文本类提取需要 text 内容")
|
||
return text
|
||
return {"kind": kind, "text": text, "filename": filename, "data_base64": data_base64}
|