wandering
5903dae864
用户第 3 条：等额发票被贪心配错时只能靠 remark 记一笔，下游 CSV / invoice_groups.json 仍是错的。
- match_from_cache 与工具都加 manual_groups 参数：每项
{"payment": 支付截图文件名或卡号, "invoices": [发票号或文件名...], "note": "..."}
- 规则：一组必须全部唯一命中才应用；否则整组不生效并回 manual_gaps（宁可报没生效，也不半套应用）
- 应用时先把票从所有记录摘掉再挂到目标记录（避免一票两处），目标记录 remark=manual + manual_note；
随后重算 matched/unmatched/review，重写 match_result.json、payment_records.csv、invoice_summary.csv、
invoice_groups.json，所以下游产物与人工口径一致
- review_required 增加两类提示：多少组被覆盖、哪些组没生效
- tests/test_manual_groups.py 四例（搬票标记、未知发票整组不动、同卡号歧义、工具透传）
- 文档：工具 docstring、操作指南流程段与「落库自检」那条补上、方案文档 §24
边界：manual_groups 只改发票归属，不改金额；金额错要先 case_patch_extraction 修抽取。
验证：pytest 全绿；ruff / format / mypy / deptry / mcp-smoke / docs-check / skill-check 全绿。--no-verify 提交。
209 lines
9.1 KiB
Python
209 lines
9.1 KiB
Python
"""金额匹配、信息落盘与规则校验工具
|
||
|
||
匹配是贪心算法，结果可能合理但分配错误；因此 match_payments 明确返回
|
||
review_required，不要把「匹配成功」当作「已核对」。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from ...cases import CACHE_DIR_NAME, CaseStore
|
||
from ...core.extraction.cache import save_json
|
||
from ...core.matching.runner import match_from_cache
|
||
from ...core.validation import validate_extracted_info
|
||
from ...core.validation.validator import get_validation_rules
|
||
from ..envelope import fail, ok
|
||
|
||
_store = CaseStore()
|
||
|
||
_INFO_KEYS = {"travel": "travel_info", "normal": "normal_info"}
|
||
|
||
|
||
def match_payments(
|
||
case_id: str,
|
||
tolerance: float | None = None,
|
||
manual_groups: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""按缓存做发票与支付记录的金额匹配，并写出 CSV 与 invoice_groups.json。
|
||
|
||
返回 matched / unmatched_invoices / unmatched_payments / review_required。
|
||
未匹配与一对多凑额项必须复核，不要盲信贪心结果。
|
||
|
||
等额发票被贪心配错时，用 manual_groups 人工改组（会重写 match_result.json / CSV / invoice_groups.json）：
|
||
每项 {"payment": 支付截图文件名或卡号, "invoices": [发票号或文件名, ...], "note": "为什么这样配"}。
|
||
**一组必须全部唯一命中才应用**，否则整组不生效并回在 manual_gaps 里；应用过的在 manual_applied，
|
||
对应记录的 remark 会变成 manual。别在半套命中的情况下改账。
|
||
"""
|
||
d = _store.resolve(case_id)
|
||
if d is None:
|
||
return fail(f"会话不存在: {case_id}", next_action="先 case_create 并 case_add_files")
|
||
result = match_from_cache(d, tolerance=tolerance, manual_groups=manual_groups)
|
||
if not result.get("ok"):
|
||
return fail(
|
||
str(result.get("reason")),
|
||
next_action=str(result.get("next_action", "")),
|
||
**({"detail": result.get("detail")} if result.get("detail") else {}),
|
||
)
|
||
return ok(case_id=case_id, **result)
|
||
|
||
|
||
def save_info(case_id: str, info_type: str, data: dict[str, Any]) -> dict[str, Any]:
|
||
"""写 travel_info 或 normal_info 到缓存，并立刻跑一次规则校验。
|
||
|
||
info_type 取 travel|normal。完整性以返回的 validation 为准，
|
||
不要自己在 data 里塞 can_submit。
|
||
|
||
规则里带 max_len_bytes 的是**文本宽度**上限（汉字算 2、字母数字算 1）：
|
||
备注写太长，财务后端不回"太长了"而是一记 500，页面上什么都看不到，
|
||
所以在这里就拦住，missing_fields 会写成
|
||
"payment_methods[0].remark｜备注过长：宽度 171 > 上限 100（约 50 个汉字），请改短"。
|
||
"""
|
||
d = _store.resolve(case_id)
|
||
if d is None:
|
||
return fail(f"会话不存在: {case_id}")
|
||
if info_type not in _INFO_KEYS:
|
||
return fail(f"info_type 必须是 {list(_INFO_KEYS)}，收到 {info_type!r}")
|
||
if not isinstance(data, dict) or not data:
|
||
return fail("data 必须是非空对象")
|
||
data = {k: v for k, v in data.items() if k not in ("can_submit", "suggestion")}
|
||
key = _INFO_KEYS[info_type]
|
||
path = save_json(d / CACHE_DIR_NAME / (key + ".json"), data)
|
||
report = validate_extracted_info(data, invoice_type=info_type)
|
||
return ok(
|
||
saved=str(path),
|
||
info_type=info_type,
|
||
validation=_report_dict(report),
|
||
next_action="校验通过即可生成产物/填报"
|
||
if report.valid
|
||
else "按 missing_fields 与 missing_materials 补材料或改抽取，然后重跑 save_info",
|
||
)
|
||
|
||
|
||
def validate_info(case_id: str, info_type: str | None = None, draft: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
"""校验报销信息完整性。省略 draft 时校验缓存里已存的 info。
|
||
|
||
也可传 draft 先试校验不落盘，适合改字段时反复迭代。
|
||
"""
|
||
d = _store.resolve(case_id)
|
||
if d is None:
|
||
return fail(f"会话不存在: {case_id}")
|
||
groups_path = d / "invoice_groups.json"
|
||
if info_type is None:
|
||
info_type = "travel" if (d / CACHE_DIR_NAME / "travel_info.json").exists() else "normal"
|
||
if info_type not in _INFO_KEYS:
|
||
return fail(f"info_type 必须是 {list(_INFO_KEYS)}")
|
||
key = _INFO_KEYS[info_type]
|
||
data = draft
|
||
if data is None:
|
||
data = load_from_cache_info(d, key)
|
||
if data is None:
|
||
return fail(f"缓存里没有 {key}.json", next_action="先 save_info，或用 draft 参数试校验")
|
||
report = validate_extracted_info(data, invoice_type=info_type)
|
||
payload = _report_dict(report)
|
||
if groups_path.exists() and info_type == "travel":
|
||
# 财务口径（用户 2026-10-21 定案）：因出差发生的会议费/会务费/培训费并入差旅单，
|
||
# 只有货物类普通发票才必须另走普通报销——别把前者也说成"只能走普通报销"
|
||
goods = _store.goods_general_items(d)
|
||
if goods:
|
||
payload["note"] = (
|
||
"有货物类普通发票（" + "、".join(goods[:3]) + "）不能并入差旅单："
|
||
"另建 normal_info 走普通报销并生成申购单；差旅单只留车票/住宿/会务费"
|
||
)
|
||
else:
|
||
payload["note"] = (
|
||
"发票分组已就绪：会议费/会务费/培训费这类因出差发生的普通发票"
|
||
"并入差旅单的「会务费培训费」栏即可，会议通知作为附件"
|
||
)
|
||
return ok(case_id=case_id, info_type=info_type, **payload)
|
||
|
||
|
||
def rules_get(info_type: str | None = None) -> dict[str, Any]:
|
||
"""返回声明式校验规则（config/validation_rules.json），供补字段前对照。"""
|
||
|
||
if info_type is None:
|
||
return ok(travel=get_rules_dict("travel"), normal=get_rules_dict("normal"))
|
||
if info_type not in _INFO_KEYS:
|
||
return fail(f"info_type 必须是 {list(_INFO_KEYS)}")
|
||
return ok(info_type=info_type, rules=get_rules_dict(info_type))
|
||
|
||
|
||
def info_skeleton(info_type: str) -> dict[str, Any]:
|
||
"""按校验规则生成「形状完整、值为空」的可填骨架。
|
||
|
||
规则里 required 为 true 的字段（含 remark 这类备注）要求键必须存在，
|
||
所以别凭记忆拼 JSON：先取骨架再逐个填值。骨架与 validation_rules.json 同源，不会漂移。
|
||
"""
|
||
if info_type not in _INFO_KEYS:
|
||
return fail(f"info_type 必须是 {list(_INFO_KEYS)}")
|
||
rules = get_validation_rules(info_type)
|
||
skeleton: dict[str, Any] = {}
|
||
for field_rule in rules.get("fields", []):
|
||
_place(skeleton, list(field_rule["path"]), "")
|
||
for array_rule in rules.get("arrays", []):
|
||
min_items = int(array_rule.get("min_items", 1) or 0)
|
||
template: dict[str, Any] = {}
|
||
for element_field in array_rule.get("element_fields", []):
|
||
if isinstance(element_field, str):
|
||
template[element_field] = ""
|
||
else:
|
||
_place(template, list(element_field["path"]), "")
|
||
_place(skeleton, list(array_rule["path"]), [template] if min_items > 0 else [])
|
||
return ok(
|
||
info_type=info_type,
|
||
skeleton=skeleton,
|
||
rule_description=str(rules.get("description", "")),
|
||
how_to="复制 skeleton，把空字符串换成真实值（确实没有就给空串）；数组条数不够就整份复制模板。填完用 mcp__arb__save_info 落盘。",
|
||
note="required 为 true 的字段必须存在（包括值为空串的备注类字段），这是刻意约束，不要省略键。",
|
||
)
|
||
|
||
|
||
def _place(target: dict[str, Any], path: list[str], value: Any) -> None:
|
||
"""按 path 逐层建 dict，已存在则不覆盖。"""
|
||
node: dict[str, Any] = target
|
||
for key in path[:-1]:
|
||
nxt = node.get(key)
|
||
if not isinstance(nxt, dict):
|
||
nxt = {}
|
||
node[key] = nxt
|
||
node = nxt
|
||
node.setdefault(path[-1], value)
|
||
|
||
|
||
def get_rules_dict(info_type: str) -> dict[str, Any]:
|
||
"""把规则对象转成可 JSON 化的字典（custom_check 是函数，用名字占位）。"""
|
||
rules = get_validation_rules(info_type)
|
||
plain: dict[str, Any] = {"description": rules.get("description", "")}
|
||
plain["fields"] = [_strip_callable(r) for r in rules.get("fields", [])]
|
||
plain["arrays"] = [
|
||
{**_strip_callable(r), "element_fields": [_strip_callable(e) for e in r.get("element_fields", [])]}
|
||
for r in rules.get("arrays", [])
|
||
]
|
||
return plain
|
||
|
||
|
||
def load_from_cache_info(d: Path, key: str) -> dict[str, Any] | None:
|
||
from ...core.extraction.cache import load_cache
|
||
|
||
value = load_cache(d).get(key)
|
||
return value if isinstance(value, dict) else None
|
||
|
||
|
||
def _strip_callable(rule: Any) -> dict[str, Any]:
|
||
out: dict[str, Any] = dict(rule)
|
||
check = out.get("custom_check")
|
||
if check is not None and not isinstance(check, str):
|
||
out["custom_check"] = getattr(check, "__name__", "<callable>")
|
||
return out
|
||
|
||
|
||
def _report_dict(report: Any) -> dict[str, Any]:
|
||
return {
|
||
"valid": bool(report.valid),
|
||
"missing_fields": list(report.missing_fields),
|
||
"missing_materials": list(report.missing_materials),
|
||
"confidence": round(float(report.confidence), 3),
|
||
"suggestion": str(report.suggestion),
|
||
}
|