"""H5 开票领域原子工具：每个工具只承担一个页面动作，可独立替换和验证。""" from __future__ import annotations import time from pathlib import Path import re from . import device, ledger, profiles, vision from .paths import OUTPUT from .tools import ToolError, observe_screen, observe_xml, tap_normalized, input_text, swipe_normalized, back def _image(context: dict) -> str: result = observe_screen({}, context) return str(result["screenshot"]) def _current(context: dict) -> dict: current = context.setdefault("invoice", {}).get("current") if not current: raise ToolError("没有当前乘客；必须先从列表选择") return current def _scroll_up(context: dict) -> dict: profile = profiles.load(str(context.get("device_serial") or "")) or {} gesture = profile.get("gestures", {}).get("scroll_up") if not isinstance(gesture, dict): raise ToolError("设备档案缺少 gestures.scroll_up；请重新采样或补充该设备配置") return swipe_normalized(gesture, context) def inspect_page(_: dict, context: dict) -> dict: image = _image(context) parsed = vision.detect_page(Path(image)) context.setdefault("invoice", {})["page"] = parsed.get("page", "other") return {"screenshot": image, **parsed} def open_next(_: dict, context: dict) -> dict: image = _image(context) rows = vision.list_rows(Path(image)) if not rows["page_ok"]: raise ToolError("当前页面不是可识别的扫码开票单列表") roster = ledger.read_roster(context.get("roster_path", "")) for row in rows["rows"]: if row.get("button_text") != "开具": continue matches = ledger.match(roster, str(row.get("masked_name", "")), str(row.get("id_first", "")), str(row.get("id_last", ""))) if len(matches) != 1: continue previous = ledger.progress_by_masked(context, str(row.get("masked_name", ""))) if previous and previous.get("status") in {"completed", "no_ticket"}: continue person = matches[0] context.setdefault("invoice", {})["current"] = {"masked_name": row["masked_name"], "real_name": person["姓名"], "id_last8": person["身份证后8位"]} ledger.record(context, person, row["masked_name"], "in_progress") return {"selected": row["masked_name"], **tap_normalized({"point": row["point"]}, context)} if rows.get("scroll_position") == "bottom": context.setdefault("invoice", {})["finished"] = True return {"finished": True, "reason": "没有待开具乘客"} return {"scrolled": True, **_scroll_up(context)} def _click_label(context: dict, text: str) -> dict: image = _image(context) target = vision.find_button(Path(image), text) if not target.get("found"): raise ToolError(f"视觉未找到「{text}」按钮") return tap_normalized({"point": target["point"]}, context) def verify(_: dict, context: dict) -> dict: current = _current(context) image = _image(context) input_target = vision.find_button(Path(image), "证件号码") if not input_target.get("found"): raise ToolError("视觉未找到证件号码输入框") tap_normalized({"point": input_target["point"]}, context) input_text({"text": current["id_last8"]}, context) return _click_label(context, "核验") def submit_info(_: dict, context: dict) -> dict: return _click_label(context, "提交") def confirm(_: dict, context: dict) -> dict: return _click_label(context, "确认") def _remote_downloads(context: dict) -> set[str]: command = "find /sdcard/Download -type f \\( -iname '*.pdf' -o -iname '*.ofd' -o -iname '*.xml' \\)" output = device.run(str(context["device_serial"]), ["shell", "sh", "-c", command], check=False).stdout return {line.strip() for line in str(output).splitlines() if line.strip().startswith("/")} def download_and_continue(_: dict, context: dict) -> dict: """在成功页下载到 NextGen 输出区，再记录完成并返回列表。""" current = _current(context) before = _remote_downloads(context) _click_label(context, "发票预览/下载") deadline = time.monotonic() + 30 downloaded: set[str] = set() while time.monotonic() < deadline: texts = observe_xml({}, context).get("texts", []) joined = "\n".join(texts) if "是否允许" in joined and "允许" in joined: _click_label(context, "允许") elif "下载发票（PDF）" in joined: _click_label(context, "下载发票（PDF）") elif "下载发票" in joined: _click_label(context, "下载发票") elif "下载进度" in joined and "100%" in joined and "确定" in joined: _click_label(context, "确定") downloaded = _remote_downloads(context) - before if downloaded: break time.sleep(1) if not downloaded: raise ToolError("未在手机 Download 目录发现新的发票文件") remote = sorted(downloaded)[-1] suffix = Path(remote).suffix.lower() or ".pdf" OUTPUT.mkdir(parents=True, exist_ok=True) local = OUTPUT / f"{context.get('id', 'invoice')}_{time.time_ns()}{suffix}" device.run(str(context["device_serial"]), ["pull", remote, str(local)]) ledger.record(context, {"姓名": current["real_name"], "身份证后8位": current["id_last8"]}, current["masked_name"], "completed", str(local)) _click_label(context, "继续开票") context.setdefault("invoice", {}).pop("current", None) return {"remote": remote, "local": str(local)} def send_email_and_continue(_: dict, context: dict) -> dict: """按旧流程的“发送至邮箱”结果路径处理；邮箱地址仍由 12306 已保存信息决定。""" current = _current(context) _click_label(context, "发送至邮箱") deadline = time.monotonic() + 20 while time.monotonic() < deadline: texts = observe_xml({}, context).get("texts", []) joined = "\n".join(texts) if "邮件发送完成" in joined: break if "邮箱地址确认" in joined: _click_label(context, "确认") time.sleep(1) else: raise ToolError("等待邮件发送完成超时") ledger.record(context, {"姓名": current["real_name"], "身份证后8位": current["id_last8"]}, current["masked_name"], "completed", "已发送至邮箱") _click_label(context, "继续开票") context.setdefault("invoice", {}).pop("current", None) return {"sent": True} def mark_no_ticket(_: dict, context: dict) -> dict: current = _current(context) ledger.record(context, {"姓名": current["real_name"], "身份证后8位": current["id_last8"]}, current["masked_name"], "no_ticket") context.setdefault("invoice", {}).pop("current", None) return back({}, context) def wait_loading(arguments: dict, context: dict) -> dict: time.sleep(max(0.2, min(float(arguments.get("seconds", 2)), 8))) return inspect_page({}, context)