diff --git a/.agents/docs/plans/MCP化改造方案-2026-10-19.md b/.agents/docs/plans/MCP化改造方案-2026-10-19.md index 13b2627..2773b30 100644 --- a/.agents/docs/plans/MCP化改造方案-2026-10-19.md +++ b/.agents/docs/plans/MCP化改造方案-2026-10-19.md @@ -708,3 +708,22 @@ capture 只出现在 camera 段、脚本里不许再出现 `setAttribute`、两 同类教训连着三条(capture 挂错、accept 被路由到相机、按钮名过度承诺)都指向一件事: 移动端网页的能力边界由设备与 WebView 决定,文案只能描述**实测到的行为**,不能描述规范里的应然。 + +## 23. 落库后自检:每步 POST 完回读页面(2026-10-21) + +用户第 2 条:add_travel_items 点完确定不核对表尾合计,于是那次 500 现场看着像「脚本没点着按钮」。 +现在每步 POST 完都回读一次页面,差值写进 warnings。 + +- `BaseBot.self_check_step(kind, step, info)`:按表头关键词找本步的表格(复用 attachment_rows 验证过的路子), + 读数据行数与表尾 ¥ 合计,跟 info 期望值比对;details/payment/subsidy 三步覆盖,basic/attachments 跳过 + (附件步本来就有 attachment_rows 的行数告警) +- 期望值来源:travel details = 三类明细行数与金额之和;normal details = 1 行 + total_amount; + payment = 笔数与金额之和;subsidy 只比行数(合计由页面按标准算,不拿 info 去比) +- **读不到表格不装成功**:返回 readable=False + 「本步没有核对过,别当成已核实」的 warning; + 静默通过正是上一轮附件假成功的成因 +- `browser_fill` 每步返回 self_check(原始读数),把 notes 并进 warnings; +- `tests/test_self_check.py` 六例:行数与合计都匹配、行数+合计不一致、支付匹配、找不到表、 + 无表格步骤跳过、补助只比行数 + +还没做的:自检的选择器只在 FakePage 上验证过,**真实页面待联机确认**(尤其是 payment/subsidy 两步的表头关键词), +与第 4 条(乘车日期选择器)一起,等下一次联机填报时一并确认。 diff --git a/docs/MCP.md b/docs/MCP.md index 855767b..e84f994 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -392,6 +392,9 @@ cases// 提交前会量这一步要 POST 的自由文本宽度(汉字算 2、字母数字算 1,上限 100, 普通报销说明 40):财务后端对超宽只回 500、页面上任何提示都没有,弹窗停在原地 看着像"按钮没点着",所以宁可在这里拦下来,失败时 too_long 会列出超长字段。 + + 每步 POST 完会做一次**落库后自检**:回读页面上的数据行数与表尾合计,跟 info 的期望值比对, + 不一致就放进 warnings(返回里的 self_check 是原始读数);读不到表格时也会明确说「本步没有核对过」。 | 参数 | 类型 | 必填 | 默认 | 说明 | |------|------|------|------|------| diff --git a/src/infra/browser/base.py b/src/infra/browser/base.py index 07fce55..b3021b6 100644 --- a/src/infra/browser/base.py +++ b/src/infra/browser/base.py @@ -677,6 +677,149 @@ class BaseBot: self.confirm_attachment() return text + # --------------------------------------------------------------- + # 落库后自检:读回页面上的行数与合计,跟 info 的期望值比对 + # --------------------------------------------------------------- + # 按表头关键词找表,跟 attachment_rows 同一条路子(那条路联机验证过)。 + # 只覆盖有表格的步骤;读不出来就如实说"没核对过",绝不静默当成功。 + STEP_TABLE_HINTS: dict[str, tuple[str, ...]] = { + "details": ("费用类别", "出行方式", "经济事项", "报销金额", "单据数"), + "payment": ("支付方式", "支付金额", "开户行"), + "subsidy": ("补助", "伙食"), + } + # 行内既不是表头也不是金额的文字,出现在这些列里说明是表头或空状态行 + HEADER_WORDS = ( + "序号", + "费用类别", + "出行方式", + "经济事项", + "支付方式", + "补助", + "附件文件名", + "没有找到匹配", + "暂无数据", + ) + + def _find_step_table(self, step: str) -> Any: + hints = self.STEP_TABLE_HINTS.get(step, ()) + if not hints: + return None + try: + tables = self.page.query_selector_all("table") + except Exception: + return None + for table in tables: + try: + head = " ".join((table.inner_text() or "").split()) + except Exception: + continue + if head and any(hint in head for hint in hints): + return table + return None + + def _read_step_table(self, step: str) -> dict[str, Any]: + """读本步表格:数据行数与表尾 ¥ 合计;读不出就 readable=False。""" + table = self._find_step_table(step) + if table is None: + return {"readable": False, "reason": "没找到本步的表格"} + rows = 0 + total = 0.0 + has_amount = False + try: + trs = table.query_selector_all("tr") + except Exception: + return {"readable": False, "reason": "表格行读不出来"} + for tr in trs: + try: + text = " ".join((tr.inner_text() or "").split()) + except Exception: + continue + if not text: + continue + if any(word in text for word in self.HEADER_WORDS): + continue + amounts = re.findall(r"[¥¥]\s*([\d,]+(?:\.\d+)?)", text) + if amounts: + for raw in amounts: + total += float(raw.replace(",", "")) + has_amount = True + continue + if len(tr.query_selector_all("td")) >= 2: + rows += 1 + elif len(text.split()) >= 3: + rows += 1 + return {"readable": True, "rows": rows, "amount": round(total, 2) if has_amount else None} + + @staticmethod + def _row_amount(row: dict[str, Any]) -> float: + """info 行里的金额字段名各版本不完全统一,按常见键依次取。""" + for key in ("amount", "total_amount", "fee", "money", "报销金额", "费用金额", "补助金额"): + raw = row.get(key) + if raw in (None, ""): + continue + try: + return float(str(raw).replace(",", "")) + except (TypeError, ValueError): + continue + return 0.0 + + def self_check_step(self, kind: str, step: str, info: dict[str, Any]) -> dict[str, Any]: + """填报后回读页面:行数/合计与 info 比对,差值变成 warning。 + + 背景:add_travel_items 点完确定不核对表尾合计,那次 500 看起来像"脚本没点着按钮"。 + 现在每步 POST 完都回读一次;读不到表格也要报"没核对过",不许装成核实了。 + """ + if step not in self.STEP_TABLE_HINTS: + return {"step": step, "skipped": "这一步没有可核对的表格"} + expected_rows = 0 + expected_amount: float | None = None + if step == "details" and kind == "travel": + details = info.get("reimbursement_details", {}) or {} + rows: list[dict[str, Any]] = [] + for key in ("transport_fee", "hotel_fee", "conference_fee"): + value = details.get(key) or [] + if isinstance(value, list): + rows.extend(r for r in value if isinstance(r, dict)) + expected_rows = len(rows) + expected_amount = round(sum(self._row_amount(r) for r in rows), 2) + elif step == "details": + details = info.get("reimbursement_details", {}) or {} + expected_rows = 1 + try: + expected_amount = round(float(details.get("total_amount", 0) or 0), 2) + except (TypeError, ValueError): + expected_amount = None + elif step == "payment": + pays = [r for r in (info.get("payment_methods") or []) if isinstance(r, dict)] + expected_rows = len(pays) + expected_amount = round(sum(self._row_amount(r) for r in pays), 2) + elif step == "subsidy": + subs = [r for r in (info.get("subsidy_list") or []) if isinstance(r, dict)] + expected_rows = len(subs) + expected_amount = None # 补助合计由页面按标准算,不拿 info 去比 + actual = self._read_step_table(step) + if not actual.get("readable"): + note = f"{step} 落库自检读不到页面表格({actual.get('reason')}):本步没有核对过,别当成已核实" + return {"step": step, "readable": False, "expected_rows": expected_rows, "notes": [note]} + diffs: list[str] = [] + if actual["rows"] != expected_rows: + diffs.append(f"行数 页面 {actual['rows']} / info {expected_rows}") + if expected_amount is not None and actual.get("amount") is not None: + if abs(actual["amount"] - expected_amount) > 0.01: + diffs.append(f"合计 页面 {actual['amount']:.2f} / info {expected_amount:.2f}") + notes = [] + if diffs: + notes.append(f"{step} 落库自检不一致({'; '.join(diffs)}):先 browser_page_state 看截图再决定是否重填") + return { + "step": step, + "readable": True, + "rows": actual["rows"], + "amount": actual.get("amount"), + "expected_rows": expected_rows, + "expected_amount": expected_amount, + "notes": notes, + } + def attachment_rows(self) -> int: """附件清单里已有的附件条数。 diff --git a/src/mcp/tools/browser.py b/src/mcp/tools/browser.py index d0abb4e..56bbbe0 100644 --- a/src/mcp/tools/browser.py +++ b/src/mcp/tools/browser.py @@ -303,6 +303,9 @@ def browser_fill(case_id: str, step: str) -> dict[str, Any]: 提交前会量这一步要 POST 的自由文本宽度(汉字算 2、字母数字算 1,上限 100, 普通报销说明 40):财务后端对超宽只回 500、页面上任何提示都没有,弹窗停在原地 看着像"按钮没点着",所以宁可在这里拦下来,失败时 too_long 会列出超长字段。 + + 每步 POST 完会做一次**落库后自检**:回读页面上的数据行数与表尾合计,跟 info 的期望值比对, + 不一致就放进 warnings(返回里的 self_check 是原始读数);读不到表格时也会明确说「本步没有核对过」。 """ d = _case_dir(case_id) ctx = _require_open(case_id) @@ -348,7 +351,14 @@ def browser_fill(case_id: str, step: str) -> dict[str, Any]: def _fill() -> dict[str, Any]: bot = ctx["bot"] filled = _apply_step(bot, kind, step, info) - return {"filled": filled, "screenshot": bot._screenshot(f"{step}_done"), "url": bot.page.url} + # 落库后自检:读回页面行数与合计(附件步另有 attachment_rows 的行数告警) + check = bot.self_check_step(kind, step, info) + return { + "filled": filled, + "screenshot": bot._screenshot(f"{step}_done"), + "url": bot.page.url, + "self_check": check, + } try: result = actor.run_browser(_fill) @@ -369,6 +379,9 @@ def browser_fill(case_id: str, step: str) -> dict[str, Any]: step=step, screenshot=shot, ) + self_check = result.get("self_check") or {} + if isinstance(self_check, dict): + notes.extend(str(n) for n in self_check.get("notes") or []) actor.mark_step(case_id, step) actor.touch(d, str(result.get("url", ""))) done_after = list(ctx.get("steps", [])) @@ -381,6 +394,7 @@ def browser_fill(case_id: str, step: str) -> dict[str, Any]: screenshot=result.get("screenshot", ""), done=done_after, remaining=remaining, + self_check=self_check, warnings=notes, next_action=( f"mcp__arb__browser_fill(step='{remaining[0]}')" if remaining else "mcp__arb__browser_page_state" diff --git a/tests/test_mcp_browser.py b/tests/test_mcp_browser.py index 44e3099..512c468 100644 --- a/tests/test_mcp_browser.py +++ b/tests/test_mcp_browser.py @@ -245,6 +245,11 @@ FakeBot._validation_hint = BaseBot._validation_hint # type: ignore[method-assig FakeBot.confirm_attachment = BaseBot.confirm_attachment # type: ignore[method-assign] FakeBot.upload_attachment = BaseBot.upload_attachment # type: ignore[method-assign] FakeBot.attachment_rows = BaseBot.attachment_rows # type: ignore[method-assign] +FakeBot.STEP_TABLE_HINTS = BaseBot.STEP_TABLE_HINTS +FakeBot._row_amount = staticmethod(BaseBot._row_amount) # type: ignore[assignment] +FakeBot.self_check_step = BaseBot.self_check_step # type: ignore[method-assign] +FakeBot._read_step_table = BaseBot._read_step_table # type: ignore[method-assign] +FakeBot._find_step_table = BaseBot._find_step_table # type: ignore[method-assign] FakeBot.row_is_ready = BaseBot.row_is_ready # type: ignore[method-assign] FakeBot._wait_any = BaseBot._wait_any # type: ignore[method-assign] FakeBot.row_cells = BaseBot.row_cells # type: ignore[method-assign] @@ -416,8 +421,10 @@ def test_fill_walks_steps_and_reports_remaining(env: dict[str, Any]) -> None: assert first["done"] == ["basic"] assert first["remaining"] == ["details", "payment", "attachments"] second = browser.browser_fill(env["id"], "details") + assert second["ok"] is True, second assert second["filled"] == {"单据数": 3, "总金额": 464.0} third = browser.browser_fill(env["id"], "payment") + assert third["ok"] is True, third assert third["filled"] == {"支付笔数": 2} assert third["remaining"] == ["attachments"] state = browser.browser_page_state(env["id"]) diff --git a/tests/test_self_check.py b/tests/test_self_check.py new file mode 100644 index 0000000..faf5210 --- /dev/null +++ b/tests/test_self_check.py @@ -0,0 +1,91 @@ +"""落库后自检:读回页面行数/合计与 info 比对,差值给 warning;读不到就如实说没核对。""" + +from typing import Any + +from src.infra.browser.base import BaseBot + + +class _Row: + def __init__(self, text: str, cells: int = 4) -> None: + self._text = text + self._cells = cells + + def inner_text(self) -> str: + return self._text + + def query_selector_all(self, selector: str) -> list[Any]: + return [object()] * self._cells if selector == "td" else [] + + +class _Table: + def __init__(self, header: str, rows: list[str]) -> None: + self._header = header + self._rows = [_Row(r) for r in rows] + + def inner_text(self) -> str: + return self._header + + def query_selector_all(self, selector: str) -> list[Any]: + return list(self._rows) if selector == "tr" else [] + + +class _Page: + def __init__(self, tables: list[_Table]) -> None: + self.tables = tables + + def query_selector_all(self, selector: str) -> list[Any]: + return list(self.tables) if selector == "table" else [] + + +class _Bot(BaseBot): + def __init__(self, tables: list[_Table]) -> None: + self.page = _Page(tables) # type: ignore[assignment] + + +def test_travel_details_match_has_no_warning() -> None: + bot = _Bot([_Table("费用类别 出行方式 报销金额", ["交通费 火车 233", "¥233.00"])]) + info = {"reimbursement_details": {"transport_fee": [{"amount": "233.00"}]}} + check = bot.self_check_step("travel", "details", info) + assert check["readable"] is True + assert check["rows"] == 1 and check["amount"] == 233.0 + assert check["notes"] == [] + + +def test_travel_details_mismatch_reports_rows_and_total() -> None: + bot = _Bot([_Table("费用类别 出行方式 报销金额", ["交通费 火车 233", "¥233.00"])]) + info = { + "reimbursement_details": { + "transport_fee": [{"amount": "233.00"}, {"amount": "233.00"}], + } + } + notes = bot.self_check_step("travel", "details", info)["notes"] + assert notes and "行数 页面 1 / info 2" in notes[0] + assert "合计 页面 233.00 / info 466.00" in notes[0] + + +def test_payment_matches() -> None: + bot = _Bot([_Table("支付方式 支付金额 开户行", ["工行公务卡 231.00 工商银行", "¥231.00"])]) + info = {"payment_methods": [{"amount": "231.00"}]} + check = bot.self_check_step("travel", "payment", info) + assert check["readable"] is True and check["notes"] == [] + + +def test_missing_table_says_not_checked() -> None: + bot = _Bot([]) + check = bot.self_check_step("travel", "payment", {"payment_methods": []}) + assert check["readable"] is False + assert "没有核对过" in check["notes"][0] + + +def test_steps_without_table_are_skipped() -> None: + bot = _Bot([]) + assert "skipped" in bot.self_check_step("travel", "basic", {}) + assert "skipped" in bot.self_check_step("travel", "attachments", {}) + + +def test_subsidy_compares_rows_only() -> None: + bot = _Bot([_Table("补助清单 伙食补助", ["张国庆 540"])]) + info = {"subsidy_list": [{"person_name": "张国庆"}]} + check = bot.self_check_step("travel", "subsidy", info) + assert check["readable"] is True and check["notes"] == [] + assert check["expected_amount"] is None