qiaoqichaoandClaude Code
669e969c41
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>
271 lines
15 KiB
Python
271 lines
15 KiB
Python
"""Bounded attachment parsing and field provenance; no macros/formula execution."""
|
||
import base64
|
||
import csv
|
||
import io
|
||
import json
|
||
import re
|
||
import zipfile
|
||
from hashlib import sha256
|
||
from pathlib import Path
|
||
from xml.etree import ElementTree as ET
|
||
|
||
from services.assistant_store import uid, now
|
||
from services.skill_executor import number
|
||
|
||
FIELDS = {
|
||
"facility_amount": "facility.amount", "lease_asset_value": "lease.total_value",
|
||
"operating_cashflow": "financial.operating_cashflow", "financial_liabilities": "financial.total_liabilities",
|
||
"due_diligence_liabilities": "due_diligence.liabilities", "duplicate_asset_count": "lease.duplicate_count",
|
||
"collection_dependency": "collection.dependency", "collection_evidence": "collection.evidence",
|
||
"financial_assets": "financial.total_assets", "credit_overdue_count": "credit.overdue_count",
|
||
"credit_report_date": "credit.report_date",
|
||
"facility_term": "facility.term",
|
||
}
|
||
NUMERIC = set(FIELDS.values()) - {"collection.dependency", "collection.evidence", "credit.report_date"}
|
||
LIMIT = 10 * 1024 * 1024
|
||
|
||
|
||
def scalar(field, value):
|
||
if value is None or value in ("", "null", "缺失"):
|
||
return None
|
||
if field in NUMERIC:
|
||
result = number(value)
|
||
if field in ("lease.duplicate_count", "credit.overdue_count") and (result < 0 or result != result.to_integral_value()):
|
||
raise ValueError("数量必须为非负整数")
|
||
if field in ("facility.amount", "lease.total_value") and result <= 0:
|
||
raise ValueError("融资金额和租赁物价值必须为正数")
|
||
return str(result)
|
||
if field == "collection.dependency":
|
||
if value not in (True, False, "true", "false", "是", "否"):
|
||
raise ValueError("回款依赖须为布尔值")
|
||
return value in (True, "true", "是")
|
||
return str(value)[:4000]
|
||
|
||
|
||
def safe_zip(content):
|
||
archive = zipfile.ZipFile(io.BytesIO(content))
|
||
if len(archive.infolist()) > 2000 or sum(i.file_size for i in archive.infolist()) > 40 * LIMIT:
|
||
archive.close()
|
||
raise ValueError("压缩包解压尺寸超过限制")
|
||
return archive
|
||
|
||
|
||
def parse(content, filename):
|
||
if not content or len(content) > LIMIT:
|
||
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, 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, source_locator=sheet_loc(stem, f"第{index}行"))
|
||
elif suffix == ".docx":
|
||
with safe_zip(content) as archive:
|
||
root = ET.fromstring(archive.read("word/document.xml"))
|
||
ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
|
||
para, table = 0, 0
|
||
for element in root.find("w:body", ns):
|
||
if element.tag.endswith("}p"):
|
||
para += 1
|
||
# 正文段落不是行列可寻址的表格，归 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,
|
||
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:
|
||
strings = []
|
||
if "xl/sharedStrings.xml" in archive.namelist():
|
||
strings = ["".join(t.text or "" for t in s.findall(".//x:t", ns)) for s in ET.fromstring(archive.read("xl/sharedStrings.xml"))]
|
||
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():
|
||
raise ValueError("工作表引用无效")
|
||
for row in ET.fromstring(archive.read(path)).findall("x:sheetData/x:row", ns):
|
||
cells = []
|
||
for cell in row.findall("x:c", ns):
|
||
raw = cell.findtext("x:v", default="", namespaces=ns)
|
||
kind = cell.get("t")
|
||
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_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_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"), source_locator=pdf(page["number"]))
|
||
if not page.get("text", "").strip():
|
||
warnings.append("第" + str(page["number"]) + "页未提取到文字，请启用 OCR 或补充可读材料")
|
||
else:
|
||
raise ValueError("支持 TXT、CSV、DOCX、XLSX、PDF、PNG、JPG；不执行宏或外链")
|
||
if sum(len(c["text"]) for c in chunks) > 300000:
|
||
raise ValueError("材料文字超过 30 万字，请拆分")
|
||
return chunks, warnings
|
||
|
||
|
||
def material(project_id, filename, content, actor, version=1):
|
||
filename = Path(filename.replace("\\", "/")).name[:180]
|
||
try:
|
||
chunks, warnings = parse(content, filename)
|
||
except (OSError, zipfile.BadZipFile, ET.ParseError, UnicodeError, csv.Error, IndexError, TypeError) as exc:
|
||
raise ValueError("材料内容损坏、编码不支持或格式不匹配，请检查原文件") from exc
|
||
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": []}
|
||
for chunk in body["chunks"]:
|
||
chunk["evidence_id"] = uid("evdoc")
|
||
asset_rows = [c for c in chunks if c.get("cells")]
|
||
if asset_rows and asset_rows[0]["cells"][:2] == ["asset_id", "value"]:
|
||
assets, duplicates, conflicting = {}, 0, False
|
||
for row in asset_rows[1:]:
|
||
cells = row["cells"]
|
||
if len(cells) < 2 or not cells[0].strip():
|
||
body["warnings"].append(row["locator"] + "缺少设备编号或价值")
|
||
continue
|
||
value = number(cells[1])
|
||
if value <= 0:
|
||
raise ValueError("设备价值必须大于零")
|
||
if cells[0] in assets:
|
||
duplicates += 1
|
||
if assets[cells[0]] != value:
|
||
conflicting = True
|
||
else:
|
||
assets[cells[0]] = value
|
||
metadata = {"document_id": body["id"], "source": filename, "locator": "设备清单去重计算；" + "、".join(c["locator"] for c in asset_rows[1:]),
|
||
"currency": "CNY", "unit": "万元", "period": "2025", "scope": "合并", "formula_version": "unique-asset-id-sum@1.0.0"}
|
||
# Only the explicit demo schema below establishes these dimensions. Other asset files require confirmation.
|
||
if asset_rows[0]["cells"][2:] != ["currency=CNY", "unit=万元", "period=2025", "scope=合并"]:
|
||
metadata.update(currency=None, unit=None, period=None, scope=None)
|
||
body["warnings"].append("设备清单未声明币种、单位、期间或口径；人工确认前不得计算覆盖率")
|
||
body["candidates"].extend([{**metadata, "id": uid("candidate"), "field": "lease.duplicate_count", "value": str(duplicates), "unit": "count"},
|
||
{**metadata, "id": uid("candidate"), "field": "lease.total_value", "value": None if conflicting or not assets else str(sum(assets.values()))}])
|
||
if conflicting:
|
||
body["warnings"].append("同一设备编号价值不一致，未计算总价值")
|
||
if body["warnings"]:
|
||
body["status"] = "partial"
|
||
for chunk in chunks:
|
||
cells = chunk.get("cells") or re.split(r"[:：=]", chunk["text"], maxsplit=1)
|
||
if len(cells) < 2:
|
||
continue
|
||
field = FIELDS.get(cells[0].strip(), cells[0].strip())
|
||
if field not in FIELDS.values():
|
||
continue
|
||
try:
|
||
value = scalar(field, cells[1].strip())
|
||
body["candidates"].append({"id": uid("candidate"), "field": field, "value": value,
|
||
"document_id": body["id"], "locator": chunk["locator"], "source": filename,
|
||
"currency": cells[2] if len(cells) > 2 else None, "unit": cells[3] if len(cells) > 3 else None,
|
||
"period": cells[4] if len(cells) > 4 else None, "scope": cells[5] if len(cells) > 5 else None})
|
||
except ValueError as exc:
|
||
body["warnings"].append(chunk["locator"] + ": " + str(exc))
|
||
body["status"] = "partial"
|
||
return body
|
||
|
||
|
||
def demo_material(project):
|
||
stream = io.StringIO(newline="")
|
||
writer = csv.writer(stream)
|
||
writer.writerow(["field", "value", "currency", "unit", "period", "scope"])
|
||
for field, value in project["facts"].items():
|
||
name = FIELDS.get(field, field)
|
||
if name not in FIELDS.values():
|
||
continue
|
||
writer.writerow([name, "null" if value is None else (json.dumps(value) if isinstance(value, bool) else value),
|
||
"CNY" if name in NUMERIC else "", "count" if name.endswith("count") else ("万元" if name in NUMERIC else ""), "2025", "合并"])
|
||
return material(project["id"], project["id"] + "演示源数据.csv", stream.getvalue().encode("utf-8-sig"), "demo-fixture")
|
||
|
||
|
||
def demo_assets(project):
|
||
"""Real parseable list; duplicate values are excluded from the amount numerator."""
|
||
amount = number(project["facts"]["lease_asset_value"])
|
||
count = int(number(project["facts"].get("duplicate_asset_count", 0)))
|
||
stream = io.StringIO(newline="")
|
||
writer = csv.writer(stream)
|
||
writer.writerow(["asset_id", "value", "currency=CNY", "unit=万元", "period=2025", "scope=合并"])
|
||
writer.writerow(["ASSET-001", str(amount)])
|
||
for _ in range(count):
|
||
writer.writerow(["ASSET-001", str(amount)])
|
||
return material(project["id"], project["id"] + "设备清单.csv", stream.getvalue().encode("utf-8-sig"), "demo-fixture")
|
||
|
||
|
||
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"}
|
||
if candidate["field"] in money_fields and value is not None and candidate.get("unit"):
|
||
from api.risk_agent_alignment import SCALE
|
||
if candidate["unit"] not in SCALE:
|
||
raise ValueError("不支持的金额单位，请确认元、万元或亿元")
|
||
evidence.update(raw_value=value, raw_unit=candidate["unit"], unit="万元")
|
||
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"])
|
||
project.setdefault("facts", {})[old] = value
|
||
return evidence
|