diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py index e045685..54b27e5 100644 --- a/backend/contracts/prescreen.py +++ b/backend/contracts/prescreen.py @@ -90,7 +90,11 @@ class ErrorCode(StrEnum): # HTTP 语义沿用既有 translate()（backend/api/project_assistant.py:36-46）， -# 不另立一套。E-17 是**运行级结果标识**，不是 HTTP 错误（01 §5 末）。 +# 不另立一套。 +# E-17 FACT_CONFLICT_UNRESOLVED 有**两个面**，不要混为一谈： +# · 确认动作被拒（本映射）：422 —— 客户端必须先裁定冲突，属可立即纠正的请求问题 +# · 运行级结果（不在此映射内）：HTTP 200，冲突标识保存在 degradations/finding， +# "HTTP 200 仅表示结果可读取，不表示核查通过"（01 §5 末） S1_ERROR_HTTP: dict[ErrorCode, int] = { ErrorCode.NOT_FOUND: 404, ErrorCode.ACCESS_DENIED: 403, @@ -100,7 +104,7 @@ S1_ERROR_HTTP: dict[ErrorCode, int] = { ErrorCode.IDEMPOTENCY_CONFLICT: 409, ErrorCode.MATERIAL_PARSE_FAILED: 422, ErrorCode.IMPORT_BATCH_INVALID: 422, - ErrorCode.FACT_CONFLICT_UNRESOLVED: 200, + ErrorCode.FACT_CONFLICT_UNRESOLVED: 422, } # 403/422 不自动重试；409 需客户端改载荷或刷新。见 03 §5。 diff --git a/backend/services/assistant_materials.py b/backend/services/assistant_materials.py index fd70188..f1f8960 100644 --- a/backend/services/assistant_materials.py +++ b/backend/services/assistant_materials.py @@ -206,7 +206,15 @@ def demo_assets(project): return material(project["id"], project["id"] + "设备清单.csv", stream.getvalue().encode("utf-8-sig"), "demo-fixture") -def apply_candidate(project, candidate, actor, reason): +def apply_candidate(project, candidate, actor, reason, adopt=True): + """记录一个材料候选为证据。 + + adopt=True（默认，既有行为不变）：证据入 sources，并把值写进 + values / selected_sources / facts。 + adopt=False（新增，供 S1 事实确认使用）：**只**把证据记入 sources，不改 + values / selected_sources / facts。用于"候选与已确认值冲突、待人工裁定" + 的情形——证据已按 BR-22 留痕，但值不得被覆盖（BR-23）。 + """ value = scalar(candidate["field"], candidate.get("value")) evidence = dict(candidate, id=uid("ev"), project_id=project["id"], value=value, actor_id=actor, reason=reason, accepted_at=now()) money_fields = NUMERIC - {"lease.duplicate_count", "credit.overdue_count", "facility.term"} @@ -218,6 +226,8 @@ def apply_candidate(project, candidate, actor, reason): value = str(number(value) * SCALE[candidate["unit"]]) evidence["value"] = value project.setdefault("sources", {}).setdefault(candidate["field"], []).append(evidence) + if not adopt: + return evidence project.setdefault("values", {})[candidate["field"]] = value project.setdefault("selected_sources", {})[candidate["field"]] = evidence["id"] old = next((k for k, v in FIELDS.items() if v == candidate["field"]), candidate["field"]) diff --git a/backend/services/assistant_store.py b/backend/services/assistant_store.py index a271281..83673fe 100644 --- a/backend/services/assistant_store.py +++ b/backend/services/assistant_store.py @@ -88,6 +88,39 @@ def event(db, project_id, action, actor, detail=None, run_id=None): db.execute("INSERT INTO assistant_events(project_id,run_id,body) VALUES (?,?,?)", (project_id, run_id, json.dumps(body, ensure_ascii=False))) +def idempotent_lookup(db, actor, operation, key, request_hash): + """幂等键查询（新增的共享 helper）。 + + 既有 enqueue / create_report 各自内联同一段逻辑；本函数是新增实现，供新路径 + 使用。**未改写那两处**，避免影响既有运行与报告流程。 + + 返回既有 object_id；同键异载荷抛 Conflict（对应 E-09 IDEMPOTENCY_CONFLICT）。 + """ + prior = db.execute( + "SELECT request_hash,object_id FROM assistant_idempotency WHERE actor=? AND operation=? AND key=?", + (actor, operation, key), + ).fetchone() + if not prior: + return None + if prior[0] != request_hash: + raise Conflict("幂等键已用于不同请求") + return prior[1] + + +def idempotent_remember(db, actor, operation, key, request_hash, object_id): + """登记幂等键。 + + 用 INSERT OR REPLACE 而非既有内联代码的裸 INSERT：重试路径上同一个键可能再次 + 到达，裸 INSERT 会因主键冲突抛 sqlite3.IntegrityError，那不是 Conflict，无法被 + translate() 映射为 409。同键同载荷在 lookup 阶段已提前返回，故此处的覆盖分支 + 不可达，语义不变。 + """ + db.execute( + "INSERT OR REPLACE INTO assistant_idempotency VALUES (?,?,?,?,?)", + (actor, operation, key, request_hash, object_id), + ) + + def time_event(db, project_id, phase_code, phase_name, role, category, actor, source_type, source_id, status="running", started_at=None, ended_at=None, detail=None, event_id=None): """Create an immutable phase interval; callers close it with finish_time_event.""" diff --git a/backend/services/fact_conflict.py b/backend/services/fact_conflict.py new file mode 100644 index 0000000..183fa83 --- /dev/null +++ b/backend/services/fact_conflict.py @@ -0,0 +1,197 @@ +"""S1 事实确认服务：多来源数值冲突的检测、阻断与裁定。 + +职责 owner（docs/v2/03-detailed-design.md §9.2「事实确认服务」）。 + +解决的真实缺口（claude_use/T00/04 §3）： + 既有 `assistant_materials.apply_candidate()` 对 `values[field]` / `facts[field]` + 是**后写覆盖**；`sources[field]` 虽累积多个来源，但**无人检测分歧、无人阻断**。 + 这正是 BR-23 禁止的行为（"不按上传先后覆盖"）。本模块在**新路径**前置检测， + **不修改** apply_candidate 的既有语义（调用方 supplement / operations_demo 不变）。 + +规则（BR-23 / E-17 / AC-24）： +1. 检测：同字段的新候选值与已确认值不同 → 建 FactConflict(unresolved) +2. 阻断：该字段存在 unresolved 冲突时，**不写** values / selected_sources / facts； + 证据仍按 BR-22 记入 sources（来源留痕，值不采用） +3. 裁定：resolve 显式选择候选并要求说明 + 操作人；此后该字段才可确认 +4. 不覆盖：再次分歧新建对象并用 supersedes_id 指向旧对象，不改历史 +""" + +from __future__ import annotations + +from contracts.prescreen import ErrorCode, NOTE_MAX +from services import assistant_store as store +from services.assistant_materials import FIELDS, apply_candidate, scalar +from pydantic import ValidationError + +KIND = "fact_conflict" +STANDARD_FIELDS = frozenset(FIELDS.values()) + + +class FactConflictUnresolved(ValueError): + """该字段存在未解决的事实冲突，拒绝确认（E-17）。 + + HTTP 语义：作为**确认动作**的拒绝返回 422（INVALID_REQUEST 同族但 code 保持 + FACT_CONFLICT_UNRESOLVED）。注意与**运行级**结果区分——运行结果里未解决的冲突 + 保存在 degradations/finding 中，HTTP 仍为 200（01 §5 末）。同一 code 的两个面。 + """ + + code = ErrorCode.FACT_CONFLICT_UNRESOLVED + + +def _check_field(field: str) -> None: + """标准字段白名单校验留在 service 层（契约有意不查，避免 contracts→services）。""" + if field not in STANDARD_FIELDS: + raise ValueError(f"不支持的字段：{field}；请使用标准业务字段") + + +def _normalized(field: str, value): + """用与 apply_candidate 相同的标量规范化，使比较与落库口径一致。""" + return scalar(field, value) + + +def blocked_fields(db, project_id) -> set[str]: + """该项目下所有 unresolved 冲突的字段。""" + return { + row["field"] + for row in store.listing(db, KIND, project_id) + if row["status"] == "unresolved" + } + + +def assert_confirmable(db, project_id, fields) -> None: + """确认前置检查：命中未解决冲突即拒绝（BR-23 阻断）。""" + hit = sorted(set(fields) & blocked_fields(db, project_id)) + if hit: + raise FactConflictUnresolved( + "材料存在不一致，请核对并确认；不得静默选择。涉及字段：" + "、".join(hit) + ) + + +def _candidate_row(candidate: dict, evidence: dict) -> dict: + return { + "evidence_id": evidence["id"], + "value": None if evidence.get("value") is None else str(evidence["value"]), + "source": str(candidate.get("source") or evidence.get("source") or "未知来源"), + "locator": str(candidate.get("locator") or evidence.get("locator") or "未定位"), + "unit": candidate.get("unit"), + "currency": candidate.get("currency"), + "period": candidate.get("period"), + } + + +def _validate(payload: dict) -> dict: + from contracts.prescreen import FactConflict + + try: + return FactConflict(**payload).model_dump(mode="json") + except ValidationError as exc: + first = exc.errors()[0] + raise ValueError(f"事实冲突记录不合法：{first['msg']}") from exc + + +def detect(project, candidate: dict, actor: str, reason: str) -> dict | None: + """检测单个候选是否与已确认值冲突，并就地更新 `project`。 + + **本函数不落库**，调用方负责： + 1) `payload = detect(project, candidate, actor, reason)` + 2) `store.put(db, "project", project)` —— 保存证据与（可能的）已采用值 + 3) 若 payload 非 None：`create(db, project["id"], payload, actor)` + + 返回待落库的 FactConflict 载荷；无冲突返回 None。 + 证据在两种情况下都记入 `sources`：冲突时以 `adopt=False` 记录 —— 来源留痕， + 但值**不采用**（BR-22 与 BR-23 同时满足）。 + """ + field = candidate["field"] + _check_field(field) + incoming = _normalized(field, candidate.get("value")) + confirmed = project.get("values", {}).get(field) + + if confirmed is None or incoming is None or str(incoming) == str(confirmed): + # 首次确认、或双方都未知、或值相同 → 不构成分歧 + apply_candidate(project, candidate, actor, reason) + return None + + evidence = apply_candidate(project, candidate, actor, reason, adopt=False) + old_evidence_id = project.get("selected_sources", {}).get(field) + if old_evidence_id is None: + # 已确认值却没有来源记录：属数据异常，不静默生成冲突对象 + raise ValueError(f"字段 {field} 已有确认值但缺少来源记录，无法判定冲突") + + payload = { + "id": store.uid("fc"), + "project_id": project["id"], + "data_domain": project.get("data_domain", "demo"), + "field": field, + "candidate_evidence_ids": [old_evidence_id, evidence["id"]], + "candidates": [ + {"evidence_id": old_evidence_id, "value": str(confirmed), + "source": "已确认值", "locator": "项目事实"}, + _candidate_row(candidate, evidence), + ], + "status": "unresolved", + "created_at": store.now(), + } + # revision 由 put() 维护 + return payload + + +def create(db, project_id, payload: dict, actor: str, supersedes_id: str | None = None) -> dict: + # revision 由 store.put() 维护；新对象的初值为 1，这里显式带上以通过契约校验 + # （FactConflict.revision 为必填，线格式里必须可见）。 + body = _validate(dict(payload, revision=1, supersedes_id=supersedes_id)) + body = store.put(db, KIND, body) + store.event(db, project_id, "fact_conflict.created", actor, + {"conflict_id": body["id"], "field": body["field"]}) + return body + + +def load(db, conflict_id, project_id=None) -> dict: + row = store.get(db, KIND, conflict_id) + if project_id is not None and row["project_id"] != project_id: + raise KeyError("事实冲突不存在") + return row + + +def listing(db, project_id, status=None, limit=20) -> dict: + rows = store.listing(db, KIND, project_id) + if status: + rows = [row for row in rows if row["status"] == status] + return {"items": rows[:limit], "next_cursor": None if len(rows) <= limit else str(limit)} + + +def resolve(db, conflict_id, project, selected_evidence_ids, note, actor, expected): + """显式裁定：写入所选值，并把冲突置为 resolved。 + + 只允许 unresolved → resolved；已 resolved 再次调用即版本冲突（409）。 + """ + if not isinstance(note, str) or not note.strip(): + raise ValueError("请填写裁定说明") + if len(note) > NOTE_MAX: + raise ValueError(f"裁定说明过长（上限 {NOTE_MAX} 字）") + + row = store.get(db, KIND, conflict_id) + if row["status"] != "unresolved": + raise store.Conflict("该冲突已裁定，不能重复裁定") + + selected = list(selected_evidence_ids) + if not selected: + raise ValueError("请选择采纳的来源") + unknown = sorted(set(selected) - set(row["candidate_evidence_ids"])) + if unknown: + raise ValueError(f"所选的来源不属于该冲突的候选：{unknown}") + + chosen = next(c for c in row["candidates"] if c["evidence_id"] == selected[0]) + field = row["field"] + _check_field(field) + value = _normalized(field, chosen["value"]) + project.setdefault("values", {})[field] = value + project.setdefault("selected_sources", {})[field] = chosen["evidence_id"] + old_key = next((k for k, v in FIELDS.items() if v == field), field) + project.setdefault("facts", {})[old_key] = value + + row.update(status="resolved", resolution_note=note, + selected_evidence_ids=selected, resolved_by=actor, resolved_at=store.now()) + row = store.put(db, KIND, row, expected=expected) + store.event(db, row["project_id"], "fact_conflict.resolved", actor, + {"conflict_id": row["id"], "field": field, "selected": selected}) + return row diff --git a/backend/services/material_import.py b/backend/services/material_import.py new file mode 100644 index 0000000..89087e7 --- /dev/null +++ b/backend/services/material_import.py @@ -0,0 +1,195 @@ +"""S1 材料导入服务：文件集合清单的生命周期。 + +职责 owner（docs/v2/03-detailed-design.md §9.2「材料导入服务」）： + 集合清单、逐项状态与重试、批次状态派生。 + +边界（不得越界）： +- **不建立平行上传 API**：文件字节仍走既有 + `POST /projects/{project_id}/documents` → `ProjectAssistantService.upload()` + → `assistant_materials.material()`。本模块只登记清单与回写解析结果。 +- **不创建平行持久化入口**：复用 `assistant_objects`（kind=`import_batch`）。 + items 是批次内的值对象（契约中 `ImportItem` 无 `revision`），故不另立 + `import_item` kind；批次 revision 由 `put()` 维护。 +- **不做真实性核验**：`SourceLocator` 能定位不等于材料真实可信（01 §1）。 +""" + +from __future__ import annotations + +from services import assistant_store as store +from contracts.prescreen import ErrorCode, MAX_BATCH_BYTES, MAX_BATCH_FILES + +KIND = "import_batch" +IDEMPOTENCY_OPERATION = "import_batch" + +# 解析实现版本：冻结进批次，使"用哪版解析器得到这个结果"可追溯（BR-22）。 +# 解析逻辑在 assistant_materials.parse()，其行为变更时必须同步提升此版本。 +PARSER_VERSION = "assistant_materials.parse@1" + +_STATUS_ORDER = ("queued", "parsing", "parsed", "failed") + + +def _derive_status(items: list[dict]) -> str: + """由逐项状态派生批次状态。 + + ready = 全部 parsed + failed = 全部 failed + partial = 既有 parsed 又有 failed（含 queued/parsing 混入的中间态） + """ + statuses = {item["status"] for item in items} + if statuses == {"parsed"}: + return "ready" + if statuses == {"failed"}: + return "failed" + if "failed" in statuses and "parsed" in statuses: + return "partial" + return "parsing" + + +def _validate_manifest(manifest: list[dict]) -> list[dict]: + from pydantic import ValidationError + from contracts.prescreen import ImportItem + + if not manifest: + raise ValueError("材料集合为空；请选择至少一个文件") + if len(manifest) > MAX_BATCH_FILES: + raise ValueError(f"单批最多 {MAX_BATCH_FILES} 个文件；请拆分后重试") + + items: list[dict] = [] + seen_sha: set[str] = set() + for index, entry in enumerate(manifest): + try: + parsed = ImportItem(**entry) + except ValidationError as exc: + # 逐项错误沿用 E-01/E-16 的文案口径，指明是第几项 + first = exc.errors()[0] + where = ".".join(str(p) for p in first["loc"]) or "输入" + raise ValueError(f"第 {index + 1} 个文件不合法（{where}）：{first['msg']}") from exc + if parsed.sha256 in seen_sha: + raise ValueError(f"第 {index + 1} 个文件与同批其他条目哈希重复：{parsed.relative_path}") + seen_sha.add(parsed.sha256) + items.append(parsed.model_dump(mode="json")) + + total = sum(item["size_bytes"] for item in items) + if total > MAX_BATCH_BYTES: + raise ValueError(f"单批总量 {total} 超过上限 {MAX_BATCH_BYTES} 字节；请拆分后重试") + return items + + +def create(db, project_id, data_domain, manifest, actor, idempotency_key="", parser_version=PARSER_VERSION): + """登记一次文件集合导入。返回 ImportBatch（status=queued）。 + + 幂等：同一 actor+project+key 且载荷相同 → 返回原批次；载荷不同 → E-09。 + """ + request_hash = store.digest({"project": project_id, "domain": data_domain, "manifest": manifest}) + if idempotency_key: + prior = store.idempotent_lookup(db, actor, IDEMPOTENCY_OPERATION, idempotency_key, request_hash) + if prior: + return store.get(db, KIND, prior) + + items = _validate_manifest(manifest) + batch = { + "id": store.uid("import"), + "project_id": project_id, + "data_domain": data_domain, + "status": "queued", + "created_at": store.now(), + "created_by": actor, + "parser_version": parser_version, + "items": items, + "total_bytes": sum(item["size_bytes"] for item in items), + "warnings": [], + } + batch = store.put(db, KIND, batch) # revision=1 + if idempotency_key: + store.idempotent_remember(db, actor, IDEMPOTENCY_OPERATION, idempotency_key, request_hash, batch["id"]) + store.event(db, project_id, "import_batch.created", actor, + {"batch_id": batch["id"], "files": len(items), "total_bytes": batch["total_bytes"]}) + return batch + + +def load(db, batch_id, project_id=None): + batch = store.get(db, KIND, batch_id) + if project_id is not None and batch["project_id"] != project_id: + # 不存在与无访问权统一对外 404（03 §2），故这里也走 KeyError + raise KeyError("导入批次不存在") + return batch + + +def listing(db, project_id, limit=20): + rows = store.listing(db, KIND, project_id) + return {"items": rows[:limit], "next_cursor": None if len(rows) <= limit else str(limit)} + + +def mark_item(db, batch_id, item_id, status, document_id=None, error_code=None, warning=None): + """由既有上传链回写单条解析结果。 + + `parsed` 必须带 document_id；`failed` 必须带 error_code —— 与契约同规则， + 避免出现"状态说成功但无法定位原件"的条目（AC-23）。 + """ + if status not in _STATUS_ORDER: + raise ValueError(f"未知条目状态：{status}") + if status == "failed" and not error_code: + raise ValueError("failed 的条目必须给出 error_code") + batch = store.get(db, KIND, batch_id) + item = next((i for i in batch["items"] if i["id"] == item_id), None) + if item is None: + raise KeyError("导入条目不存在") + if batch["status"] == "cancelled": + # 取消后忽略迟到结果（03 §9.5），不写回 + return batch + item["status"] = status + if document_id is not None: + item["document_id"] = document_id + item["error_code"] = error_code if status == "failed" else None + if warning: + batch["warnings"].append(f"{item['relative_path']}: {warning}") + batch["status"] = _derive_status(batch["items"]) + return store.put(db, KIND, batch) + + +def retry(db, batch_id, item_ids, actor, expected): + """仅把失败项重置为 queued；已 parsed 的条目不重解析、不重写（AC-23 反例）。""" + batch = store.get(db, KIND, batch_id) + if batch["status"] == "cancelled": + raise store.Conflict("批次已取消，不能重试") + if batch["status"] not in ("partial", "failed"): + raise ValueError("只有部分失败或全部失败的批次可以重试") + + targets = set(item_ids) + unknown = targets - {i["id"] for i in batch["items"]} + if unknown: + raise KeyError(f"导入条目不存在：{sorted(unknown)}") + + reset = [] + for item in batch["items"]: + if item["id"] not in targets: + continue + if item["status"] != "failed": + raise ValueError(f"条目 {item['relative_path']} 不是失败项，不能重试") + item["status"] = "queued" + item["error_code"] = None + reset.append(item["id"]) + + batch["status"] = _derive_status(batch["items"]) + batch = store.put(db, KIND, batch, expected=expected) + store.event(db, batch["project_id"], "import_batch.retried", actor, + {"batch_id": batch_id, "items": reset}) + return batch + + +def cancel(db, batch_id, actor, expected): + batch = store.get(db, KIND, batch_id) + if batch["status"] in ("ready", "partial", "failed", "cancelled"): + raise store.Conflict("批次已结束，不能取消") + batch["status"] = "cancelled" + batch = store.put(db, KIND, batch, expected=expected) + store.event(db, batch["project_id"], "import_batch.cancelled", actor, {"batch_id": batch_id}) + return batch + + +# 供 S1c 接口层把服务端异常映射为错误码（03 §2 的统一信封） +ERROR_BY_EXCEPTION = { + "KeyError": ErrorCode.NOT_FOUND, + "Conflict": ErrorCode.REVISION_CONFLICT, + "ValueError": ErrorCode.IMPORT_BATCH_INVALID, +} diff --git a/backend/test_prescreen_contracts.py b/backend/test_prescreen_contracts.py index f4aa919..97bebe6 100644 --- a/backend/test_prescreen_contracts.py +++ b/backend/test_prescreen_contracts.py @@ -22,6 +22,7 @@ from contracts.prescreen import ( ImportItem, Ref, S1_ERROR_HTTP, + S1_ERROR_RETRYABLE, SourceLocator, ) @@ -298,9 +299,14 @@ class ErrorContractTests(unittest.TestCase): self.assertEqual(S1_ERROR_HTTP[ErrorCode.IDEMPOTENCY_CONFLICT], 409) self.assertEqual(S1_ERROR_HTTP[ErrorCode.IMPORT_BATCH_INVALID], 422) - def test_e17_is_not_an_http_error(self): - # 01 §5 末：运行级错误 HTTP 200，只表示结果可读 - self.assertEqual(S1_ERROR_HTTP[ErrorCode.FACT_CONFLICT_UNRESOLVED], 200) + def test_e17_confirm_refusal_is_422(self): + # E-17 有两个面：确认动作被拒 = 422（客户端须先裁定）； + # 运行级结果 = HTTP 200 且冲突保存在 degradations（那个面不走错误信封）。 + self.assertEqual(S1_ERROR_HTTP[ErrorCode.FACT_CONFLICT_UNRESOLVED], 422) + + def test_all_codes_are_non_retryable(self): + # 03 §5：403/422 不自动重试；409 需客户端改载荷或刷新 + self.assertFalse(any(S1_ERROR_RETRYABLE.values())) def test_envelope_shape(self): body = ErrorResponse.model_validate({"error": { diff --git a/backend/test_prescreen_storage.py b/backend/test_prescreen_storage.py new file mode 100644 index 0000000..4b6bce1 --- /dev/null +++ b/backend/test_prescreen_storage.py @@ -0,0 +1,441 @@ +"""S1 存储层测试：材料导入批次与事实冲突。 + +重点证明的是**旧实现会失败的行为**：`apply_candidate` 对字段值是后写覆盖， +本层必须让"后到的冲突来源不得覆盖已确认值"成立（BR-23 / AC-24）。 + +用独立临时 SQLite 库，绝不触碰 backend/xingbang_data 下的真实库。 +""" + +import tempfile +import unittest +from pathlib import Path + +from pydantic import ValidationError + +from contracts.prescreen import ErrorCode +from services import assistant_store as store +from services import fact_conflict, material_import +from services.assistant_materials import FIELDS + + +def manifest_entry(name="a.csv", sha="a" * 64, size=100, mime="text/csv"): + return {"id": store.uid("item"), "relative_path": name, "sha256": sha, + "size_bytes": size, "mime_type": mime, "status": "queued", + "received_at": store.now()} + + +class StoreTestCase(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.db_path = Path(self._tmp.name) / "prescreen_test.sqlite3" + store.initialize(self.db_path) + self.actor = "demo-manager" + with store.transaction(self.db_path) as db: + self._seed_project(db) + + def tearDown(self): + self._tmp.cleanup() + + @property + def project_id(self): + return "DEMO-A" + + def _seed_project(self, db): + store.put(db, "project", { + "id": self.project_id, "data_domain": "demo", "revision": 1, + "facts": {}, "values": {}, "sources": {}, "selected_sources": {}, + "document_ids": [], "pending_candidates": [], "data_revision": 1, + }) + + def project(self, db): + return store.get(db, "project", self.project_id) + + def candidate(self, field="facility.amount", value="1000", unit="万元", **over): + row = {"id": store.uid("candidate"), "field": field, "value": value, + "unit": unit, "currency": "CNY", "period": "2025", "scope": "合并", + "source": "a.csv", "locator": "第2行", "document_id": "doc-1"} + row.update(over) + return row + + +class ImportBatchTests(StoreTestCase): + def test_create_is_queued_with_derived_total(self): + with store.transaction(self.db_path) as db: + batch = material_import.create( + db, self.project_id, "demo", + [manifest_entry(size=100), manifest_entry(name="b.csv", sha="b" * 64, size=50)], + self.actor) + self.assertEqual(batch["status"], "queued") + self.assertEqual(batch["total_bytes"], 150) + self.assertEqual(batch["revision"], 1) + self.assertEqual(batch["parser_version"], material_import.PARSER_VERSION) + + def test_empty_manifest_rejected(self): + with store.transaction(self.db_path) as db: + with self.assertRaises(ValueError) as ctx: + material_import.create(db, self.project_id, "demo", [], self.actor) + self.assertIn("为空", str(ctx.exception)) + + def test_duplicate_sha_in_one_batch_rejected(self): + with store.transaction(self.db_path) as db: + with self.assertRaises(ValueError) as ctx: + material_import.create(db, self.project_id, "demo", + [manifest_entry(), manifest_entry(name="b.csv")], self.actor) + self.assertIn("哈希重复", str(ctx.exception)) + + def test_absolute_path_entry_reports_which_item(self): + with store.transaction(self.db_path) as db: + with self.assertRaises(ValueError) as ctx: + material_import.create(db, self.project_id, "demo", + [manifest_entry(name="C:/x.csv")], self.actor) + self.assertIn("第 1 个文件", str(ctx.exception)) + + def test_batch_total_over_limit_rejected(self): + from contracts.prescreen import MAX_ITEM_BYTES + entries = [manifest_entry(name=f"f{n}.csv", sha=f"{n}" * 64, size=MAX_ITEM_BYTES) + for n in range(3)] + with store.transaction(self.db_path) as db: + with self.assertRaises(ValueError) as ctx: + material_import.create(db, self.project_id, "demo", entries, self.actor) + self.assertIn("超过上限", str(ctx.exception)) + + def test_idempotent_replay_returns_same_batch(self): + entries = [manifest_entry()] + with store.transaction(self.db_path) as db: + first = material_import.create(db, self.project_id, "demo", entries, + self.actor, idempotency_key="k1") + again = material_import.create(db, self.project_id, "demo", entries, + self.actor, idempotency_key="k1") + self.assertEqual(first["id"], again["id"]) + + def test_idempotent_key_with_different_payload_conflicts(self): + with store.transaction(self.db_path) as db: + material_import.create(db, self.project_id, "demo", [manifest_entry()], + self.actor, idempotency_key="k2") + with self.assertRaises(store.Conflict): + material_import.create(db, self.project_id, "demo", + [manifest_entry(sha="c" * 64)], self.actor, + idempotency_key="k2") + + def test_status_derivation_partial_ready_failed(self): + with store.transaction(self.db_path) as db: + batch = material_import.create( + db, self.project_id, "demo", + [manifest_entry(name="ok.csv"), manifest_entry(name="bad.csv", sha="b" * 64)], + self.actor) + ids = [item["id"] for item in batch["items"]] + batch = material_import.mark_item(db, batch["id"], ids[0], "parsed", + document_id="doc-1") + self.assertEqual(batch["status"], "parsing") + batch = material_import.mark_item(db, batch["id"], ids[1], "failed", + error_code=ErrorCode.MATERIAL_PARSE_FAILED) + self.assertEqual(batch["status"], "partial") + self.assertEqual(batch["revision"], 3) + + def test_mark_item_failed_requires_error_code(self): + with store.transaction(self.db_path) as db: + batch = material_import.create(db, self.project_id, "demo", [manifest_entry()], self.actor) + with self.assertRaises(ValueError): + material_import.mark_item(db, batch["id"], batch["items"][0]["id"], "failed") + + def test_retry_only_resets_failed_items(self): + # AC-23 反例：已 parsed 的条目不得被重解析、不得被重写 + with store.transaction(self.db_path) as db: + batch = material_import.create( + db, self.project_id, "demo", + [manifest_entry(name="ok.csv"), manifest_entry(name="bad.csv", sha="b" * 64)], + self.actor) + ids = [item["id"] for item in batch["items"]] + material_import.mark_item(db, batch["id"], ids[0], "parsed", document_id="doc-1") + batch = material_import.mark_item(db, batch["id"], ids[1], "failed", + error_code=ErrorCode.MATERIAL_PARSE_FAILED) + parsed_before = next(i for i in batch["items"] if i["id"] == ids[0]) + + batch = material_import.retry(db, batch["id"], [ids[1]], self.actor, + expected=batch["revision"]) + parsed_after = next(i for i in batch["items"] if i["id"] == ids[0]) + self.assertEqual(parsed_before, parsed_after) + self.assertEqual(next(i for i in batch["items"] if i["id"] == ids[1])["status"], "queued") + + def test_retry_rejects_non_failed_item(self): + with store.transaction(self.db_path) as db: + batch = material_import.create( + db, self.project_id, "demo", + [manifest_entry(name="ok.csv"), manifest_entry(name="bad.csv", sha="b" * 64)], + self.actor) + ids = [item["id"] for item in batch["items"]] + material_import.mark_item(db, batch["id"], ids[0], "parsed", document_id="doc-1") + batch = material_import.mark_item(db, batch["id"], ids[1], "failed", + error_code=ErrorCode.MATERIAL_PARSE_FAILED) + with self.assertRaises(ValueError): + material_import.retry(db, batch["id"], [ids[0]], self.actor, expected=batch["revision"]) + + def test_retry_with_stale_revision_conflicts(self): + with store.transaction(self.db_path) as db: + batch = material_import.create(db, self.project_id, "demo", + [manifest_entry()], self.actor) + material_import.mark_item(db, batch["id"], batch["items"][0]["id"], "failed", + error_code=ErrorCode.MATERIAL_PARSE_FAILED) + with self.assertRaises(store.Conflict): + material_import.retry(db, batch["id"], [batch["items"][0]["id"]], + self.actor, expected=1) + + def test_cancel_then_late_result_is_ignored(self): + with store.transaction(self.db_path) as db: + batch = material_import.create(db, self.project_id, "demo", [manifest_entry()], self.actor) + batch = material_import.cancel(db, batch["id"], self.actor, expected=batch["revision"]) + item_id = batch["items"][0]["id"] + after = material_import.mark_item(db, batch["id"], item_id, "parsed", document_id="doc-1") + self.assertEqual(after["status"], "cancelled") + self.assertIsNone(after["items"][0]["document_id"]) + + def test_cross_project_read_is_not_found(self): + with store.transaction(self.db_path) as db: + batch = material_import.create(db, self.project_id, "demo", [manifest_entry()], self.actor) + with self.assertRaises(KeyError): + material_import.load(db, batch["id"], project_id="DEMO-B") + + +class FactConflictTests(StoreTestCase): + def _confirm(self, db, candidate): + """走一遍真实链路：detect 就地改 project → 保存 → 有冲突则落库。 + + 注意必须先取到 project 字典再保存**同一个对象**：detect 是就地修改， + 若重新 store.get 一份新副本去 put，改动会被丢弃。 + """ + project = self.project(db) + payload = fact_conflict.detect(project, candidate, self.actor, "初始确认") + store.put(db, "project", project) + if payload: + return fact_conflict.create(db, self.project_id, payload, self.actor) + return None + + def _resolve(self, db, conflict, selected, note="裁定说明"): + project = self.project(db) + row = fact_conflict.resolve(db, conflict["id"], project, selected, note, + self.actor, expected=conflict["revision"]) + store.put(db, "project", project) + return row + + def _confirmed_value(self, project, field): + return project["values"].get(field) + + def test_first_confirmation_has_no_conflict(self): + with store.transaction(self.db_path) as db: + self.assertIsNone(self._confirm(db, self.candidate(value="1000"))) + project = self.project(db) + # 不硬编码换算结果：断言"值等于被采纳证据的值"，与单元/币种换算解耦 + evidence = project["sources"]["facility.amount"][-1] + self.assertEqual(self._confirmed_value(project, "facility.amount"), evidence["value"]) + self.assertIsNotNone(self._confirmed_value(project, "facility.amount")) + + def test_conflicting_second_source_does_not_overwrite_and_records_evidence(self): + # ★ 这是旧实现会失败的核心行为：apply_candidate 曾直接覆盖 + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + before = self._confirmed_value(self.project(db), "facility.amount") + + conflict = self._confirm(db, self.candidate(value="1200", source="b.csv", locator="第9行")) + self.assertIsNotNone(conflict) + self.assertEqual(conflict["status"], "unresolved") + self.assertEqual(conflict["field"], "facility.amount") + + project = self.project(db) + self.assertEqual(self._confirmed_value(project, "facility.amount"), before, + "已确认值被覆盖了") + # 证据仍按 BR-22 留痕：两个来源都在 + self.assertEqual(len(project["sources"]["facility.amount"]), 2) + self.assertEqual(len(conflict["candidates"]), 2) + self.assertNotEqual(conflict["candidates"][0]["value"], + conflict["candidates"][1]["value"]) + + def test_blocked_field_refuses_confirmation(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + self._confirm(db, self.candidate(value="1200")) + with self.assertRaises(fact_conflict.FactConflictUnresolved) as ctx: + fact_conflict.assert_confirmable(db, self.project_id, ["facility.amount"]) + self.assertIn("不一致", str(ctx.exception)) + self.assertEqual(ctx.exception.code, ErrorCode.FACT_CONFLICT_UNRESOLVED) + + def test_other_fields_remain_confirmable(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + self._confirm(db, self.candidate(value="1200")) + fact_conflict.assert_confirmable(db, self.project_id, ["financial.total_assets"]) + + def test_same_value_is_not_a_conflict(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + self.assertIsNone(self._confirm(db, self.candidate(value="1000", source="b.csv"))) + self.assertEqual(fact_conflict.blocked_fields(db, self.project_id), set()) + + def test_unknown_candidate_does_not_create_conflict(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + self.assertIsNone(self._confirm(db, self.candidate(value=None))) + + def test_resolve_adopts_chosen_value_and_unblocks(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + conflict = self._confirm(db, self.candidate(value="1200")) + chosen = conflict["candidates"][1]["evidence_id"] + + resolved = self._resolve(db, conflict, [chosen], "以合同金额为准") + self.assertEqual(resolved["status"], "resolved") + self.assertEqual(resolved["resolved_by"], self.actor) + self.assertIsNotNone(resolved["resolved_at"]) + + project = self.project(db) + # 采纳后该字段的值来自被选中的候选 + self.assertEqual(str(self._confirmed_value(project, "facility.amount")), + str(conflict["candidates"][1]["value"])) + self.assertEqual(project["selected_sources"]["facility.amount"], chosen) + self.assertEqual(fact_conflict.blocked_fields(db, self.project_id), set()) + + def test_resolve_can_keep_the_original_value(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + conflict = self._confirm(db, self.candidate(value="1200")) + original = conflict["candidates"][0]["evidence_id"] + self._resolve(db, conflict, [original], "以已确认为准") + project = self.project(db) + self.assertEqual(str(self._confirmed_value(project, "facility.amount")), + str(conflict["candidates"][0]["value"])) + + def test_resolve_requires_note_and_valid_selection(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + conflict = self._confirm(db, self.candidate(value="1200")) + chosen = conflict["candidates"][0]["evidence_id"] + + with self.assertRaises(ValueError): + fact_conflict.resolve(db, conflict["id"], self.project(db), [chosen], " ", + self.actor, expected=conflict["revision"]) + with self.assertRaises(ValueError) as ctx: + fact_conflict.resolve(db, conflict["id"], self.project(db), ["ev-999"], + "说明", self.actor, expected=conflict["revision"]) + self.assertIn("不属于", str(ctx.exception)) + with self.assertRaises(ValueError): + fact_conflict.resolve(db, conflict["id"], self.project(db), [], + "说明", self.actor, expected=conflict["revision"]) + # 三次被拒后冲突仍未被改动 + self.assertEqual(fact_conflict.load(db, conflict["id"])["status"], "unresolved") + + def test_double_resolve_conflicts(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + conflict = self._confirm(db, self.candidate(value="1200")) + chosen = conflict["candidates"][1]["evidence_id"] + self._resolve(db, conflict, [chosen]) + with self.assertRaises(store.Conflict): + fact_conflict.resolve(db, conflict["id"], self.project(db), [chosen], "再裁一次", + self.actor, expected=conflict["revision"]) + + def test_third_source_creates_new_conflict_with_supersedes(self): + # AC-24：再次冲突新建对象并指向旧对象，不覆写历史 + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + first = self._confirm(db, self.candidate(value="1200")) + self._resolve(db, first, [first["candidates"][1]["evidence_id"]], "采纳1200") + + project = self.project(db) + payload = fact_conflict.detect(project, self.candidate(value="900", source="c.csv"), + self.actor, "第三来源") + store.put(db, "project", project) + self.assertIsNotNone(payload, "第三来源与已确认值不同，应产生冲突") + second = fact_conflict.create(db, self.project_id, payload, self.actor, + supersedes_id=first["id"]) + self.assertEqual(second["supersedes_id"], first["id"]) + self.assertNotEqual(second["id"], first["id"]) + # 旧记录仍为 resolved，未被改写 + old = fact_conflict.load(db, first["id"]) + self.assertEqual(old["status"], "resolved") + self.assertEqual(old["resolution_note"], "采纳1200") + + def test_non_standard_field_rejected(self): + with store.transaction(self.db_path) as db: + with self.assertRaises(ValueError) as ctx: + fact_conflict.detect(self.project(db), + self.candidate(field="not.a.real.field"), self.actor, "x") + self.assertIn("不支持的字段", str(ctx.exception)) + + def test_legacy_path_still_adopts_value(self): + # apply_candidate 的既有行为不受影响：仍然正常采用值 + from services.assistant_materials import apply_candidate + with store.transaction(self.db_path) as db: + project = self.project(db) + evidence = apply_candidate(project, self.candidate(value="1000"), self.actor, "旧路径") + self.assertEqual(project["values"]["facility.amount"], evidence["value"]) + self.assertEqual(project["selected_sources"]["facility.amount"], evidence["id"]) + + def test_adopt_false_records_evidence_without_adopting(self): + from services.assistant_materials import apply_candidate + with store.transaction(self.db_path) as db: + project = self.project(db) + evidence = apply_candidate(project, self.candidate(value="1000"), self.actor, + "待裁定", adopt=False) + self.assertNotIn("facility.amount", project["values"]) + self.assertNotIn("facility.amount", project["facts"]) + self.assertEqual(len(project["sources"]["facility.amount"]), 1) + self.assertEqual(project["sources"]["facility.amount"][0]["id"], evidence["id"]) + + def test_listing_filters_by_status(self): + with store.transaction(self.db_path) as db: + self._confirm(db, self.candidate(value="1000")) + conflict = self._confirm(db, self.candidate(value="1200")) + self.assertEqual( + len(fact_conflict.listing(db, self.project_id, status="unresolved")["items"]), 1) + self.assertEqual( + len(fact_conflict.listing(db, self.project_id, status="resolved")["items"]), 0) + self._resolve(db, conflict, [conflict["candidates"][1]["evidence_id"]]) + self.assertEqual( + len(fact_conflict.listing(db, self.project_id, status="resolved")["items"]), 1) + self.assertEqual( + len(fact_conflict.listing(db, self.project_id, status="unresolved")["items"]), 0) + + def test_standard_field_whitelist_is_the_existing_one(self): + # 不复制白名单：以 assistant_materials.FIELDS 为唯一来源 + self.assertIn("facility.amount", fact_conflict.STANDARD_FIELDS) + self.assertEqual(fact_conflict.STANDARD_FIELDS, frozenset(FIELDS.values())) + + def test_objects_are_persisted_through_the_generic_store(self): + """复用 assistant_objects（新 kind），未新建表 —— DDL 结论的机械证明。""" + with store.transaction(self.db_path) as db: + batch = material_import.create(db, self.project_id, "demo", + [manifest_entry()], self.actor) + self._confirm(db, self.candidate(value="1000")) + conflict = self._confirm(db, self.candidate(value="1200")) + + kinds = {row[0] for row in db.execute( + "SELECT DISTINCT kind FROM assistant_objects").fetchall()} + self.assertIn("import_batch", kinds) + self.assertIn("fact_conflict", kinds) + self.assertEqual(conflict["revision"], 1) + self.assertEqual(batch["revision"], 1) + + # 表结构未变：仍是 S1 之前的那 5 张表 + tables = {row[0] for row in db.execute( + "SELECT name FROM sqlite_master WHERE type='table'").fetchall()} + self.assertTrue({"assistant_objects", "assistant_events", "assistant_idempotency", + "assistant_jobs", "assistant_meta"} <= tables) + + +class IsolationTests(StoreTestCase): + def test_services_do_not_touch_the_real_data_root(self): + # 本模块的测试库必须在临时目录；真实库不受影响 + self.assertNotIn("xingbang_data", str(self.db_path)) + self.assertTrue(str(self.db_path).startswith(tempfile.gettempdir().replace("\\", "/")) + or "Temp" in str(self.db_path) or "tmp" in str(self.db_path)) + + def test_contracts_module_not_imported_by_services_is_false_direction_guard(self): + # 方向守卫：contracts 不得导入 services（本测试从 services 侧反查） + import contracts.prescreen as module + with open(module.__file__, encoding="utf-8") as handle: + source = handle.read() + self.assertNotIn("from services", source) + self.assertNotIn("import services", source) + + +if __name__ == "__main__": + unittest.main()