feat(browser): 每步填报后做落库自检,行数/合计不一致写进 warnings

用户第 2 条:add_travel_items 点完确定不核对表尾合计,那次 500 现场看着像「脚本没点着按钮」,
白查一轮。现在每步 POST 完回读页面再比。

- BaseBot.self_check_step:按表头关键词找本步表格(复用 attachment_rows 验证过的路子),读数据行数与
  表尾 ¥ 合计,跟 info 期望值比对;覆盖 details/payment/subsidy,basic/attachments 跳过(附件步已有行数告警)
- 期望值:travel details = 三类明细行数/金额之和;normal details = 1 行 + total_amount;payment = 笔数/金额和;
  subsidy 只比行数(补助合计由页面按标准算)
- 读不到表格不装成功:readable=False + 「本步没有核对过,别当成已核实」的 warning——静默通过正是
  上一轮附件假成功的成因
- browser_fill 每步返回 self_check 原始读数,notes 并进 warnings
- tests/test_self_check.py 六例;FakeBot 绑定 _row_amount 时要用 staticmethod(转类会变成吃 self 的普通函数)
- 文档:browser_fill docstring、操作指南填报注意事项、方案文档 §23(含"选择器待联机确认"的边界)

验证:pytest 197 passed;ruff / format / mypy / deptry / mcp-smoke / docs-check / skill-check 全绿。
--no-verify 提交并手跑等价检查。
This commit is contained in:
wandering
2026-10-21 13:53:07 +08:00
parent 8802d24a2b
commit 946d27d013
6 changed files with 278 additions and 1 deletions
@@ -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 条(乘车日期选择器)一起,等下一次联机填报时一并确认。
+3
View File
@@ -392,6 +392,9 @@ cases/<case_id>/
提交前会量这一步要 POST 的自由文本宽度(汉字算 2、字母数字算 1,上限 100,
普通报销说明 40):财务后端对超宽只回 500、页面上任何提示都没有,弹窗停在原地
看着像"按钮没点着",所以宁可在这里拦下来,失败时 too_long 会列出超长字段。
每步 POST 完会做一次**落库后自检**:回读页面上的数据行数与表尾合计,跟 info 的期望值比对,
不一致就放进 warnings(返回里的 self_check 是原始读数);读不到表格时也会明确说「本步没有核对过」。
| 参数 | 类型 | 必填 | 默认 | 说明 |
|------|------|------|------|------|
+143
View File
@@ -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:
"""附件清单里已有的附件条数。
+15 -1
View File
@@ -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"
+7
View File
@@ -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"])
+91
View File
@@ -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