From 81997121b3ddeea6ae65690fa69e19fa76e80952 Mon Sep 17 00:00:00 2001 From: xinxin6623 Date: Thu, 11 Jun 2026 13:41:06 +0800 Subject: [PATCH] fix: accept invoice output args in vision run --- scripts/invoice_tool.py | 187 ++++++++++++++++++++++++++++++++++------ 1 file changed, 162 insertions(+), 25 deletions(-) diff --git a/scripts/invoice_tool.py b/scripts/invoice_tool.py index bbe69aa..8fd1bc6 100755 --- a/scripts/invoice_tool.py +++ b/scripts/invoice_tool.py @@ -865,6 +865,138 @@ def vision_detect_list_page() -> tuple[bool, str]: return parsed.get("page") == "list", parsed.get("summary", "") +def vision_detect_page() -> dict: + """截图 + 视觉判断当前页面类型,用于 XML 文本未稳定时兜底。""" + out = call_vision_subcommand(["detect-page"]) + content = out.get("content", "") + try: + parsed = json.loads(content) + except Exception: + match = re.search(r"\{.*\}", content, re.S) + if not match: + return {"page": "other", "summary": content[:120], "raw": content} + try: + parsed = json.loads(match.group(0)) + except Exception: + return {"page": "other", "summary": content[:120], "raw": content} + if not isinstance(parsed, dict): + return {"page": "other", "summary": str(parsed)[:120]} + return parsed + + +def xml_page_kind(nodes: list[dict[str, str]] | None = None) -> str: + """用 uiautomator 文本快速判断页面,比视觉便宜,作为主判据。""" + if nodes is None: + nodes = adb_dump_nodes() + page_text = "\n".join(adb_visible_texts(nodes)) + if ( + "扫码开票单" in page_text + and "开票流水号" in page_text + and "开具" in page_text + and "请输入乘车人" not in page_text + and "发票开具成功" not in page_text + ): + return "list" + if "请输入乘车人" in page_text and "证件号码后8位" in page_text: + return "verify" + if "发票信息确认" in page_text: + return "invoice_confirm" + if "发票信息" in page_text and ("提交" in page_text or "确认" in page_text): + return "invoice_info" + if "发票开具成功" in page_text: + return "success" + if "邮箱地址确认" in page_text: + return "email_confirm" + if "邮件发送完成，请注意查收。" in page_text: + return "mail_sent" + if "您没有可开具" in page_text or ("发票管理" in page_text and "客票" in page_text): + return "no_ticket" + if "发票开具中" in page_text or "发票正在开具中" in page_text: + return "loading" + if "业务须知" in page_text and "国家税务总局" in page_text: + return "instructions" + if "查询车票" in page_text or ("火车票" in page_text and "飞机票" in page_text): + return "home" + return "other" + + +def wait_for_page_transition( + args: argparse.Namespace, + expected_pages: set[str], + *, + label: str, + network_sensitive: bool = False, +) -> str: + """等待页面切到目标状态;先高频 XML 轮询,必要时截图视觉兜底。 + + network_sensitive=True 的步骤通常触发 12306 后端请求,先给短等待,若仍是 loading/other + 就自动延长,避免固定把每步都 sleep 很久。 + """ + quick_timeout = float(getattr(args, "smart_wait_quick", 2.0)) + network_timeout = float(getattr(args, "smart_wait_network", 12.0)) if network_sensitive else 0.0 + vision_interval = float(getattr(args, "smart_wait_vision_interval", 1.2)) + deadline = time.perf_counter() + max(quick_timeout, 0.1) + network_deadline = deadline + max(network_timeout, 0.0) + last_vision_at = 0.0 + last_kind = "other" + phase = "quick" + + while True: + nodes = adb_dump_nodes() + kind = xml_page_kind(nodes) + last_kind = kind + if kind in expected_pages: + print(f"wait_page_hit label={label} via=xml page={kind} phase={phase}") + return kind + + now = time.perf_counter() + if getattr(args, "screenshot_wait", False) and now - last_vision_at >= vision_interval: + last_vision_at = now + try: + detected = vision_detect_page() + v_page = str(detected.get("page") or "other") + summary = str(detected.get("summary") or "") + if v_page == "success" and ("开具中" in summary or "正在开具" in summary): + v_page = "loading" + print(f"wait_page_probe label={label} via=vision page={v_page} summary={summary}") + if v_page in expected_pages: + return v_page + if v_page not in {"other", "launcher"}: + last_kind = v_page + except Exception as exc: + print(f"wait_page_vision_error label={label}: {exc}") + + if now >= deadline: + if network_sensitive and now < network_deadline: + if phase != "network": + print(f"wait_page_more label={label} reason=network_or_loading last={last_kind} extra={network_timeout}s") + phase = "network" + time.sleep(0.5) + continue + print(f"wait_page_timeout label={label} expected={sorted(expected_pages)} last={last_kind}") + return last_kind + time.sleep(0.25) + + +def recover_to_h5_list(args: argparse.Namespace, row_label: str, *, reason: str) -> bool: + """子流程断档时回列表;back 不够就扫码重进,让下次循环靠进度表续接。""" + print(f"recover_start row={row_label} reason={reason}") + if is_on_list_page(): + print(f"recover_ok row={row_label} already_list") + return True + recovered = safe_back_until_list(max_back=getattr(args, "recover_max_back", 4)) + if recovered: + print(f"recover_ok row={row_label} via=back") + return True + if is_on_instructions_page() or getattr(args, "recover_auto_scan", True): + print(f"recover_try_auto_scan row={row_label}") + if auto_scan_qr_to_list(): + print(f"recover_ok row={row_label} via=auto_scan") + return True + print(f"recover_failed row={row_label}") + return False + + def vision_list_rows() -> dict: """截屏 + 视觉识别列表页所有可见乘客行,返回 parsed dict(已反归一到绝对像素)。 @@ -1330,13 +1462,7 @@ def drive_invoice_subflow(args: argparse.Namespace, row_label: str) -> str: mark_clicked_as_runtime_error() consecutive_errors += 1 if consecutive_errors >= 2: - recovered = safe_back_until_list(max_back=4) - if not recovered: - # 可能在法条说明页,尝试自动扫码回到列表 - if is_on_instructions_page(): - print(f"sub_recover via auto_scan row={row_label}") - if auto_scan_qr_to_list(): - recovered = True + recovered = recover_to_h5_list(args, row_label, reason="sub_step_errors") print(f"sub_abort row={row_label} consecutive_errors=2 recovered_to_list={recovered}") return "error" time.sleep(args.between_steps) @@ -1692,7 +1818,12 @@ def android_h5_page_run(args: argparse.Namespace) -> None: print(f"retry_failed_attempt {masked}") update_progress(masked, real_name, id8, "in_progress") run_adb(["shell", "input", "tap", str(tap_point[0]), str(tap_point[1])]) - wait_short(args, ["请输入乘车人", "证件号码后8位", "发票管理", "您没有可开具"]) + wait_for_page_transition( + args, + {"verify", "no_ticket", "invoice_info", "invoice_confirm", "success"}, + label=f"after_list_open:{masked}", + network_sensitive=True, + ) result = drive_invoice_subflow(args, masked) total_tapped += 1 @@ -1706,13 +1837,7 @@ def android_h5_page_run(args: argparse.Namespace) -> None: elif p_row.get("masked_name") == masked and p_row.get("status") == "no_ticket": total_no_ticket += 1 else: - recovered = safe_back_until_list(max_back=4) - if not recovered: - # 可能在法条说明页,尝试自动扫码 - if is_on_instructions_page(): - print(f"recover via auto_scan row={masked}") - if auto_scan_qr_to_list(): - recovered = True + recovered = recover_to_h5_list(args, masked, reason=result) if not recovered: print(f"STOP row={masked} sub_result={result} 无法回到列表页") print(f"进度已保存,手机恢复后从 UI 重新启动即可续接") @@ -1722,6 +1847,9 @@ def android_h5_page_run(args: argparse.Namespace) -> None: # back 后 scroll 重置到顶部,跳出 actions 循环重新截屏 if idx < len(actions) - 1: print(f" (back 重置 scroll,重新截屏找下一个)") + if args.max_actions and total_tapped >= args.max_actions: + print(f"stop reason=max_actions total_tapped={total_tapped}") + return break # dry-run 模式下当前屏全部看完后自动滚动 @@ -1925,18 +2053,18 @@ def android_invoice_step(args: argparse.Namespace) -> None: if "邮箱地址确认" in page_text: if not _xml_or_vision_click(nodes, "确认", "email_confirm", clickable=True): raise RuntimeError("邮箱确认页未找到确认按钮") - wait_short(args, ["邮件发送完成", "发票开具成功"]) + wait_for_page_transition(args, {"mail_sent", "success"}, label="after_email_confirm", network_sensitive=True) return if "发票信息确认" in page_text: if not _xml_or_vision_click(nodes, "确认", "invoice_info_confirm", clickable=True): raise RuntimeError("发票信息确认页未找到确认按钮") - wait_long(args, ["发票开具成功"]) + wait_for_page_transition(args, {"success", "no_ticket"}, label="after_invoice_confirm", network_sensitive=True) return if "发票开具中" in page_text or "发票正在开具中" in page_text: print("invoice_opening_wait") - wait_and_probe(1, getattr(args, "wait_secondary", 10), ["发票开具成功", "邮件发送完成", "发票管理"]) + wait_for_page_transition(args, {"success", "mail_sent", "no_ticket"}, label="invoice_opening", network_sensitive=True) return if "发票开具成功" in page_text: @@ -1953,7 +2081,7 @@ def android_invoice_step(args: argparse.Namespace) -> None: row = current_row update_progress(row.get("masked_name", ""), row.get("real_name", ""), row.get("id_last8", ""), "completed") if _xml_or_vision_click(nodes, "发送至邮箱", "send_email"): - wait_short(args, ["邮箱地址确认", "邮件发送完成"]) + wait_for_page_transition(args, {"email_confirm", "mail_sent"}, label="after_send_email", network_sensitive=True) return if success_action == "download" and in_progress: local_path = click_invoice_download(args, nodes, row_label) @@ -1973,6 +2101,7 @@ def android_invoice_step(args: argparse.Namespace) -> None: update_progress(row.get("masked_name", ""), row.get("real_name", ""), row.get("id_last8", ""), "completed") if not _xml_or_vision_click(nodes, "继续开票", "continue_invoice"): raise RuntimeError("成功页未找到继续开票按钮(XML+视觉均失败)") + wait_for_page_transition(args, {"list", "instructions", "no_ticket"}, label="after_continue_invoice", network_sensitive=False) return if "请输入乘车人" in page_text and "证件号码后8位" in page_text: @@ -2001,7 +2130,7 @@ def android_invoice_step(args: argparse.Namespace) -> None: update_progress(masked_name, real_name, id8, "in_progress") if not _xml_or_vision_click(adb_dump_nodes(), "核验", "verify"): raise RuntimeError("未找到核验按钮(XML+视觉均失败)") - wait_short(args, ["发票管理", "待开票", "发票信息"]) + wait_for_page_transition(args, {"no_ticket", "invoice_info", "invoice_confirm", "success"}, label="after_verify", network_sensitive=True) return if "身份核验" in page_text and "证件号码的后8位" in page_text: @@ -2018,7 +2147,7 @@ def android_invoice_step(args: argparse.Namespace) -> None: adb_input_text(matches[0]["身份证后8位"]) if not _xml_or_vision_click(adb_dump_nodes(), "开始核验", "agent_verify"): raise RuntimeError("代办人身份核验页未找到开始核验按钮") - wait_short(args, ["扫码开票单", "开票流水号"]) + wait_for_page_transition(args, {"list"}, label="after_agent_verify", network_sensitive=True) return if "发票信息" in page_text and "提交" in page_text: @@ -2028,7 +2157,7 @@ def android_invoice_step(args: argparse.Namespace) -> None: raise RuntimeError("发票信息页未检测到默认抬头/税号，暂停，需手工确认") if not _xml_or_vision_click(nodes, "提交", "submit_invoice_info"): raise RuntimeError("未找到提交按钮(XML+视觉均失败)") - wait_short(args, ["发票信息确认"]) + wait_for_page_transition(args, {"invoice_confirm"}, label="after_submit_invoice_info", network_sensitive=True) return if "发票管理" in page_text and "您没有可开具电子发票的客票" in page_text: @@ -2052,13 +2181,13 @@ def android_invoice_step(args: argparse.Namespace) -> None: pt = _vision_find_button("开具") if pt: run_adb(["shell", "input", "tap", str(pt[0]), str(pt[1])]) - wait_short(args, ["发票信息"]) + wait_for_page_transition(args, {"invoice_info", "invoice_confirm"}, label="after_ticket_open_vision", network_sensitive=True) return raise RuntimeError("发票管理页未找到单张开具按钮(XML+视觉均失败)") # Avoid bottom batch button; choose first visible normal button. node = sorted(open_nodes, key=node_center_y)[0] click_node(node, "ticket_open") - wait_short(args, ["发票信息"]) + wait_for_page_transition(args, {"invoice_info", "invoice_confirm"}, label="after_ticket_open", network_sensitive=True) return if "扫码开票单" in page_text: @@ -2068,7 +2197,7 @@ def android_invoice_step(args: argparse.Namespace) -> None: masked_name, row, button = passenger print(f"passenger={masked_name} matched={row.get('姓名')} id_last8={row.get('身份证后8位')}") click_node(button, f"list_open:{masked_name}") - wait_short(args, ["请输入乘车人", "证件号码后8位"]) + wait_for_page_transition(args, {"verify", "no_ticket", "invoice_info"}, label=f"after_xml_list_open:{masked_name}", network_sensitive=True) return if "com.bbk.launcher2" in page_text or "launcher" in page_text: @@ -2510,6 +2639,8 @@ def build_parser() -> argparse.ArgumentParser: p_vision_run = subparsers.add_parser("android-vision-page-run", help="视觉驱动批量开票:截图识别 -> anchor 去重 -> 按页批次 tap -> 子流程 -> 滑动") p_vision_run.add_argument("--roster", default="work/passenger_roster.csv") p_vision_run.add_argument("--send-email", action="store_true") + p_vision_run.add_argument("--invoice-success-action", choices=["download", "email", "continue"], default="", help="成功页动作: download=下载发票,email=发送邮箱,continue=仅继续") + p_vision_run.add_argument("--invoice-output-dir", default="", help="PDF/OFD/XML 保存到电脑本地目录") p_vision_run.add_argument("--fast", action="store_true") p_vision_run.add_argument("--dry-run", action="store_true", help="只识别+匹配,不真 tap;用于验证视觉精度") p_vision_run.add_argument("--no-precheck", dest="precheck", action="store_false", help="跳过开始前的视觉自检") @@ -2543,6 +2674,12 @@ def build_parser() -> argparse.ArgumentParser: p_h5_run.add_argument("--wait-secondary", type=int, default=10, help="扩展等待秒数") p_h5_run.add_argument("--short-wait", type=int, default=1, help="短等待秒数(页面切换)") p_h5_run.add_argument("--between-steps", type=float, default=0.3) + p_h5_run.add_argument("--screenshot-wait", action="store_true", help="等待页面跳转时用截图视觉判页兜底") + p_h5_run.add_argument("--smart-wait-quick", type=float, default=1.2, help="非网络跳转的快速等待秒数") + p_h5_run.add_argument("--smart-wait-network", type=float, default=12.0, help="疑似走网络时额外等待秒数") + p_h5_run.add_argument("--smart-wait-vision-interval", type=float, default=1.5, help="截图判页最小间隔秒数") + p_h5_run.add_argument("--recover-max-back", type=int, default=4, help="子流程断档时最多 back 次数") + p_h5_run.add_argument("--no-recover-auto-scan", dest="recover_auto_scan", action="store_false", help="回列表失败时不自动扫码重进") p_h5_run.add_argument("--retry-failed-once", dest="retry_failed_once", type=int, default=1, help="当前运行内对可见 failed/error 人员最多重试次数") p_h5_run.add_argument("--no-resume-after-last-progress", dest="resume_after_last_progress", action="store_false", help="不按进度表最后处理人定位续跑") # H5 滚动参数(页面内滚动,不是翻页)