336 lines
14 KiB
Python
336 lines
14 KiB
Python
# ============================================================
|
||
# 矩阵 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"])
|