From 669e969c418972aac120b95c929050377b4ccdf5 Mon Sep 17 00:00:00 2001 From: qiaoqichao Date: Tue, 29 Sep 2026 18:21:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T03=20=E6=9D=A5=E6=BA=90=E5=AE=9A?= =?UTF-8?q?=E4=BD=8D=E6=8E=A5=E4=B8=8A=E5=A5=91=E7=BA=A6=EF=BC=8C=E5=B9=B6?= =?UTF-8?q?=E9=94=81=E5=AE=9A=E8=B7=A8=E9=A1=B9=E7=9B=AE=E5=BC=95=E7=94=A8?= =?UTF-8?q?=E6=8B=92=E7=BB=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T03 卡的要求是"复用上传/确认入口，增加导入清单、定位和冲突候选"。核对后： 导入清单与冲突候选已由 T02/S1 交付，但**"定位"只定义了契约、没有接上**—— parse() 仍只产出自由字符串 locator，SourceLocator 类型化投影是空的。 本次补上这条链路（契约与实现不一致是我的遗留）。 - assistant_materials.parse()：每条 chunk 增加 source_locator（kind + 坐标）， 按 T00/04 §5.3 的映射：pdf→page、xlsx 单元格/行与 csv/txt/docx 表格→sheet+cell_range、 docx 正文段→manual(note) - material() 补 document_id/document_version 后**按契约校验**；漏了定位或定位不合法 直接抛错，不落半成品进库 - **locator 自由字符串保留**：enqueue() 依赖它算证据 ID 的稳定哈希，改它会破坏 BR-13 实现中修掉一个自己引入的 bug：类型化定位的辅助函数初版命名为 table，与 docx 分支里 的表格计数器 table 同名，该分支内 helper 被整数遮蔽（'int' object is not callable）。 已改名 sheet_loc。 AC-08 补测：跨项目候选引用被 supplement 拒（"不属于项目"），同项目可用作对照， 未知 candidate_id 同样被拒。既有 test_project_assistant 只覆盖了跨项目文档下载。 测试：新增 test_sourcelocator.py 13 项，含真实 XLSX/DOCX fixture（zipfile 构造）， 覆盖五种格式的定位派生与契约校验。全量 248 项 OK（235 → 248）；verify.py 10/10 PASS。 Co-Authored-By: Claude Code --- backend/services/assistant_materials.py | 53 +++++- backend/test_sourcelocator.py | 210 ++++++++++++++++++++++++ 2 files changed, 254 insertions(+), 9 deletions(-) create mode 100644 backend/test_sourcelocator.py diff --git a/backend/services/assistant_materials.py b/backend/services/assistant_materials.py index f1f8960..18fdc79 100644 --- a/backend/services/assistant_materials.py +++ b/backend/services/assistant_materials.py @@ -55,18 +55,29 @@ def parse(content, filename): raise ValueError("材料为空或超过 10 MB") suffix = Path(filename).suffix.lower() chunks, warnings = [], [] + stem = Path(filename).stem def add(locator, text, **extra): if len(chunks) >= 15000: raise ValueError("材料内容超过首版处理上限") chunks.append({"locator": locator, "text": str(text)[:16000], **extra}) + # 类型化定位：与自由字符串 locator 并存（claude_use/T00/04 §5.3）。 + # **保留 locator** —— enqueue() 用它算证据 ID 的稳定哈希，改它会破坏版本冻结。 + # document_id / document_version 由 material() 补齐后按契约校验。 + def pdf(page): + return {"kind": "pdf", "page": page, "excerpt": f"第{page}页"} + def sheet_loc(sheet, cell_range): + return {"kind": "table", "sheet": sheet, "cell_range": cell_range, + "excerpt": f"{sheet}!{cell_range}" if sheet else cell_range} + def manual(note): + return {"kind": "manual", "note": note, "excerpt": note} if suffix in (".txt", ".csv"): text = content.decode("utf-8-sig", errors="strict") if suffix == ".txt": for index, line in enumerate(text.splitlines(), 1): - add(f"第{index}行", line) + add(f"第{index}行", line, source_locator=sheet_loc(stem, f"第{index}行")) else: for index, row in enumerate(csv.reader(io.StringIO(text)), 1): - add(f"第{index}行", " | ".join(row), cells=row) + add(f"第{index}行", " | ".join(row), cells=row, source_locator=sheet_loc(stem, f"第{index}行")) elif suffix == ".docx": with safe_zip(content) as archive: root = ET.fromstring(archive.read("word/document.xml")) @@ -75,12 +86,15 @@ def parse(content, filename): for element in root.find("w:body", ns): if element.tag.endswith("}p"): para += 1 - add(f"正文第{para}段", "".join(t.text or "" for t in element.findall(".//w:t", ns))) + # 正文段落不是行列可寻址的表格，归 manual（设计的三值枚举里没有第四种） + add(f"正文第{para}段", "".join(t.text or "" for t in element.findall(".//w:t", ns)), + source_locator=manual(f"正文第{para}段")) elif element.tag.endswith("}tbl"): table += 1 for r, row in enumerate(element.findall("w:tr", ns), 1): cells = ["".join(t.text or "" for t in cell.findall(".//w:t", ns)) for cell in row.findall("w:tc", ns)] - add(f"表{table}第{r}行", " | ".join(cells), cells=cells) + add(f"表{table}第{r}行", " | ".join(cells), cells=cells, + source_locator=sheet_loc(f"表{table}", f"第{r}行")) elif suffix == ".xlsx": ns = {"x": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"} with safe_zip(content) as archive: @@ -90,6 +104,7 @@ def parse(content, filename): rels = {r.get("Id"): r.get("Target") for r in ET.fromstring(archive.read("xl/_rels/workbook.xml.rels"))} book = ET.fromstring(archive.read("xl/workbook.xml")) for sheet in book.findall("x:sheets/x:sheet", ns): + sheet_name = sheet.get("name") target = rels.get(sheet.get("{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"), "") path = target.lstrip("/") if target.startswith("/") else "xl/" + target if ".." in path or path not in archive.namelist(): @@ -102,15 +117,19 @@ def parse(content, filename): text = strings[int(raw)] if kind == "s" and raw else ("".join(t.text or "" for t in cell.findall(".//x:t", ns)) if kind == "inlineStr" else raw) formula = cell.findtext("x:f", namespaces=ns) if formula is not None and raw == "": - warnings.append(sheet.get("name") + "!" + cell.get("r") + "公式无缓存，未计算") - add(sheet.get("name") + "!" + cell.get("r"), text, formula=formula, cached_value=raw if formula is not None else None) + warnings.append(sheet_name + "!" + cell.get("r") + "公式无缓存，未计算") + add(sheet_name + "!" + cell.get("r"), text, formula=formula, + cached_value=raw if formula is not None else None, + source_locator=sheet_loc(sheet_name, cell.get("r"))) cells.append(text) if cells: - add(sheet.get("name") + "!第" + row.get("r") + "行", " | ".join(cells), cells=cells) + add(sheet_name + "!第" + row.get("r") + "行", " | ".join(cells), cells=cells, + source_locator=sheet_loc(sheet_name, f"第{row.get('r')}行")) elif suffix in (".pdf", ".png", ".jpg", ".jpeg"): from api.collateral_agent import extract_pages for page in extract_pages(content, filename): - add("第" + str(page["number"]) + "页", page.get("text", ""), method=page.get("method"), status=page.get("status")) + add("第" + str(page["number"]) + "页", page.get("text", ""), method=page.get("method"), + status=page.get("status"), source_locator=pdf(page["number"])) if not page.get("text", "").strip(): warnings.append("第" + str(page["number"]) + "页未提取到文字，请启用 OCR 或补充可读材料") else: @@ -126,7 +145,23 @@ def material(project_id, filename, content, actor, version=1): chunks, warnings = parse(content, filename) except (OSError, zipfile.BadZipFile, ET.ParseError, UnicodeError, csv.Error, IndexError, TypeError) as exc: raise ValueError("材料内容损坏、编码不支持或格式不匹配，请检查原文件") from exc - body = {"id": uid("doc"), "project_id": project_id, "name": filename, "version": version, + document_id = uid("doc") + # 补齐并**按契约校验**类型化定位：document_id/document_version 只有到这一步才知道， + # 而 SourceLocator 要求二者必填。校验失败即视为解析产物不合规（不能落到库里当"定位"）。 + from pydantic import ValidationError + from contracts.prescreen import SourceLocator + for index, chunk in enumerate(chunks): + locator = chunk.get("source_locator") + if locator is None: + raise ValueError(f"解析产物缺少类型化定位（第 {index + 1} 段）") + try: + chunk["source_locator"] = SourceLocator( + **locator, document_id=document_id, document_version=version + ).model_dump(mode="json") + except ValidationError as exc: + first = exc.errors()[0] + raise ValueError(f"定位不合法（{first['loc'][-1] if first['loc'] else '未知'}）：{first['msg']}") from exc + body = {"id": document_id, "project_id": project_id, "name": filename, "version": version, "content_hash": sha256(content).hexdigest(), "created_at": now(), "actor_id": actor, "size": len(content), "chunks": chunks, "warnings": warnings, "status": "partial" if warnings else "parsed", "raw_base64": base64.b64encode(content).decode(), "candidates": []} diff --git a/backend/test_sourcelocator.py b/backend/test_sourcelocator.py new file mode 100644 index 0000000..4d696e0 --- /dev/null +++ b/backend/test_sourcelocator.py @@ -0,0 +1,210 @@ +"""T03 类型化来源定位测试：parse()/material() 必须产出契约合法的 SourceLocator。 + +背景：T02 定义了 `contracts.prescreen.SourceLocator`，但当时**没有接上** +（parse() 只产出自由字符串 locator）。本文件锁定这条链路： + parse() 产出 kind + 坐标 → material() 补 document_id/document_version → 契约校验 + +同时锁定 `locator` 自由字符串**必须保留**：enqueue() 用它算证据 ID 的稳定哈希， +改动它会破坏 BR-13 的版本冻结。 +""" + +import io +import unittest +import zipfile +from pathlib import Path + +from contracts.prescreen import SourceLocator +from services.assistant_materials import material, parse + + +def xlsx_bytes(rows=(("field", "value"), ("facility.amount", "100"))): + """构造最小可解析 XLSX（inlineStr，无需 sharedStrings）。""" + ns = 'xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"' + rel = 'xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"' + sheet_rows = "".join( + "%s" % ( + index, + "".join( + f'{cell}' + for col, cell in enumerate(row) + ), + ) + for index, row in enumerate(rows, 1) + ) + parts = { + "xl/workbook.xml": f'', + "xl/_rels/workbook.xml.rels": '' + '', + "xl/worksheets/sheet1.xml": f'{sheet_rows}', + } + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + for name, body in parts.items(): + archive.writestr(name, body) + return buffer.getvalue() + + +def docx_bytes(): + """构造最小可解析 DOCX：一段正文 + 一张表。""" + ns = 'xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main"' + body = ( + f'正文第一段' + f'甲' + f'乙' + ) + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr("word/document.xml", f"{body}") + return buffer.getvalue() + + +class TypedLocatorTests(unittest.TestCase): + def locators(self, content, filename): + document = material("DEMO-A", filename, content, "tester") + return document, [chunk["source_locator"] for chunk in document["chunks"]] + + def test_csv_rows_become_table_locators(self): + content = "field,value\nfacility.amount,100\n".encode("utf-8-sig") + document, locators = self.locators(content, "流水.csv") + first = locators[0] + self.assertEqual(first["kind"], "table") + self.assertEqual(first["sheet"], "流水") # 文件名去扩展名 + self.assertEqual(first["cell_range"], "第1行") + self.assertEqual(first["document_id"], document["id"]) + self.assertEqual(first["document_version"], 1) + + def test_txt_rows_become_table_locators(self): + content = "第一行\n第二行\n".encode("utf-8-sig") + _, locators = self.locators(content, "说明.txt") + self.assertEqual(locators[1]["cell_range"], "第2行") + self.assertEqual(locators[1]["kind"], "table") + + def test_xlsx_cells_and_rows_become_table_locators(self): + _, locators = self.locators(xlsx_bytes(), "台账.xlsx") + kinds = {item["kind"] for item in locators} + self.assertEqual(kinds, {"table"}) + # 单元格级定位：A1 必须是 sheet + cell_range，而不是行号 + cell = next(item for item in locators if item["cell_range"] == "A1") + self.assertEqual(cell["sheet"], "Sheet1") + # 行级定位也保留 + self.assertTrue(any(item["cell_range"].endswith("行") for item in locators)) + + def test_docx_paragraph_is_manual_and_table_is_table(self): + _, locators = self.locators(docx_bytes(), "说明.docx") + by_kind = {} + for item in locators: + by_kind.setdefault(item["kind"], []).append(item) + self.assertIn("manual", by_kind, "DOCX 正文段应归 manual（三值枚举里没有段落子类型）") + self.assertIn("table", by_kind, "DOCX 表格应归 table") + self.assertEqual(by_kind["manual"][0]["note"], "正文第1段") + self.assertEqual(by_kind["table"][0]["sheet"], "表1") + + def test_every_locator_validates_against_the_contract(self): + # material() 已按契约校验；这里再独立复核一遍，防止校验被绕过 + for content, name in [(xlsx_bytes(), "台账.xlsx"), (docx_bytes(), "说明.docx"), + ("a,b\n".encode("utf-8-sig"), "t.csv")]: + with self.subTest(name=name): + _, locators = self.locators(content, name) + self.assertTrue(locators) + for item in locators: + SourceLocator(**item) # 不合法会抛 ValidationError + + def test_free_string_locator_is_preserved(self): + """locator 必须保留：enqueue() 依赖它计算证据 ID 的稳定哈希（BR-13）。""" + content = "field,value\nfacility.amount,100\n".encode("utf-8-sig") + document = material("DEMO-A", "流水.csv", content, "tester") + strings = [chunk["locator"] for chunk in document["chunks"]] + self.assertIn("第1行", strings) + self.assertIn("第2行", strings) + # 每个 chunk 同时有两种形态 + for chunk in document["chunks"]: + self.assertIn("locator", chunk) + self.assertIn("source_locator", chunk) + self.assertIsInstance(chunk["locator"], str) + + def test_parse_without_material_fillup_has_no_document_id(self): + """parse() 本身不知道 document_id——这正是补齐放在 material() 的原因。""" + chunks, _ = parse("a\n".encode("utf-8-sig"), "t.txt") + raw = chunks[0]["source_locator"] + self.assertNotIn("document_id", raw) + self.assertEqual(raw["kind"], "table") + + def test_document_version_is_recorded_not_assumed(self): + content = "a,b\n".encode("utf-8-sig") + document = material("DEMO-A", "v.csv", content, "tester", version=7) + self.assertEqual(document["version"], 7) + self.assertEqual(document["chunks"][0]["source_locator"]["document_version"], 7) + + def test_locator_cannot_cross_documents(self): + """两个文档的定位互不串用：document_id 是各自的。""" + a = material("DEMO-A", "a.csv", "x,y\n".encode("utf-8-sig"), "tester") + b = material("DEMO-A", "b.csv", "x,y\n".encode("utf-8-sig"), "tester") + self.assertNotEqual(a["id"], b["id"]) + self.assertEqual(a["chunks"][0]["source_locator"]["document_id"], a["id"]) + self.assertEqual(b["chunks"][0]["source_locator"]["document_id"], b["id"]) + # 文档 a 的定位不可能出现在文档 b 里 + self.assertNotIn(a["id"], [c["source_locator"]["document_id"] for c in b["chunks"]]) + + def test_missing_typed_locator_is_rejected(self): + """防御性断言：若某条解析路径漏了 source_locator，material() 必须失败而不是静默产半成品。""" + import services.assistant_materials as materials + from unittest.mock import patch + original = materials.parse + + def stripped(content, filename): + chunks, warnings = original(content, filename) + for chunk in chunks: + chunk.pop("source_locator", None) + return chunks, warnings + + with patch.object(materials, "parse", stripped): + with self.assertRaises(ValueError) as ctx: + materials.material("DEMO-A", "t.csv", "a,b\n".encode("utf-8-sig"), "tester") + self.assertIn("类型化定位", str(ctx.exception)) + + +class ReferenceValidationTests(unittest.TestCase): + """AC-08：无效或跨项目引用必须被拒，且不暴露无权原文。""" + + def setUp(self): + import tempfile + from services.project_assistant import ProjectAssistantService + self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.service = ProjectAssistantService(Path(self.temp.name) / "assistant.sqlite3") + + def tearDown(self): + self.temp.cleanup() + + def test_candidate_from_another_project_is_rejected(self): + payload = "field,value,currency,unit,period,scope\nfacility.amount,100,CNY,万元,2025,合并\n".encode("utf-8-sig") + document = self.service.upload("DEMO-A", "流水.csv", payload, "demo-manager") + self.assertTrue(document["candidates"], "解析应产出候选") + candidate_id = document["candidates"][0]["id"] + + with self.assertRaises(ValueError) as ctx: + self.service.supplement("DEMO-B", "demo-manager", {}, [], + self.service.get_project("DEMO-B")["revision"], + "跨项目引用测试", candidate_ids=[candidate_id]) + self.assertIn("不属于项目", str(ctx.exception)) + + def test_candidate_reference_is_project_scoped(self): + """同项目可用、跨项目不可用——证明拒绝来自项目范围而不是候选不存在。""" + payload = "field,value,currency,unit,period,scope\nfacility.amount,100,CNY,万元,2025,合并\n".encode("utf-8-sig") + document = self.service.upload("DEMO-A", "流水.csv", payload, "demo-manager") + candidate_id = document["candidates"][0]["id"] + + project = self.service.get_project("DEMO-A") + result = self.service.supplement("DEMO-A", "demo-manager", {}, [], project["revision"], + "同项目确认来源", candidate_ids=[candidate_id]) + self.assertEqual(result["project_id"], "DEMO-A") + self.assertIn("facility.amount", result["changed_fields"]) + + def test_unknown_candidate_id_is_also_rejected(self): + project = self.service.get_project("DEMO-A") + with self.assertRaises(ValueError): + self.service.supplement("DEMO-A", "demo-manager", {}, [], project["revision"], + "伪造引用", candidate_ids=["candidate-does-not-exist"]) + + +if __name__ == "__main__": + unittest.main()