qiaoqichao/mini-openclaw-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v2): T03 来源定位接上契约，并锁定跨项目引用拒绝

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
qiaoqichao
2026-09-29 18:21:51 +08:00
co-authored by Claude Code
parent 985e2f6a53
commit 669e969c41
2 changed files with 254 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/assistant_materials.py
+44 -9
View File
@@ -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": []}
backend/test_sourcelocator.py
+210
View File
@@ -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(
"<row r=\"%d\">%s</row>" % (
index,
"".join(
f'<c r="{chr(65 + col)}{index}" t="inlineStr"><is><t>{cell}</t></is></c>'
for col, cell in enumerate(row)
),
)
for index, row in enumerate(rows, 1)
)
parts = {
"xl/workbook.xml": f'<workbook {ns}><sheets><sheet name="Sheet1" sheetId="1" r:id="rId1" {rel}/></sheets></workbook>',
"xl/_rels/workbook.xml.rels": '<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
'<Relationship Id="rId1" Target="worksheets/sheet1.xml"/></Relationships>',
"xl/worksheets/sheet1.xml": f'<worksheet {ns}><sheetData>{sheet_rows}</sheetData></worksheet>',
}
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'<w:p><w:r><w:t>正文第一段</w:t></w:r></w:p>'
f'<w:tbl><w:tr><w:tc><w:p><w:r><w:t>甲</w:t></w:r></w:p></w:tc>'
f'<w:tc><w:p><w:r><w:t>乙</w:t></w:r></w:p></w:tc></w:tr></w:tbl>'
)
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as archive:
archive.writestr("word/document.xml", f"<w:document {ns}><w:body>{body}</w:body></w:document>")
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()