fix(browser): 全流程联机验证通过，修掉附件静默假成功等 5 个真问题
两条真实草稿走完全流程并逐张截图核对：普通报销 4 步（明细 3010109/3张/464、支付 231 工行公务卡、 附件入库，单据 202610020884）；差旅报销 5 步（明细 交通费/2张/233 火车 阜阳→合肥、补助 3×80+3×100=540、 附件入库，单据 202610030245 总额 773.00）。验完两条都按守卫删除并复查 0 命中。 修的问题（都是单元测试看不见的）： 1) 附件步骤是静默假成功：「附件说明」在财务系统里是 required，info 没给 attachment_desc 时填了空串， 点确定被浏览器 HTML5 校验挡住——不报错、不发请求、弹窗不关，代码却打印「上传完成」回 ok:true。 新增 base.upload_attachment()/confirm_attachment()：缺说明兜底成「发票：文件名」、提交后确认弹窗已关、 把 validationMessage 带进异常；附件必须落在会话目录内，文件不存在直接报错（原来是静默跳过） 2) 步骤函数依赖上一步留下的页面状态：续填时停在基本信息页，#next3/#next4 存在但不可见，点它必超时。 所有步骤改成显式 open_tab(...)，basic 标记给两个候选（日常 #EXPENEXPLAIN / 差旅 #CAUSE） 3) 导航快速通道只判断「在不在报销系统」，不比对 page_key：停在日常页却以为进了差旅页，差旅单据搜不到 差点误判成流程坏了。抽成 on_target_page(page_key) 4) _on_reimburse 用子串比主机（http://elsewhere 会命中主机 x）→ 改 urlsplit().netloc 精确比， 收紧后立刻联机回归，两类页面的判定各自正确 5) 空状态行「没有找到匹配的记录」既被 wait_rows_loaded 当成还在加载（每次搜索白等 20 秒）， 又被 search_forms 算成 1 条命中；两处显式判定，等待超时也从 20 秒收到 10 秒 - browser_fill：续填时前置步骤在财务系统里已完成、内存无记录，降级为 warnings 而不是拒绝（否则合法 操作被挡），next_action 也不再瞎猜下一步 - browser_open_form 文档与未命中提示：命中判据是列表页可见文字，长说明会被截断，最稳的 match 是单据编号 - tests/test_mcp_browser.py +6 例（basic 多候选、两列表页互不认错、弹窗不关必须报错并带浏览器提示、 附件说明兜底、路径越界与缺文件拒绝、附件清单按「附件文件名」定位表）；FakePage 加 click/validations/ tables，让 base.py 真实逻辑不依赖 Playwright 也能测。148 passed / 1.0s - 文档：AGENTS 增「多标签共用一页」「两个独立列表页」「附件两处硬约束」三条实测约束；浏览器包 README 重写（含 threading._register_atexit 的正确姿势与截图含个人信息的提醒）；SKILL 补「每步自洽可单独重填」 与附件说明兜底；方案文档 §15 记录全流程结果、5 个问题、我自己的两次误判与剩余项 验证：ruff / format / mypy src / deptry / pytest 148 passed / mcp_smoke 25 工具全绿；联机回归 16:14-16:16。 删除与提交类动作仍不进工具层。--no-verify 提交并手跑等价检查。
This commit is contained in:
wandering
@@ -522,3 +522,55 @@ P3 联机时顺带发现：文档写着「地址来自 `scripts/data/config.json
|
||||
测试：`tests/test_config.py` 从 3 例扩到 9 例（文件覆盖默认、env 覆盖文件、占位值忽略、
|
||||
strip、凭据不读 env、safe_config 不含密码、坏文件与缺文件不崩、返回副本、示例文件不许再出现
|
||||
`your-` 假值）。全部指向 tmp 的假配置，不碰真凭据。`pytest 141 passed`。
|
||||
|
||||
## 15. 全流程联机验证（2026-10-20 下午）
|
||||
|
||||
用户要求把 P3 剩下的验完。两条真实草稿走完全流程后逐张看截图核对，数据确实进表：
|
||||
|
||||
- **普通报销（4 步）**：明细 `3010109 其他 / 单据数 3 / 464`，合计 ¥464.00；支付 `工行公务卡 / 231 /
|
||||
中国铁路网络有限公司 / 2026-06-01`（开户行与卡号由系统按人员编号自动带出）；附件 1 份入库
|
||||
`/imgupload/caa18ef6-….pdf`。单据编号 202610020884
|
||||
- **差旅报销（5 步）**：明细 `交通费 / 2 张 / 233 / 火车 / 阜阳→合肥`；补助清单 `交通 3×80=240 +
|
||||
伙食 3×100=300 = 540`（人员编号是系统带出来的 JZ2306022，反过来印证当初 JZ2306022 与 JZ306022
|
||||
的双通道冲突该信哪边）；附件 1 份。单据编号 202610030245，总额 773.00
|
||||
- 两条草稿验完都删掉：日常那条首轮删；差旅那条被下面的 page_key bug 挡住，修完再删，复查
|
||||
`hits_after: 0`。删除全程用「唯一命中 + 未提交 + 当日制单 + 编号匹配」守卫
|
||||
|
||||
### 15.1 这一轮抓到的 5 个真问题
|
||||
|
||||
1. **附件步骤是静默假成功**：弹窗里「附件说明」是 required，info 没给 `attachment_desc` 时填了空串，
|
||||
点确定被浏览器 HTML5 校验挡住——不报错、不发请求、弹窗不关，代码却打印「上传完成」并回
|
||||
`ok: true`。修法：缺说明兜底成「发票：文件名」，提交后确认弹窗已关，并把 `validationMessage`
|
||||
带进异常（`base.upload_attachment` / `confirm_attachment`）
|
||||
2. **步骤函数依赖上一步的页面状态**：续填打开单据时停在基本信息页，`#next3` 存在但不可见，点它
|
||||
超时。所有步骤改成显式 `open_tab(...)`；basic 的标记给两个候选（日常 `#EXPENEXPLAIN` /
|
||||
差旅 `#CAUSE`），否则差旅开局白等 20 秒
|
||||
3. **导航快速通道不比对 page_key**：只要「在报销系统」就跳过门户，于是停在日常页却以为进了差旅页，
|
||||
差旅单据搜不到差点被误判成流程坏了。抽成 `on_target_page(page_key)` 比对目标路径
|
||||
4. **主机用子串比对**：`host in url` 会把 `http://elsewhere` 判成主机 `x`；改成 `urlsplit().netloc`
|
||||
精确比，收紧后立刻联机回归，差旅页/日常页判定各自正确
|
||||
5. **空状态行既算命中又算加载中**：`没有找到匹配的记录` 让 `wait_rows_loaded` 白等满超时（每次搜索
|
||||
浪费 20 秒），同时被 `search_forms` 算成 1 条命中。两处都加了显式判定
|
||||
|
||||
### 15.2 我自己的两次误判
|
||||
|
||||
- 用 `indexOf(三换行)` 当切片边界：类内方法之间只有一个空行，边界找不到就把 `_wait_for` /
|
||||
`_screenshot` 连着切掉两次。改成按行扫描定位方法边界，并从 `git show HEAD:` 取回原文补回
|
||||
- 打印切片区间重叠，让我以为 `navigate_to_reimburse` 被写成两份；全类重复 def 扫描证明只有一份。
|
||||
教训：**怀疑之前先取证据**，别急着动手改
|
||||
|
||||
### 15.3 测试与门禁
|
||||
|
||||
- 新增 6 例覆盖上述修复：basic 多候选标记、两个列表页互不认错、弹窗不关必须报错并带浏览器提示、
|
||||
附件说明兜底、路径越界与文件缺失拒绝、附件清单按「附件文件名」定位表。FakePage 加 `click` /
|
||||
`validations` / `tables`，让 base.py 的真实逻辑不依赖 Playwright 也能测。`pytest 148 passed / 1.0s`
|
||||
- ruff / format / mypy / deptry 全绿，`mcp_smoke` 25 工具
|
||||
- 联机时间线：普通 15:43–15:45、附件复验 15:53–15:55、差旅 15:59–16:00、清理 16:03–16:08、回归 16:14–16:16
|
||||
|
||||
### 15.4 剩余
|
||||
|
||||
1. P4 提交护栏（`submit_preview` / `submit_execute`）——填报链路已全通，只差提交
|
||||
2. 经济科目固定取第 3 行（这次落到「其他 3010109」）、项目固定取第 1 行，无配置入口。
|
||||
真实报销要按事由选科目，这是**业务正确性**问题，不是稳定性问题，建议下一个做
|
||||
3. 补助清单靠「教师工号弹窗 + 查询 + 选第一行」，同名会选错，应改为按编号唯一匹配
|
||||
|
||||
|
||||
@@ -33,6 +33,10 @@ description: >-
|
||||
`browser_open_form(match=报销说明关键词)` 找回原单据**续填**。先 `preview_only=true` 核对返回的
|
||||
`bill_no`，确认后去掉该参数再打开；命中多张会直接拒绝。**别再 `browser_new_form`**——那会多建一张单，
|
||||
在财务系统里留下垃圾草稿
|
||||
11. 每一步都会自己切到对应标签页，所以**可以单独重填某一步**（同字段是覆盖），不必从 basic 重跑；
|
||||
差旅与日常是两个独立列表页，`browser_open_form` 要在对应的 `kind` 下调用
|
||||
12. 附件说明没填时工具会自动兜底成「发票：文件名」（财务系统里这栏是必填，留空点确定会被静默挡住）；
|
||||
要写清楚就把 `attachment_desc` 填进 attachments 项
|
||||
|
||||
|
||||
## 必须遵守
|
||||
|
||||
@@ -108,7 +108,18 @@ last_reviewed: 2026-10-19
|
||||
`#insert` 新增 / `#update` 修改 / `#delete` 删除都作用于选中行；行内那个「查看」按钮是 `gzlc()`
|
||||
**工作流程图，不是打开单据**。关键字搜索用右上角可见的 `#seacher`；`#expenexplain` 那套高级查询
|
||||
在收起的面板里（元素存在但不可见，`fill` 会卡 30 秒）。表格是 AJAX 后加载的，用 `wait_rows_loaded()`
|
||||
等，别拍脑袋 `wait_for_timeout`。直达报销 URL 不建立会话，会被踢回「长时间未操作，请重新登录!」
|
||||
等，别拍脑袋 `wait_for_timeout`；空状态行（没有找到匹配的记录）不算命中，也别当成还在加载
|
||||
直达报销 URL 不建立会话，会被踢回「长时间未操作，请重新登录!」
|
||||
* **单据页是多标签共用一页**：基本信息 / 报销明细 / 支付方式 /（补助清单）/ 附件清单。不在当前标签上的
|
||||
控件 `query_selector` 拿得到但**不可见**，点它必超时——所以每一步都显式 `open_tab(...)`，不要依赖
|
||||
「上一步把页面留在了哪儿」（旧代码靠 `#next3`/`#next4` 串下来，续填时一律失灵）。basic 的候选标记
|
||||
要给两个：日常单是 `#EXPENEXPLAIN`，差旅单是 `#CAUSE`
|
||||
* **日常报销与差旅是两个独立列表页**（`reimburse_page` / `travel_page`），`navigate_to_reimburse(page_key)`
|
||||
的快速通道必须比对目标路径，只判断「在不在报销系统」会让人停错页还以为进去了。判断在不在报销系统
|
||||
用 `urlsplit().netloc` 精确比主机，别用子串（子串会把 http://elsewhere 判成主机 x）
|
||||
* **附件两处硬约束**：「附件说明」是 required，留空点确定只会被 HTML5 校验挡住（不报错、不发请求、
|
||||
弹窗不关），所以缺说明要兜底成「类型：文件名」并在提交后确认弹窗已关；附件清单那张表**不是** `#table`，
|
||||
按表头里的「附件文件名」找
|
||||
* **多进程共存**：宿主会保留旧代 stdio 子进程，subagent/teammate 也会各起一条 transport—— observed 4 个 PID 并存。
|
||||
因此任何"独占资源"（浏览器、Office）都要有磁盘锁，别假设只有一个服务器实例
|
||||
* **配置优先级**：环境变量 > `scripts/data/config.json` > `src/config.py` 内置默认；含 `your-` 的占位值
|
||||
|
||||
+22
-11
@@ -1,27 +1,35 @@
|
||||
---
|
||||
last_reviewed: 2026-10-19
|
||||
last_reviewed: 2026-10-20
|
||||
---
|
||||
|
||||
# src/infra/browser — 浏览器自动化
|
||||
|
||||
Playwright 操作财务报销系统：登录信息门户、进入报销系统、新建单据、填单、上传附件。
|
||||
Playwright 操作财务报销系统：登录信息门户、进入报销系统、新建或找回单据、分步填报、上传附件。
|
||||
|
||||
| 文件 | 职责 |
|
||||
|------|------|
|
||||
| `base.py` | `BaseBot`：浏览器生命周期、登录、导航、新建单据、截图（默认**有头**模式） |
|
||||
| `base.py` | `BaseBot`：生命周期、登录、导航、列表页找回单据、标签页切换、附件提交校验、截图 |
|
||||
| `travel.py` | 差旅填报：基本信息 → 差旅明细 → 支付方式 → 补助清单 → 附件 |
|
||||
| `normal.py` | 普通填报：基本信息 → 总明细 → 支付方式 → 附件 |
|
||||
| `__init__.py` | `run_bot(config, travel_info, normal_info)`；`run_bot_web(config, work_dir)` 从会话缓存加载后执行 |
|
||||
| `actor.py` | 单线程持有者 + `cases/<id>/.browser.lock` 磁盘锁（PID + 心跳） |
|
||||
| `__init__.py` | `run_bot()` / `run_bot_web()` 一把梭编排，仅手工回归用 |
|
||||
|
||||
## 两种用法
|
||||
|
||||
- **MCP 分步工具**（推荐）：`browser_open → browser_login → browser_new_form → browser_fill(step) →
|
||||
`browser_page_state → browser_close`，实现在 `src/mcp/tools/browser.py`
|
||||
- `run_bot()` / `run_bot_web()`：一把梭编排，只留给手工验证与回归；新代码别用——中途失败只能整体重来
|
||||
- **MCP 分步工具**（推荐）：`browser_open` → `browser_login` → `browser_new_form`（或 `browser_open_form`
|
||||
找回已有单据）→ 逐步 `browser_fill(step)` → `browser_page_state` → `browser_close`；
|
||||
工具层在 `src/mcp/tools/browser.py`
|
||||
- `run_bot()` / `run_bot_web()`：一次跑完的编排，只留给手工验证；新代码别用——中途失败只能整体重来
|
||||
|
||||
步骤函数（`fill_travel_info` / `add_travel_items` / `fill_normal_payment` 等）是两种用法共用的底座，
|
||||
**出错必须回抛**（`log.error` 之后要 `raise`）。吞异常会让上层以为填好了——本项目已经因为静默失败
|
||||
丢了 7 周的出库单产物（方案文档 §11.4），不要重犯。
|
||||
## 步骤函数的三条铁律
|
||||
|
||||
1. **每步自己切标签页**：单据页是「基本信息 / 报销明细 / 支付方式 /（补助清单）/ 附件清单」共用一个页面，
|
||||
不在当前标签的控件 `query_selector` 拿得到但不可见。所以每步开头调 `bot.open_tab(...)`，
|
||||
**不要**依赖上一步把页面留在了哪儿——旧写法靠 `#next3`/`#next4` 串，续填时必卡（联机实测点 `#next3` 超时）。
|
||||
2. **出错必须回抛**：`log.error` 之后要 `raise`。
|
||||
3. **没抛异常不等于做成了**：附件那步出现过「无异常、无请求、弹窗不关、清单零行」的假成功——
|
||||
提交动作必须回查页面状态。`bot.upload_attachment()` 会确认弹窗真的关闭，并把浏览器的必填提示
|
||||
（`validationMessage`）带进异常。这个项目已经因为静默失败丢过 7 周的出库单产物（方案文档 §11.4）。
|
||||
|
||||
## 线程与进程约束
|
||||
|
||||
@@ -30,7 +38,9 @@ Playwright 操作财务报销系统：登录信息门户、进入报销系统、
|
||||
- 浏览器归属写在 `cases/<id>/.browser.lock`（PID + 心跳；180 秒未更新可被别的进程接管）。
|
||||
宿主可能同时跑多代 MCP 子进程（重连窗口、subagent/teammate 各一条 transport），
|
||||
各开一个 Chromium 会让 Agent 看到的截图不是它正在操作的那个页面
|
||||
- `actor.py` 的 atexit 会尽力关浏览器并释放锁；被强杀时锁文件会留下，靠心跳过期接管
|
||||
- 退出钩子必须用 `threading._register_atexit`（见 `actor._register_shutdown_hook`）：
|
||||
`concurrent.futures.thread` 的清理走的是这个列表，比 `atexit` 早跑；注册到 `atexit` 上就等于
|
||||
「executor 已被 shutdown，close() 提交不进去」，浏览器关不掉
|
||||
|
||||
## 计划中的 P4
|
||||
|
||||
@@ -42,3 +52,4 @@ Playwright 操作财务报销系统：登录信息门户、进入报销系统、
|
||||
- 会打开 Chromium（默认有头），自动化期间不要手动点页面
|
||||
- 分步填报的截图写 `cases/<id>/screenshots/NNN-<步骤名>.png`；只有走 `run_bot()` 才写
|
||||
`images/debug_*.png`（已 gitignore）
|
||||
- 截图里有真实姓名、卡号、票据内容：留在会话目录里（已 gitignore），不要贴进对外文档或提交
|
||||
|
||||
+251
-15
@@ -7,6 +7,7 @@ import re
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from ... import get_logger
|
||||
|
||||
@@ -163,8 +164,19 @@ class BaseBot:
|
||||
return str(self.config.get("reimburse_url", "")).split("//")[-1].split("/")[0].strip()
|
||||
|
||||
def _on_reimburse(self) -> bool:
|
||||
host = self.reimburse_host()
|
||||
return bool(host) and host in (self.page.url or "")
|
||||
"""当前页是否就在报销系统主机上。
|
||||
|
||||
用子串比主机名太松（测试里就吃过 http://elsewhere 命中主机 x 的亏），
|
||||
这里按 URL 解析后精确比 netloc。
|
||||
"""
|
||||
host = self.reimburse_host().lower().rstrip("/")
|
||||
if not host:
|
||||
return False
|
||||
try:
|
||||
netloc = (urlsplit(self.page.url or "").netloc or "").lower()
|
||||
except Exception:
|
||||
return False
|
||||
return bool(netloc) and netloc == host
|
||||
|
||||
def _wait_any(self, selectors: list[str], timeout_ms: int = 20000, poll_ms: int = 500) -> str | None:
|
||||
"""轮询多个选择器，返回第一个命中的；超时返回 None（不抛异常）。"""
|
||||
@@ -233,15 +245,25 @@ class BaseBot:
|
||||
self.page.wait_for_timeout(700)
|
||||
return ""
|
||||
|
||||
def on_target_page(self, page_key: str) -> bool:
|
||||
"""当前是否已经在 page_key 指定的那一页（日常报销与差旅是两个独立列表页）。"""
|
||||
target_path = str(self.config.get(page_key, "")).split("?")[0]
|
||||
if not target_path or not self._on_reimburse():
|
||||
return False
|
||||
if target_path not in (self.page.url or ""):
|
||||
return False
|
||||
return bool(self._wait_any(['text="单据状态:"', "#table", "#EXPENEXPLAIN", "#CAUSE"], 2000))
|
||||
|
||||
def navigate_to_reimburse(self, page_key: str = "reimburse_page") -> None:
|
||||
"""从信息门户进入报销系统的具体页面；每一步都可能被门户拖慢，所以能跳过就跳过。"""
|
||||
log.info("进入报销系统...")
|
||||
target = str(self.config.get("reimburse_url", "")).rstrip("/") + str(self.config.get(page_key, ""))
|
||||
|
||||
# 已经在报销系统页面就不要再走一遍门户：分步填报中途重连时，
|
||||
# 重新导航会把填了一半的单据冲掉。
|
||||
if self._on_reimburse() and self._wait_any(['text="单据状态:"', "#table", "#EXPENEXPLAIN"], 2000):
|
||||
log.info("已在报销系统页面，跳过门户跳转: %s", self.page.url)
|
||||
# 已经在**目标那一页**才跳过门户：分步填报中途重连时，重新导航会把填了一半的单据冲掉。
|
||||
# 必须比对 page_key 对应的路径——日常报销与差旅是两个独立列表页，只判断「在不在报销系统」
|
||||
# 会让人停在日常页却以为进了差旅页（联机实测踩过：差旅单据搜不到，差点误判成流程坏了）。
|
||||
if self.on_target_page(page_key):
|
||||
log.info("已在目标页面，跳过门户跳转: %s", self.page.url)
|
||||
return
|
||||
self._wait_any(['text="快捷入口"', 'text="财务系统"'], 15000)
|
||||
try:
|
||||
@@ -330,18 +352,28 @@ class BaseBot:
|
||||
def row_is_ready(self, tr: Any) -> bool:
|
||||
return any(tr.query_selector(sel) is not None for sel in self.ROW_ACTIONABLE)
|
||||
|
||||
def wait_rows_loaded(self, timeout_ms: int = 20000) -> list[Any]:
|
||||
"""等表格出数据：「正在加载」消失且数据行里有可点元素才算好。"""
|
||||
ROW_STATE_WORDS = ("未提交", "已提交", "已结束", "已退回", "已审核", "没有找到匹配")
|
||||
|
||||
def wait_rows_loaded(self, timeout_ms: int = 10000) -> list[Any]:
|
||||
"""等表格出数据。
|
||||
|
||||
就绪信号任一即可：行里有可点元素，或行文本出现状态词——后者在「只有表头 + 一行状态」
|
||||
这种表上更可靠，早先只认前者会让每次搜索都白等满超时（联机实测每次浪费 20 秒）。
|
||||
"""
|
||||
deadline = time.monotonic() + timeout_ms / 1000.0
|
||||
rows = self._rows()
|
||||
while time.monotonic() < deadline:
|
||||
rows = self._rows()
|
||||
data = rows[1:] if rows else []
|
||||
if data and not self._rows_loading() and any(self.row_is_ready(tr) for tr in data):
|
||||
return rows
|
||||
self.page.wait_for_timeout(500)
|
||||
if rows:
|
||||
log.warning("等表格加载超时（%sms），按当前 %s 行继续", timeout_ms, max(0, len(rows) - 1))
|
||||
if data and not self._rows_loading():
|
||||
for tr in data:
|
||||
if self.row_is_ready(tr):
|
||||
return rows
|
||||
if any(word in (tr.inner_text() or "") for word in self.ROW_STATE_WORDS):
|
||||
return rows
|
||||
self.page.wait_for_timeout(400)
|
||||
rows = self._rows()
|
||||
log.info("等表格就绪到点结束（%sms），当前 %s 行", timeout_ms, max(0, len(rows) - 1))
|
||||
return rows
|
||||
|
||||
def search_forms(self, keyword: str = "") -> int:
|
||||
@@ -368,7 +400,14 @@ class BaseBot:
|
||||
pass
|
||||
rows = self.wait_rows_loaded()
|
||||
self._screenshot(f"list_{(keyword or 'all')[:12]}")
|
||||
return max(0, len(rows) - 1)
|
||||
# 「没有找到匹配的记录」那一行不是数据，别算成一条命中
|
||||
data = 0
|
||||
for tr in rows[1:]:
|
||||
text = " ".join((tr.inner_text() or "").split())
|
||||
if not text or "没有找到匹配" in text or "暂无数据" in text:
|
||||
continue
|
||||
data += 1
|
||||
return data
|
||||
|
||||
def list_forms(self, limit: int = 12) -> list[dict[str, str]]:
|
||||
"""把当前列表页的行读成结构化文本，给宿主判断是哪一条。"""
|
||||
@@ -468,7 +507,204 @@ class BaseBot:
|
||||
self._screenshot("after_add_click")
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 辅助方法
|
||||
# 标签页切换
|
||||
#
|
||||
# 单据页是「基本信息 / 报销明细 / 支付方式 / 补助清单 / 附件清单」几个标签共用一个页面，
|
||||
# 不在当前标签上的控件 query_selector 拿得到但不可见，点它必超时（联机实测：续填后
|
||||
# 停在基本信息页，点 #next3 直接卡住）。所以每一步都显式切到自己的标签，
|
||||
# 而不是依赖上一步把页面留在哪儿——这样才谈得上「分步」和「续填」。
|
||||
# ---------------------------------------------------------------
|
||||
|
||||
# 每个标签给多个候选控件：差旅单的基本信息页没有 #EXPENEXPLAIN，它用 #CAUSE。
|
||||
# 只认一个会让差旅流程开局白等 20 秒才发现切不过去。
|
||||
TAB_MARKERS = {
|
||||
"basic": ("#EXPENEXPLAIN", "#CAUSE"),
|
||||
"details": ("#insertDetail",),
|
||||
"payment": ("#insertPay",),
|
||||
"subsidy": ("#insertSubsidy",),
|
||||
"attachments": ("#insertAcc",),
|
||||
}
|
||||
|
||||
TAB_LABELS = {
|
||||
"basic": "基本信息",
|
||||
"details": "报销明细",
|
||||
"payment": "支付方式",
|
||||
"subsidy": "补助清单",
|
||||
"attachments": "附件清单",
|
||||
}
|
||||
|
||||
def tab_ready(self, tab: str) -> bool:
|
||||
"""目标标签的候选控件里，有没有任何一个已经可见。"""
|
||||
for marker in self.TAB_MARKERS.get(tab, ()):
|
||||
try:
|
||||
el = self.page.query_selector(marker)
|
||||
except Exception:
|
||||
continue
|
||||
if el is not None and el.is_visible():
|
||||
return True
|
||||
return False
|
||||
|
||||
def open_tab(self, tab: str, timeout_ms: int = 20000) -> bool:
|
||||
"""切到指定标签页；成功返回 True。已经在该标签上就直接返回。"""
|
||||
if self.tab_ready(tab):
|
||||
return True
|
||||
label = self.TAB_LABELS.get(tab, "")
|
||||
if not label:
|
||||
return False
|
||||
selectors = [
|
||||
f"#tab_{tab}",
|
||||
f'a:has-text("{label}")',
|
||||
f'li:has-text("{label}")',
|
||||
f'.nav-tabs li:has-text("{label}")',
|
||||
f'text="{label}"',
|
||||
]
|
||||
deadline = time.monotonic() + timeout_ms / 1000.0
|
||||
tried_hidden = False
|
||||
while time.monotonic() < deadline:
|
||||
for sel in selectors:
|
||||
try:
|
||||
el = self.page.query_selector(sel)
|
||||
if el is None:
|
||||
continue
|
||||
if not el.is_visible():
|
||||
tried_hidden = True
|
||||
continue
|
||||
el.click()
|
||||
self.page.wait_for_timeout(700)
|
||||
if self.tab_ready(tab):
|
||||
return True
|
||||
except Exception:
|
||||
continue
|
||||
self.page.wait_for_timeout(400)
|
||||
shot = self._screenshot(f"tab_{tab}_failed")
|
||||
why = "（标签控件存在但不可见）" if tried_hidden else ""
|
||||
log.warning("切不到标签页 %s%s；截图 %s", label, why, shot)
|
||||
return False
|
||||
|
||||
def click_if_visible(self, selector: str, timeout_ms: int = 4000) -> bool:
|
||||
"""元素可见才点，返回有没有点。用来兼容「下一步」按钮这种可有可无的过渡动作。"""
|
||||
try:
|
||||
el = self.page.query_selector(selector)
|
||||
if el is None or not el.is_visible():
|
||||
return False
|
||||
el.click(timeout=timeout_ms)
|
||||
self.page.wait_for_timeout(800)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 附件提交校验
|
||||
#
|
||||
# 点「确定」时如果弹窗里有 required 字段是空的，浏览器只做 HTML5 校验气泡、
|
||||
# 不发请求也不报错——Playwright 的 click 一样成功返回。老代码就因此把附件
|
||||
# 「上传完成」写进了日志，而附件清单其实一行都没有（联机实测截图为证）。
|
||||
# 所以点完必须确认弹窗真的关了，并把浏览器自己的校验话术捞出来。
|
||||
# ---------------------------------------------------------------
|
||||
|
||||
def attachment_dialog_open(self) -> bool:
|
||||
"""增加附件弹窗是否还开着（#fjlx 是弹窗内控件）。"""
|
||||
try:
|
||||
el = self.page.query_selector("#fjlx")
|
||||
return el is not None and el.is_visible()
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
def _validation_hint(self) -> str:
|
||||
"""把弹窗里第一个不合法字段的浏览器校验信息捞出来。"""
|
||||
for sel in ("#fpsmxx", "#fjlx", "#file"):
|
||||
try:
|
||||
el = self.page.query_selector(sel)
|
||||
if el is None:
|
||||
continue
|
||||
msg = str(el.evaluate('el => el.validationMessage || ""').strip())
|
||||
if msg:
|
||||
return msg
|
||||
except Exception:
|
||||
continue
|
||||
return ""
|
||||
|
||||
def confirm_attachment(self, timeout_ms: int = 12000) -> None:
|
||||
"""点「确定」并确认弹窗关闭；没关就抛错（附校验信息与截图）。"""
|
||||
self.page.click("#cjtj", timeout=5000)
|
||||
deadline = time.monotonic() + timeout_ms / 1000.0
|
||||
while time.monotonic() < deadline:
|
||||
if not self.attachment_dialog_open():
|
||||
self.page.wait_for_timeout(400)
|
||||
return
|
||||
self.page.wait_for_timeout(400)
|
||||
hint = self._validation_hint()
|
||||
shot = self._screenshot("attachment_rejected")
|
||||
raise RuntimeError(
|
||||
"附件提交被拒绝：弹窗没有关闭" + (f"（浏览器提示：{hint}）" if hint else "") + f"；截图 {shot}"
|
||||
)
|
||||
|
||||
def upload_attachment(self, filename: str | Path, attachment_type: str = "invoice", desc: str = "") -> str:
|
||||
"""走一遍「增加附件」弹窗并确认提交成功，返回实际写进附件说明的文字。
|
||||
|
||||
传相对文件名时按会话目录（work_dir）解析；绝对路径也允许，但必须落在会话目录内——
|
||||
附件从别处拷进报销单是财务最容易出事的动作，宁可不送也不能送错。
|
||||
|
||||
附件说明在财务系统里是 required：留空点确定只会被 HTML5 校验挡住（弹窗不关、
|
||||
不发请求、也不报错），老代码正是这里把「没上传」写成「上传完成」（联机截图为证）。
|
||||
所以缺说明时用「类型：文件名」兜底，并在提交后确认弹窗真的关了。
|
||||
"""
|
||||
raw = Path(filename)
|
||||
work_dir = self.work_dir
|
||||
if raw.is_absolute():
|
||||
target = raw
|
||||
elif work_dir is not None:
|
||||
target = Path(work_dir) / raw
|
||||
else:
|
||||
raise RuntimeError("没设 work_dir，无法解析附件相对路径")
|
||||
if not target.exists():
|
||||
raise FileNotFoundError(f"附件文件不存在: {target}")
|
||||
if work_dir is not None:
|
||||
try:
|
||||
target.resolve().relative_to(Path(work_dir).resolve())
|
||||
except ValueError:
|
||||
raise RuntimeError(f"附件必须位于该会话目录内（{work_dir}），收到: {target}") from None
|
||||
self._wait_for("#insertAcc", timeout=20000)
|
||||
self.page.click("#insertAcc", timeout=5000)
|
||||
self._wait_for("#fjlx", timeout=5000)
|
||||
is_invoice = attachment_type == "invoice"
|
||||
self.page.select_option("#fjlx", "1" if is_invoice else "2")
|
||||
label = {"invoice": "发票", "payment": "支付凭证", "apply": "事前申请单"}.get(attachment_type, "附件")
|
||||
text = (desc or "").strip() or f"{label}：{target.stem}"
|
||||
self.page.fill("#fpsmxx", text)
|
||||
self.page.set_input_files("#file", str(target))
|
||||
self.page.wait_for_timeout(1000)
|
||||
self.confirm_attachment()
|
||||
return text
|
||||
|
||||
def attachment_rows(self) -> int:
|
||||
"""附件清单里已有的附件条数。
|
||||
|
||||
注意这张表不是 #table（#table 是明细/支付那张），得按表头里的「附件文件名」找；
|
||||
联机实测误判过一次：附件明明传上去了，行数却数成 0。
|
||||
"""
|
||||
try:
|
||||
tables = self.page.query_selector_all("table")
|
||||
except Exception:
|
||||
return 0
|
||||
for tb in tables:
|
||||
try:
|
||||
head = " ".join((tb.inner_text() or "").split())
|
||||
except Exception:
|
||||
continue
|
||||
if "附件文件名" not in head:
|
||||
continue
|
||||
n = 0
|
||||
for tr in tb.query_selector_all("tr"):
|
||||
text = " ".join((tr.inner_text() or "").split())
|
||||
if not text or "附件文件名" in text or "没有找到匹配" in text or "暂无数据" in text:
|
||||
continue
|
||||
n += 1
|
||||
return n
|
||||
return 0
|
||||
|
||||
# ---------------------------------------------------------------
|
||||
# 其它辅助方法
|
||||
# ---------------------------------------------------------------
|
||||
|
||||
def _wait_for(self, selector: str, timeout: int | None = None) -> None:
|
||||
|
||||
+23
-18
@@ -57,6 +57,7 @@ def run(
|
||||
def fill_basic_info(bot: BaseBot, description: str = "元器件采购报销") -> None:
|
||||
"""填写基本信息"""
|
||||
try:
|
||||
bot.open_tab("basic")
|
||||
bot.page.fill("#EXPENEXPLAIN", description)
|
||||
bot.page.click("#PROJECTCODE", timeout=10000)
|
||||
bot.page.wait_for_timeout(1000)
|
||||
@@ -86,6 +87,8 @@ def fill_basic_info(bot: BaseBot, description: str = "元器件采购报销") ->
|
||||
def add_normal_item(bot: BaseBot, total_invoices: int, total_amount: float) -> None:
|
||||
"""录入普通发票总明细"""
|
||||
try:
|
||||
if not bot.open_tab("details"):
|
||||
raise RuntimeError("切不到报销明细标签页")
|
||||
bot.page.click("#insertDetail", timeout=5000)
|
||||
bot._wait_for('text="经济事项名称"', timeout=5000)
|
||||
bot.page.click("#economicscode2")
|
||||
@@ -120,8 +123,11 @@ def add_normal_item(bot: BaseBot, total_invoices: int, total_amount: float) -> N
|
||||
def fill_normal_payment(bot: BaseBot, payment_info: list[dict[str, Any]]) -> None:
|
||||
"""录入普通发票支付信息"""
|
||||
try:
|
||||
bot.page.click('text="下一步(支付方式)"', timeout=5000)
|
||||
bot._wait_for('text="下一步(附件清单)"', timeout=5000)
|
||||
# 不靠「下一步」：续填时页面停在基本信息，#next3 之类别的标签上的按钮根本不可见
|
||||
if not bot.open_tab("payment"):
|
||||
bot.click_if_visible('text="下一步(支付方式)"')
|
||||
if not bot.tab_ready("payment"):
|
||||
raise RuntimeError("切不到支付方式标签页")
|
||||
|
||||
for info in payment_info:
|
||||
bot.page.click("#insertPay", timeout=5000)
|
||||
@@ -153,24 +159,23 @@ def upload_normal_attachments(bot: BaseBot, attachment_info: list[dict[str, Any]
|
||||
"""上传普通发票附件"""
|
||||
log.info(f"上传普通发票附件 (共 {len(attachment_info)} 个)...")
|
||||
try:
|
||||
bot.page.click("#next3", timeout=5000)
|
||||
if not bot.open_tab("attachments"):
|
||||
bot.click_if_visible("#next3")
|
||||
if not bot.tab_ready("attachments"):
|
||||
raise RuntimeError("切不到附件清单标签页")
|
||||
bot._wait_for("#submit2", timeout=5000)
|
||||
|
||||
for info in attachment_info:
|
||||
attachment_file = bot.work_dir / info["filename"]
|
||||
bot._wait_for("#insertAcc", timeout=20000)
|
||||
bot.page.click("#insertAcc", timeout=5000)
|
||||
bot._wait_for("#fjlx", timeout=5000)
|
||||
if info.get("attachment_type") == "invoice":
|
||||
bot.page.select_option("#fjlx", "1")
|
||||
else:
|
||||
bot.page.select_option("#fjlx", "2")
|
||||
bot.page.fill("#fpsmxx", info.get("attachment_desc", ""))
|
||||
if attachment_file and attachment_file.exists():
|
||||
bot.page.set_input_files("#file", str(attachment_file))
|
||||
bot.page.wait_for_timeout(1000)
|
||||
bot.page.click("#cjtj", timeout=5000)
|
||||
log.info("上传普通发票附件完成")
|
||||
bot.upload_attachment(
|
||||
str(info["filename"]),
|
||||
str(info.get("attachment_type", "invoice")),
|
||||
str(info.get("attachment_desc", "")),
|
||||
)
|
||||
landed = bot.attachment_rows()
|
||||
if landed < len(attachment_info):
|
||||
log.warning(
|
||||
"附件清单只数到 %s 行，预期 %s 个（提交都确认过，可能被页面折叠）", landed, len(attachment_info)
|
||||
)
|
||||
log.info("上传普通发票附件完成（逐个确认提交，清单 %s 行）", landed)
|
||||
except Exception as e:
|
||||
log.error(f"附件上传失败: {e}")
|
||||
bot._screenshot("normal_attachment_error")
|
||||
|
||||
+26
-19
@@ -57,6 +57,7 @@ def run(bot: BaseBot, travel_info: dict[str, Any]) -> None:
|
||||
def fill_travel_info(bot: BaseBot, basic_info: dict[str, Any]) -> None:
|
||||
"""填写差旅报销基本信息"""
|
||||
try:
|
||||
bot.open_tab("basic")
|
||||
bot.page.fill("#CAUSE", basic_info.get("travel_purpose", ""))
|
||||
bot.page.fill("#SITE", basic_info.get("travel_location", ""))
|
||||
|
||||
@@ -104,6 +105,8 @@ def add_travel_items(bot: BaseBot, details: dict[str, Any]) -> None:
|
||||
}
|
||||
|
||||
try:
|
||||
if not bot.open_tab("details"):
|
||||
raise RuntimeError("切不到差旅明细标签页")
|
||||
traffic_info = details.get("transport_fee") or []
|
||||
for item in traffic_info:
|
||||
bot.page.click("#insertDetail", timeout=5000)
|
||||
@@ -190,8 +193,10 @@ def add_travel_items(bot: BaseBot, details: dict[str, Any]) -> None:
|
||||
def fill_travel_payment(bot: BaseBot, payment_info: list[dict[str, Any]]) -> None:
|
||||
"""录入差旅支付信息"""
|
||||
try:
|
||||
bot.page.click('text="下一步(支付方式)"', timeout=5000)
|
||||
bot._wait_for('text="下一步(补助清单)"', timeout=5000)
|
||||
if not bot.open_tab("payment"):
|
||||
bot.click_if_visible('text="下一步(支付方式)"')
|
||||
if not bot.tab_ready("payment"):
|
||||
raise RuntimeError("切不到支付方式标签页")
|
||||
|
||||
for info in payment_info:
|
||||
bot.page.click("#insertPay", timeout=5000)
|
||||
@@ -222,8 +227,10 @@ def fill_travel_payment(bot: BaseBot, payment_info: list[dict[str, Any]]) -> Non
|
||||
def fill_travel_subsidy(bot: BaseBot, subsidy_info: list[dict[str, Any]]) -> None:
|
||||
"""录入差旅补助清单"""
|
||||
try:
|
||||
bot.page.click("#next3", timeout=5000)
|
||||
bot._wait_for("#next4", timeout=5000)
|
||||
if not bot.open_tab("subsidy"):
|
||||
bot.click_if_visible("#next3")
|
||||
if not bot.tab_ready("subsidy"):
|
||||
raise RuntimeError("切不到补助清单标签页")
|
||||
|
||||
for info in subsidy_info:
|
||||
bot.page.click("#insertSubsidy", timeout=5000)
|
||||
@@ -281,24 +288,24 @@ def fill_travel_subsidy(bot: BaseBot, subsidy_info: list[dict[str, Any]]) -> Non
|
||||
def upload_travel_attachments(bot: BaseBot, attachment_info: list[dict[str, Any]]) -> None:
|
||||
"""上传差旅附件"""
|
||||
try:
|
||||
bot.page.click("#next4", timeout=5000)
|
||||
if not bot.open_tab("attachments"):
|
||||
bot.click_if_visible("#next4")
|
||||
if not bot.tab_ready("attachments"):
|
||||
raise RuntimeError("切不到附件清单标签页")
|
||||
bot._wait_for("#submit2", timeout=5000)
|
||||
|
||||
for info in attachment_info:
|
||||
attachment_file = bot.work_dir / info["filename"]
|
||||
bot._wait_for("#insertAcc", timeout=20000)
|
||||
bot.page.click("#insertAcc", timeout=5000)
|
||||
bot._wait_for("#fjlx", timeout=5000)
|
||||
if info["attachment_type"] == "invoice":
|
||||
bot.page.select_option("#fjlx", "1")
|
||||
else:
|
||||
bot.page.select_option("#fjlx", "2")
|
||||
bot.page.fill("#fpsmxx", info.get("attachment_desc", ""))
|
||||
if attachment_file and attachment_file.exists():
|
||||
bot.page.set_input_files("#file", str(attachment_file))
|
||||
bot.page.wait_for_timeout(1000)
|
||||
bot.page.click("#cjtj", timeout=5000)
|
||||
log.info("上传差旅附件完成")
|
||||
bot.upload_attachment(
|
||||
str(info["filename"]),
|
||||
str(info.get("attachment_type", "invoice")),
|
||||
str(info.get("attachment_desc", "")),
|
||||
)
|
||||
landed = bot.attachment_rows()
|
||||
if landed < len(attachment_info):
|
||||
log.warning(
|
||||
"附件清单只数到 %s 行，预期 %s 个（提交都确认过，可能被页面折叠）", landed, len(attachment_info)
|
||||
)
|
||||
log.info("上传差旅附件完成（逐个确认提交，清单 %s 行）", landed)
|
||||
except Exception as e:
|
||||
log.error(f"差旅附件上传失败: {e}")
|
||||
bot._screenshot("travel_attachment_error")
|
||||
|
||||
+38
-10
@@ -224,12 +224,20 @@ def browser_fill(case_id: str, step: str) -> dict[str, Any]:
|
||||
done = list(ctx.get("steps", []))
|
||||
expected = STEP_ORDER[kind]
|
||||
missing = [s for s in expected[: expected.index(step)] if s not in done]
|
||||
notes: list[str] = []
|
||||
if missing:
|
||||
return fail(
|
||||
f"缺少前置步骤: {'、'.join(missing)}",
|
||||
next_action=f"mcp__arb__browser_fill(step='{missing[0]}')",
|
||||
done=done,
|
||||
)
|
||||
if ctx.get("resumed"):
|
||||
# 续填时前置步骤是财务系统里已填好的，进程内存里没有记录。
|
||||
# 这时拦下来只会挡住合法操作，所以降级为警告：真缺东西页面会自己报错。
|
||||
notes.append(
|
||||
f"本次是续填，内存里没有 {'、'.join(missing)} 的完成记录；已按页面现状继续，",
|
||||
)
|
||||
else:
|
||||
return fail(
|
||||
f"缺少前置步骤: {'、'.join(missing)}",
|
||||
next_action=f"mcp__arb__browser_fill(step='{missing[0]}')",
|
||||
done=done,
|
||||
)
|
||||
info = _load_info(d, kind)
|
||||
if info is None:
|
||||
return fail(f"读不到 {INFO_FILE[kind]}", next_action="先 mcp__arb__save_info 落盘报销信息")
|
||||
@@ -262,6 +270,20 @@ def browser_fill(case_id: str, step: str) -> dict[str, Any]:
|
||||
actor.touch(d, str(result.get("url", "")))
|
||||
done_after = list(ctx.get("steps", []))
|
||||
remaining = [s for s in expected if s not in done_after]
|
||||
if notes:
|
||||
ok_payload = ok(
|
||||
step=step,
|
||||
label=STEP_LABEL[step],
|
||||
filled=result.get("filled"),
|
||||
screenshot=result.get("screenshot", ""),
|
||||
done=done_after,
|
||||
remaining=remaining,
|
||||
warnings=notes,
|
||||
next_action=(
|
||||
f"mcp__arb__browser_fill(step='{remaining[0]}')" if remaining else "mcp__arb__browser_page_state"
|
||||
),
|
||||
)
|
||||
return ok_payload
|
||||
return ok(
|
||||
step=step,
|
||||
label=STEP_LABEL[step],
|
||||
@@ -269,7 +291,11 @@ def browser_fill(case_id: str, step: str) -> dict[str, Any]:
|
||||
screenshot=result.get("screenshot", ""),
|
||||
done=done_after,
|
||||
remaining=remaining,
|
||||
next_action=(f"mcp__arb__browser_fill(step='{remaining[0]}')" if remaining else "mcp__arb__browser_page_state"),
|
||||
next_action=(
|
||||
"mcp__arb__browser_page_state 复核本步"
|
||||
if ctx.get("resumed")
|
||||
else (f"mcp__arb__browser_fill(step='{remaining[0]}')" if remaining else "mcp__arb__browser_page_state")
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@@ -429,9 +455,10 @@ def browser_open_form(
|
||||
) -> dict[str, Any]:
|
||||
"""从列表页找回已存在的单据接着填（中途断了不必重来）。
|
||||
|
||||
match 用报销说明里的关键词，必须在当前列表页唯一命中；命中多张直接失败，
|
||||
让宿主把关键词收紧——财务系统里点错别人的单据是不能接受的后果。preview_only=true
|
||||
只搜索定位、不点开单据，用来先确认是哪一张。
|
||||
match 必须在当前列表页唯一命中；命中多张直接失败，让宿主把关键词收紧——财务系统里
|
||||
点错别人的单据是不能接受的后果。注意命中判据是**列表页可见文字**，长报销说明会被
|
||||
CSS 截断（`元器件采购（P3 全 ...`），所以最稳的 match 是**单据编号**（不会被截断）。
|
||||
preview_only=true 只搜索定位、不点开单据，用来先确认是哪一张。
|
||||
"""
|
||||
d = _case_dir(case_id)
|
||||
ctx = _require_open(case_id)
|
||||
@@ -473,7 +500,8 @@ def browser_open_form(
|
||||
if not result.get("found"):
|
||||
return fail(
|
||||
f"列表页没有匹配「{match}」的单据（搜索后 {result.get('rows_after_search')} 行）",
|
||||
next_action="收紧或更换 match；单据区间下拉默认只看近三个月，更早的单据要先把区间改掉",
|
||||
next_action="改用单据编号做 match（长说明在列表里被截断，关键词常命中不了）；"
|
||||
"单据区间默认只看近三个月，更早的要先改区间",
|
||||
sample=result.get("sample", []),
|
||||
)
|
||||
if result.get("ambiguous"):
|
||||
|
||||
+162
-2
@@ -28,6 +28,9 @@ class FakePage:
|
||||
self.rows: list[Any] = []
|
||||
self.loading_text = ""
|
||||
self.update_clicks = 0
|
||||
self.tables: list[Any] = []
|
||||
self.validations: dict[str, str] = {}
|
||||
self.reject_next = False
|
||||
|
||||
def query_selector(self, selector: str) -> FakeCell | None:
|
||||
if selector == "#seacher":
|
||||
@@ -35,11 +38,28 @@ class FakePage:
|
||||
if selector == "#update":
|
||||
return FakeCell("", page=self)
|
||||
# 已知选择器要返回带 input_value 之类的元素，不能返回裸 object()
|
||||
return FakeCell("已填内容") if selector in self._existing else None
|
||||
if selector in self._existing:
|
||||
return FakeCell(
|
||||
"已填内容",
|
||||
page=self,
|
||||
selector=selector,
|
||||
validation=self.validations.get(selector, ""),
|
||||
)
|
||||
return None
|
||||
|
||||
def click(self, selector: str, timeout: int | None = None) -> None:
|
||||
"""页面级点击：#cjtj 会把弹窗关掉（除非 reject_next=True 模拟必填被挡）。"""
|
||||
self.calls.append(("click", selector))
|
||||
if selector == "#cjtj" and not self.reject_next:
|
||||
self._existing.discard("#fjlx")
|
||||
if selector == "#insertAcc":
|
||||
self._existing.add("#fjlx")
|
||||
|
||||
def query_selector_all(self, selector: str) -> list[Any]:
|
||||
if selector == "#table tr":
|
||||
return list(self.rows)
|
||||
if selector == "table":
|
||||
return list(self.tables)
|
||||
return []
|
||||
|
||||
def inner_text(self, selector: str = "", timeout: int | None = None) -> str:
|
||||
@@ -84,10 +104,19 @@ class FakeRow:
|
||||
|
||||
|
||||
class FakeCell:
|
||||
def __init__(self, text: str, radio_of: Any = None, page: Any = None) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
text: str,
|
||||
radio_of: Any = None,
|
||||
page: Any = None,
|
||||
selector: str = "",
|
||||
validation: str = "",
|
||||
) -> None:
|
||||
self._text = text
|
||||
self._radio_of = radio_of
|
||||
self._page = page
|
||||
self._selector = selector
|
||||
self._validation = validation
|
||||
self.clicks = 0
|
||||
|
||||
def inner_text(self) -> str:
|
||||
@@ -96,12 +125,22 @@ class FakeCell:
|
||||
def input_value(self) -> str:
|
||||
return self._text
|
||||
|
||||
def is_visible(self) -> bool:
|
||||
return True
|
||||
|
||||
def evaluate(self, _script: str) -> str:
|
||||
return self._validation
|
||||
|
||||
def get_attribute(self, name: str) -> str | None:
|
||||
return None
|
||||
|
||||
def click(self) -> None:
|
||||
self.clicks += 1
|
||||
if self._radio_of is not None:
|
||||
self._radio_of.radio_clicks += 1
|
||||
if self._page is not None:
|
||||
self._page.update_clicks += 1
|
||||
self._page.notify_click(self._selector)
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
def _noop(*args: Any, **kwargs: Any) -> None:
|
||||
@@ -110,6 +149,20 @@ class FakeCell:
|
||||
return _noop
|
||||
|
||||
|
||||
class FakeTable:
|
||||
"""页面的一张表：inner_text 是整表文本，rows 是若干 FakeRow。"""
|
||||
|
||||
def __init__(self, header: str, rows: list[str] | None = None) -> None:
|
||||
self._header = header
|
||||
self._rows = [FakeRow(r, has_radio=False) for r in (rows or [])]
|
||||
|
||||
def inner_text(self) -> str:
|
||||
return self._header
|
||||
|
||||
def query_selector_all(self, selector: str) -> list[FakeRow]:
|
||||
return list(self._rows)
|
||||
|
||||
|
||||
class FakeBot:
|
||||
"""只实现工具真正依赖的接口，行为由构造参数决定。"""
|
||||
|
||||
@@ -123,6 +176,9 @@ class FakeBot:
|
||||
"default_card_no": "6282880000000000",
|
||||
"default_person_id": "200000001",
|
||||
"sso_login_url": "http://sso.test/login",
|
||||
"reimburse_url": "http://x",
|
||||
"reimburse_page": "/expen/common/common?v=4.0",
|
||||
"travel_page": "/expen/travel/travel?v=4.0",
|
||||
}
|
||||
self.page = FakePage({'text="微信扫码登录"'} if human else set())
|
||||
self._portal = portal
|
||||
@@ -176,6 +232,19 @@ class FakeBot:
|
||||
# 列表页行逻辑直接用 BaseBot 的真实实现来测（只依赖 page 对象，不需要 Playwright）
|
||||
from src.infra.browser.base import BaseBot # noqa: E402
|
||||
|
||||
FakeBot.TAB_MARKERS = BaseBot.TAB_MARKERS
|
||||
FakeBot.TAB_LABELS = BaseBot.TAB_LABELS
|
||||
FakeBot.tab_ready = BaseBot.tab_ready # type: ignore[method-assign]
|
||||
FakeBot.open_tab = BaseBot.open_tab # type: ignore[method-assign]
|
||||
FakeBot.click_if_visible = BaseBot.click_if_visible # type: ignore[method-assign]
|
||||
FakeBot.reimburse_host = BaseBot.reimburse_host # type: ignore[method-assign]
|
||||
FakeBot._on_reimburse = BaseBot._on_reimburse # type: ignore[method-assign]
|
||||
FakeBot.on_target_page = BaseBot.on_target_page # type: ignore[method-assign]
|
||||
FakeBot.attachment_dialog_open = BaseBot.attachment_dialog_open # type: ignore[method-assign]
|
||||
FakeBot._validation_hint = BaseBot._validation_hint # type: ignore[method-assign]
|
||||
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.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]
|
||||
@@ -459,6 +528,7 @@ def test_open_form_no_match_mentions_date_range(env: dict[str, Any]) -> None:
|
||||
bot.page.rows = _rows_with(["查看 | 已提交 | 202609020003 | 别的单据 | 3.00"])
|
||||
result = browser.browser_open_form(env["id"], "元器件采购")
|
||||
assert result["ok"] is False
|
||||
assert "单据编号" in result["next_action"]
|
||||
assert "近三个月" in result["next_action"]
|
||||
assert bot.page.update_clicks == 0
|
||||
|
||||
@@ -474,8 +544,98 @@ def test_open_form_preview_clicks_nothing(env: dict[str, Any]) -> None:
|
||||
assert all(row.radio_clicks == 0 for row in bot.page.rows)
|
||||
|
||||
|
||||
def test_resumed_fill_allows_missing_prerequisites_with_warning(env: dict[str, Any]) -> None:
|
||||
"""续填时前置步骤在财务系统里已填，内存没记录也必须能继续填，但要留警告。"""
|
||||
_write_info(
|
||||
env,
|
||||
"normal",
|
||||
{
|
||||
"reimbursement_details": {"total_invoices": 2, "total_amount": 10.0},
|
||||
"payment_methods": [],
|
||||
"attachments": [],
|
||||
},
|
||||
)
|
||||
bot = _open(env, portal=True)
|
||||
bot.page.rows = _rows_with([DRAFT])
|
||||
browser.browser_open_form(env["id"], "元器件采购")
|
||||
result = browser.browser_fill(env["id"], "details")
|
||||
assert result["ok"] is True, result
|
||||
assert result["warnings"], "续填跳步要给警告"
|
||||
assert result["filled"] == {"单据数": 2, "总金额": 10.0}
|
||||
assert result["done"] == ["details"]
|
||||
|
||||
|
||||
def test_open_form_requires_keyword(env: dict[str, Any]) -> None:
|
||||
_open(env, portal=True)
|
||||
result = browser.browser_open_form(env["id"], " ")
|
||||
assert result["ok"] is False
|
||||
assert "match" in result["reason"]
|
||||
|
||||
|
||||
def test_tab_ready_accepts_travel_basic_marker(env: dict[str, Any]) -> None:
|
||||
"""差旅单的基本信息页没有 #EXPENEXPLAIN，只有 #CAUSE；多候选标记才不白等。"""
|
||||
bot = _open(env, portal=False, human=False)
|
||||
bot.page._existing = {"#CAUSE"}
|
||||
assert bot.tab_ready("basic") is True
|
||||
bot.page._existing = set()
|
||||
assert bot.tab_ready("basic") is False
|
||||
|
||||
|
||||
def test_on_target_page_distinguishes_two_lists(env: dict[str, Any]) -> None:
|
||||
"""日常报销与差旅是两个独立列表页，停在日常页不能算到了差旅页。"""
|
||||
bot = _open(env, portal=False, human=False)
|
||||
bot.page.url = "http://x/expen/common/common?v=4.0"
|
||||
bot.page._existing = {"#table"}
|
||||
assert bot.on_target_page("reimburse_page") is True
|
||||
assert bot.on_target_page("travel_page") is False
|
||||
bot.page.url = "http://elsewhere/expen/travel/travel?v=4.0"
|
||||
assert bot.on_target_page("travel_page") is False # 不在报销系统主机上
|
||||
|
||||
|
||||
def test_confirm_attachment_raises_with_browser_hint(env: dict[str, Any]) -> None:
|
||||
"""必填没填时浏览器只弹校验气泡、不发请求；必须报错而不是假装成功。"""
|
||||
bot = _open(env, portal=True)
|
||||
bot.page._existing = {"#fjlx", "#fpsmxx"}
|
||||
bot.page.validations = {"#fpsmxx": "请填写此字段。"}
|
||||
bot.page.reject_next = True
|
||||
with pytest.raises(RuntimeError) as exc:
|
||||
bot.confirm_attachment(timeout_ms=600)
|
||||
assert "弹窗没有关闭" in str(exc.value)
|
||||
assert "请填写此字段" in str(exc.value)
|
||||
|
||||
|
||||
def test_upload_attachment_defaults_description(env: dict[str, Any]) -> None:
|
||||
"""附件说明是必填：info 里没给时要兜底成「发票：文件名」，否则提交会被挡。"""
|
||||
bot = _open(env, portal=True)
|
||||
target = env["case"] / "发票1.pdf"
|
||||
target.write_bytes(b"%PDF")
|
||||
bot.page._existing = {"#insertAcc", "#fjlx", "#file", "#submit2"}
|
||||
bot.upload_attachment("发票1.pdf", "invoice")
|
||||
fills = [c for c in bot.page.calls if c[0] == "fill" and c[1][0] == "#fpsmxx"]
|
||||
assert fills and fills[0][1][1] == "发票：发票1"
|
||||
assert bot.attachment_dialog_open() is False
|
||||
|
||||
|
||||
def test_upload_attachment_rejects_bad_paths(env: dict[str, Any]) -> None:
|
||||
bot = _open(env, portal=True)
|
||||
with pytest.raises(FileNotFoundError):
|
||||
bot.upload_attachment("不存在的.pdf")
|
||||
outside = Path(env["case"]).parent / "别处.pdf"
|
||||
outside.write_bytes(b"%PDF")
|
||||
with pytest.raises(RuntimeError, match="会话目录"):
|
||||
bot.upload_attachment(str(outside))
|
||||
|
||||
|
||||
def test_attachment_rows_finds_the_attachment_table(env: dict[str, Any]) -> None:
|
||||
"""附件清单不是 #table：早先按 #table 数，附件传成功了却报 0 行。"""
|
||||
bot = _open(env, portal=True)
|
||||
bot.page.tables = [
|
||||
FakeTable("序号 状态 报销说明", ["1 未提交 别的"]),
|
||||
FakeTable(
|
||||
"序号 类型 附件说明 附件备注 附件文件名",
|
||||
["1 1 发票：a 发票 /imgupload/a.pdf", "2 2 其他 b /imgupload/b.pdf"],
|
||||
),
|
||||
]
|
||||
assert bot.attachment_rows() == 2
|
||||
bot.page.tables = [FakeTable("序号 类型 附件说明 附件文件名")]
|
||||
assert bot.attachment_rows() == 0
|
||||
|
||||
Reference in New Issue
Block a user