"""原子工具注册表。编排器与 LLM 都只能从这里选择动作。""" from __future__ import annotations import xml.etree.ElementTree as ET import re import time from dataclasses import dataclass from pathlib import Path from typing import Callable from . import device from .paths import WORK, ensure_workdirs from .profiles import normalized_point class ToolError(RuntimeError): pass @dataclass(frozen=True) class ToolSpec: name: str handler: Callable[[dict, dict], dict] requires_submit_permission: bool = False def _serial(context: dict) -> str: serial = str(context.get("device_serial") or "") return device.selected_device(serial).serial def device_status(_: dict, context: dict) -> dict: serial = _serial(context) return {"device": device.fingerprint(serial)} def observe_screen(_: dict, context: dict) -> dict: serial = _serial(context) ensure_workdirs() task_id = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(context.get("id") or serial)) target = WORK / "screens" / task_id / f"{time.time_ns()}.png" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(device.run(serial, ["exec-out", "screencap", "-p"], binary=True).stdout) return {"screenshot": str(target), "screen": device.screen_size(serial)} def observe_xml(_: dict, context: dict) -> dict: serial = _serial(context) remote = "/sdcard/autotrain-next-ui.xml" device.run(serial, ["shell", "uiautomator", "dump", remote]) xml = device.run(serial, ["exec-out", "cat", remote]).stdout if isinstance(xml, bytes): xml = xml.decode("utf-8", "ignore") try: root = ET.fromstring(xml) except ET.ParseError: return {"xml_available": False, "texts": []} texts: list[str] = [] for node in root.iter("node"): for key in ("text", "content-desc"): value = (node.attrib.get(key) or "").strip() if value and value not in texts: texts.append(value) return {"xml_available": True, "texts": texts[:80]} def tap_normalized(arguments: dict, context: dict) -> dict: point = arguments.get("point") if not isinstance(point, list) or len(point) != 2: raise ToolError("ui.tap_normalized 需要 point: [x, y]") serial = _serial(context) x, y = normalized_point((float(point[0]), float(point[1])), device.screen_size(serial)) device.run(serial, ["shell", "input", "tap", str(x), str(y)]) return {"tap": {"normalized": point, "pixel": [x, y]}} def swipe_normalized(arguments: dict, context: dict) -> dict: start, end = arguments.get("from"), arguments.get("to") if not all(isinstance(value, list) and len(value) == 2 for value in (start, end)): raise ToolError("ui.swipe_normalized 需要 from 和 to") serial = _serial(context) screen = device.screen_size(serial) x1, y1 = normalized_point((float(start[0]), float(start[1])), screen) x2, y2 = normalized_point((float(end[0]), float(end[1])), screen) duration = max(50, min(int(arguments.get("duration_ms", 350)), 3000)) device.run(serial, ["shell", "input", "swipe", str(x1), str(y1), str(x2), str(y2), str(duration)]) return {"swipe": {"from": [x1, y1], "to": [x2, y2], "duration_ms": duration}} def input_text(arguments: dict, context: dict) -> dict: value = str(arguments.get("text") or "").upper() if not re.fullmatch(r"[0-9]{7}[0-9X]", value): raise ToolError("ui.input_id_suffix 只接受身份证后八位的数字或 X，避免 ADB shell 转义风险") device.run(_serial(context), ["shell", "input", "text", value]) return {"length": len(value)} def back(_: dict, context: dict) -> dict: device.run(_serial(context), ["shell", "input", "keyevent", "4"]) return {"keyevent": "BACK"} def invoice_inspect(arguments: dict, context: dict) -> dict: from .invoice_tools import inspect_page return inspect_page(arguments, context) def invoice_open_next(arguments: dict, context: dict) -> dict: from .invoice_tools import open_next return open_next(arguments, context) def invoice_verify(arguments: dict, context: dict) -> dict: from .invoice_tools import verify return verify(arguments, context) def invoice_submit_info(arguments: dict, context: dict) -> dict: from .invoice_tools import submit_info return submit_info(arguments, context) def invoice_confirm(arguments: dict, context: dict) -> dict: from .invoice_tools import confirm return confirm(arguments, context) def invoice_download(arguments: dict, context: dict) -> dict: from .invoice_tools import download_and_continue return download_and_continue(arguments, context) def invoice_email(arguments: dict, context: dict) -> dict: from .invoice_tools import send_email_and_continue return send_email_and_continue(arguments, context) def invoice_no_ticket(arguments: dict, context: dict) -> dict: from .invoice_tools import mark_no_ticket return mark_no_ticket(arguments, context) def invoice_wait(arguments: dict, context: dict) -> dict: from .invoice_tools import wait_loading return wait_loading(arguments, context) def nav_push_qr(arguments: dict, context: dict) -> dict: from .navigation import push_qr return push_qr(arguments, context) def nav_open_scan(arguments: dict, context: dict) -> dict: from .navigation import open_scan return open_scan(arguments, context) def nav_open_orders(arguments: dict, context: dict) -> dict: from .navigation import open_orders return open_orders(arguments, context) def nav_open_album(arguments: dict, context: dict) -> dict: from .navigation import open_album return open_album(arguments, context) def nav_select_qr(arguments: dict, context: dict) -> dict: from .navigation import select_qr return select_qr(arguments, context) def nav_wait_list(arguments: dict, context: dict) -> dict: from .navigation import wait_list return wait_list(arguments, context) REGISTRY = { spec.name: spec for spec in ( ToolSpec("device.status", device_status), ToolSpec("observe.screen", observe_screen), ToolSpec("observe.xml", observe_xml), ToolSpec("ui.tap_normalized", tap_normalized), ToolSpec("ui.swipe_normalized", swipe_normalized), ToolSpec("ui.input_digits", input_text), ToolSpec("ui.back", back), ToolSpec("invoice.inspect", invoice_inspect), ToolSpec("invoice.open_next", invoice_open_next, requires_submit_permission=True), ToolSpec("invoice.verify", invoice_verify, requires_submit_permission=True), ToolSpec("invoice.submit_info", invoice_submit_info, requires_submit_permission=True), ToolSpec("invoice.confirm", invoice_confirm, requires_submit_permission=True), ToolSpec("invoice.download", invoice_download, requires_submit_permission=True), ToolSpec("invoice.email", invoice_email, requires_submit_permission=True), ToolSpec("invoice.no_ticket", invoice_no_ticket, requires_submit_permission=True), ToolSpec("invoice.wait", invoice_wait), ToolSpec("nav.push_qr", nav_push_qr), ToolSpec("nav.open_scan", nav_open_scan, requires_submit_permission=True), ToolSpec("nav.open_orders", nav_open_orders, requires_submit_permission=True), ToolSpec("nav.open_album", nav_open_album, requires_submit_permission=True), ToolSpec("nav.select_qr", nav_select_qr, requires_submit_permission=True), ToolSpec("nav.wait_list", nav_wait_list), ) } def execute(name: str, arguments: dict, context: dict) -> dict: spec = REGISTRY.get(name) if not spec: raise ToolError(f"不允许的工具：{name}") if spec.requires_submit_permission and not context.get("allow_submit"): raise ToolError(f"{name} 可能触发开票，任务未启用 allow_submit") return spec.handler(arguments, context)