# ============================================================ # 矩阵 101（方向 JJ）：图片/附件元数据提取器黄金测试 # 覆盖：文件名模式提取（订单/物料/日期线索）、元数据候选（路径/大小/类型）、 # 低置信走既有门禁出 P2 卡不写世界、无 OCR 不编造（空候选/数字守卫）、 # VLM 注册覆盖后行为切换、网关接线（/api/multimodal/extract + ingest）。 # ============================================================ from __future__ import annotations import pytest from fastapi.testclient import TestClient from server.agent_core import harness from server.aps_domain.multimodal import ( IMAGE_META_MAX_CONFIDENCE, ExtractionCandidate, ImageMetaExtractor, candidates_to_batches, default_registry, ingest_candidates, ) from server.aps_domain.workflow import execute_confirmed from server.auth.context import IdentityContext, bind_identity, reset_identity from server.state.seed import seed_world _PLANNER = IdentityContext(1001, "planner", "计划员", "tenant-multimodal-image", roles=("planner",)) _THRESHOLD = 0.7 # 默认确认阈值（stub image_meta 封顶 0.6 < 0.7） def _next_id(world): counters: dict[str, int] = {} def next_id(kind: str) -> int: table = "auditEvents" if kind == "audit" else kind + "s" if kind not in counters: rows = world.get(table, []) counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0) counters[kind] += 1 return counters[kind] return next_id class _WorldStore: def __init__(self, world, next_id_fn, *, tenant_uuid="tenant-multimodal-image", world_key="personal-1001"): self.data = world self._next_id_fn = next_id_fn self.tenant_uuid = tenant_uuid self.world_key = world_key def next_id(self, kind: str) -> int: return self._next_id_fn(kind) def save(self) -> None: pass def _domain_row_count(world) -> int: return (len(world.get("flexOrders") or []) + len(world.get("salesOrders") or []) + len(world.get("flexMaterials") or []) + len(world.get("materials") or [])) # ---------------- 文件名模式提取 ---------------- def test_image_meta_order_filename_pattern_extracts_order_no(): cands = ImageMetaExtractor().extract( {"filename": "订单-102285668.xlsx", "size": 4096}) c = cands[0] assert c.kind == "image_meta" and c.source == "STUB_IMAGE_META" assert c.complete is True assert c.value["orderNo"] == "102285668" assert c.value["attachmentKind"] == "order" assert c.value["requiresVlm"] is True assert c.fields["requiresVlm"] is True assert c.value["mime"] == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" assert c.value["size"] == 4096 # 0.2 基础 + 单号 0.1 + 附件类型 0.1 + 大小 0.1 = 0.5（封顶 0.6） assert c.confidence == pytest.approx(0.5) assert c.confidence <= IMAGE_META_MAX_CONFIDENCE def test_image_meta_material_bom_and_month_date(): cands = ImageMetaExtractor().extract( {"filename": "物料BOM-2026-08.xlsx", "size": 8192}) c = cands[0] assert c.value["materialCode"] == "BOM-2026-08" assert c.value["date"] == "2026-08" assert c.value["attachmentKind"] == "material" assert c.value["requiresVlm"] is True # 0.2 + 物料 0.1 + 日期 0.1 + 类型 0.1 + 大小 0.1 = 0.6（封顶） assert c.confidence == pytest.approx(IMAGE_META_MAX_CONFIDENCE) def test_image_meta_compact_date_and_order_digit_guard(): e = ImageMetaExtractor() img = e.extract({"filename": "IMG_20260802.png", "size": 12345})[0] assert img.value["date"] == "2026-08-02" assert img.value["attachmentKind"] == "image" assert img.value["orderNo"] == "" # 8 位数字紧邻订单前缀：只认单号，不重复识别为日期（不编造） so = e.extract({"filename": "SO-20260801.csv"})[0] assert so.value["orderNo"] == "20260801" assert so.value["date"] == "" # 9 位订单数字：紧凑日期必须整段 8 位，不得从订单号中截取 long = e.extract({"filename": "订单-102285668.xlsx"})[0] assert long.value["orderNo"] == "102285668" assert long.value["date"] == "" # ---------------- 元数据候选 ---------------- def test_image_meta_metadata_candidate_from_local_path(tmp_path): p = tmp_path / "IMG_20260802.png" p.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 64) cands = ImageMetaExtractor().extract({"path": str(p)}) assert len(cands) == 1 c = cands[0] assert c.value["filename"] == "IMG_20260802.png" assert c.value["path"] == str(p) assert c.value["size"] == p.stat().st_size assert c.value["mime"] == "image/png" assert c.value["date"] == "2026-08-02" assert c.value["requiresVlm"] is True assert c.fields["size"] == p.stat().st_size # 0.2 + 日期 0.1 + 类型 0.1 + 真实大小 0.1 = 0.5 assert c.confidence == pytest.approx(0.5) # ---------------- 无 OCR 不编造 ---------------- def test_image_meta_no_clues_returns_empty_no_fabrication(): e = ImageMetaExtractor() assert e.extract({"filename": "random_photo.png"}) == [] assert e.extract({"filename": "资料归档.txt"}) == [] assert e.extract("IMG_unknown.png") == [] with pytest.raises(ValueError): e.extract({}) with pytest.raises(ValueError): e.extract(12345) # ---------------- 既有门禁接入：低置信 → P2 确认卡 ---------------- def test_image_meta_candidates_to_batches_attachments(): cands = ImageMetaExtractor().extract( {"filename": "订单-102285668.xlsx", "size": 4096}) batches = candidates_to_batches(cands) assert len(batches) == 1 b = batches[0] assert b["kind"] == "attachments" and b["sheet"] == "image-meta" row = b["okRows"][0] assert row["filename"] == "订单-102285668.xlsx" assert row["orderNo"] == "102285668" assert row["requiresVlm"] is True def test_image_meta_low_confidence_stages_p2_card_without_world_write(): world = seed_world() next_id = _next_id(world) token = bind_identity(_PLANNER) try: cands = ImageMetaExtractor().extract( {"filename": "订单-102285668.xlsx", "size": 4096}) assert cands[0].confidence < _THRESHOLD # stub 封顶 0.6，必然低置信 before = _domain_row_count(world) result = ingest_candidates(world, next_id, cands, session_id="mm-img", threshold=_THRESHOLD) finally: reset_identity(token) assert result["status"] == "staged" confirm_id = result["confirmId"] assert result["block"]["type"] == "confirm-card" assert "低置信" in result["block"]["props"]["title"] # 确认前绝不写世界状态（域表 + 直通审计均无变化） assert _domain_row_count(world) == before assert not any(e["action"] == "multimodal.apply" for e in world["auditEvents"]) stages = [e for e in world["auditEvents"] if e["action"] == "multimodal.stage"] assert len(stages) == 1 assert stages[0]["rationale"]["confirmId"] == confirm_id assert stages[0]["rationale"]["action"] == "import.commit" assert stages[0]["result"] == "PENDING" # stage 审计携带候选明细（确认卡冻结参数存于审批记录，UI 块不含 params） assert stages[0]["rationale"]["candidates"][0]["kind"] == "image_meta" assert stages[0]["rationale"]["candidates"][0]["source"] == "STUB_IMAGE_META" def test_image_meta_confirm_records_audit_without_fabricating_domain_rows(): world = seed_world() next_id = _next_id(world) token = bind_identity(_PLANNER) try: cands = ImageMetaExtractor().extract( {"filename": "订单-102285668.xlsx", "size": 4096}) staged = ingest_candidates(world, next_id, cands, session_id="mm", threshold=_THRESHOLD) cid = staged["confirmId"] store = _WorldStore(world, next_id) before = _domain_row_count(world) message = execute_confirmed(store, cid, True, actor="planner") # attachments 批次本轮无域表写入分支（真实 VLM 入库接线为外部扩展）： # 确认只落审计与快照，绝不落虚假业务行——元数据保留在确认卡/审计中。 assert "文件导入完成" in message assert "共 0 条" in message assert _domain_row_count(world) == before assert any(e["action"] == "import.commit" for e in world["auditEvents"]) commit = next(e for e in world["auditEvents"] if e["action"] == "import.commit") assert commit["beforeSnapshot"] is not None # 驳回：不写世界 + 留痕 cands2 = ImageMetaExtractor().extract({"filename": "IMG_20260802.png"}) staged2 = ingest_candidates(world, next_id, cands2, session_id="mm", threshold=_THRESHOLD) message2 = execute_confirmed(store, staged2["confirmId"], False, actor="planner") assert "已驳回" in message2 assert _domain_row_count(world) == before assert any(e["action"] == "import.commit.reject" for e in world["auditEvents"]) finally: reset_identity(token) # ---------------- VLM 注册覆盖：行为切换 ---------------- def test_image_meta_vlm_registration_overrides_stub(): registry = default_registry() assert "image_meta" in registry.list_kinds() stub = registry.extract("image_meta", {"filename": "订单-102285668.xlsx"}) assert stub[0].source == "STUB_IMAGE_META" assert stub[0].confidence <= IMAGE_META_MAX_CONFIDENCE assert stub[0].value["requiresVlm"] is True class FakeVlmExtractor: """真实 VLM 外部模型服务的替身：实现 Extractor 协议，kind 与 stub 同名。""" kind = "image_meta" source = "VLM_EXT" def extract(self, raw): name = str((raw or {}).get("filename") or "") return [ExtractionCandidate( kind="image_meta", value={"filename": name, "ocrText": "真实 OCR 输出（外部 VLM）", "orderNo": "VLM-001", "requiresVlm": False}, confidence=0.92, source="VLM_EXT", fields={"ocr": {"engine": "fake-vlm"}}, complete=True)] with pytest.raises(ValueError): # 重复注册默认拒绝 registry.register(FakeVlmExtractor()) registry.register(FakeVlmExtractor(), replace=True) # 覆盖 stub → 真实提取器 real = registry.extract("image_meta", {"filename": "IMG_1.png"}) assert len(real) == 1 assert real[0].source == "VLM_EXT" assert real[0].confidence == pytest.approx(0.92) assert real[0].value["requiresVlm"] is False assert real[0].value["ocrText"].startswith("真实 OCR") # ---------------- 网关接线：/api/multimodal/* ---------------- class _AuditStore: def __init__(self, tenant_uuid: str = "tenant-multimodal-image"): self.tenant_uuid = tenant_uuid self.world_key = "personal-1001" self.data: dict = seed_world() self._counter = 0 def next_id(self, _kind: str) -> int: self._counter += 1 return self._counter def save(self) -> None: pass class _CheckpointStore: def create(self, *_args, **_kwargs) -> dict: return {"pairId": "mm-img-pair"} def get(self, pair_id: str) -> dict | None: return {"pairId": pair_id} class _ProjectStore: def active_world_key(self) -> str: return "default" def require_active_write(self) -> None: pass @pytest.fixture def client_factory(monkeypatch): import server.gateway.app as gateway_module import server.state.projects as projects_module from tests.auth_provider import install_test_auth install_test_auth(monkeypatch, "tenant-multimodal-image") def factory(): store = _AuditStore() monkeypatch.setattr(gateway_module, "get_store", lambda: store) monkeypatch.setattr(gateway_module, "get_checkpoints", lambda: _CheckpointStore()) monkeypatch.setattr(projects_module, "get_project_store", lambda: _ProjectStore()) client = TestClient(gateway_module.create_app()) login = client.post("/api/auth/login", json={"username": "planner", "password": "test"}) assert login.status_code == 200, login.text return client, store return factory def test_image_meta_api_extract_and_ingest_staged(client_factory): client, store = client_factory() resp = client.post("/api/multimodal/extract", json={ "kind": "image_meta", "filename": "订单-102285668.xlsx"}) assert resp.status_code == 200, resp.text body = resp.json() assert body["kind"] == "image_meta" assert "image_meta" in body["registeredKinds"] cands = body["candidates"] assert len(cands) == 1 assert cands[0]["kind"] == "image_meta" assert cands[0]["source"] == "STUB_IMAGE_META" assert cands[0]["confidence"] <= IMAGE_META_MAX_CONFIDENCE assert cands[0]["value"]["orderNo"] == "102285668" assert cands[0]["value"]["requiresVlm"] is True before = len(store.data["flexOrders"]) + len(store.data["salesOrders"]) ingest = client.post("/api/multimodal/ingest", json={ "candidates": cands, "threshold": 0.7}) assert ingest.status_code == 200, ingest.text staged = ingest.json() assert staged["status"] == "staged" assert len(store.data["flexOrders"]) + len(store.data["salesOrders"]) == before assert any(e["action"] == "multimodal.stage" for e in store.data["auditEvents"]) assert not any(e["action"] == "multimodal.apply" for e in store.data["auditEvents"])