- web_ui.py: major refactor with improved runner, mode selection, progress tracking - scripts: add device_runtime, device_calibrate, calibration_trace, record_android_flow - docs: add phone-control-manual, update device-adaptation and windows-migration - invoice_tool: enhance H5 automation flow and visual fallback - calibrate.py: expand calibration capabilities - add skills/ and tests/ directories - remove obsolete .opencode/skills/invoice-wrapup Co-Authored-By: Claude <noreply@anthropic.com>
70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
||
"""自动校准期间的轻量回放证据记录。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import device_runtime
|
||
|
||
|
||
def session_dir() -> Path | None:
|
||
value = os.environ.get("AUTOTRAIN_CALIBRATION_DIR", "").strip()
|
||
return Path(value) if value else None
|
||
|
||
|
||
def _page_hint(text: str) -> str:
|
||
if "请输入乘车人" in text or "证件号码后8位" in text:
|
||
return "verify"
|
||
if "发票信息确认" in text:
|
||
return "invoice_confirm"
|
||
if "发票开具成功" in text:
|
||
return "success"
|
||
if "发票信息" in text and ("提交" in text or "确认" in text):
|
||
return "invoice_info"
|
||
if "没有可开具电子发票" in text:
|
||
return "no_ticket"
|
||
if "开具" in text and "*" in text:
|
||
return "list"
|
||
return "other"
|
||
|
||
|
||
def capture(label: str, command: list[str] | None = None) -> None:
|
||
"""保存当前截图、XML、可见文字及动作元数据;失败不影响真实流程。"""
|
||
root = session_dir()
|
||
if root is None:
|
||
return
|
||
try:
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
index = len(list(root.glob("*_*.png"))) + 1
|
||
safe = re.sub(r"[^a-zA-Z0-9_-]+", "_", label)[:48] or "state"
|
||
stem = f"{index:02d}_{safe}"
|
||
shot = device_runtime.run_adb_bytes(["exec-out", "screencap", "-p"], check=False)
|
||
(root / f"{stem}.png").write_bytes(shot.stdout)
|
||
device_runtime.run_adb(["shell", "uiautomator", "dump", "/sdcard/autotrain_calibration.xml"], check=False)
|
||
xml = device_runtime.run_adb_bytes(["exec-out", "cat", "/sdcard/autotrain_calibration.xml"], check=False)
|
||
xml_text = xml.stdout.decode("utf-8", "ignore")
|
||
(root / f"{stem}.xml").write_text(xml_text, encoding="utf-8")
|
||
texts = re.findall(r'(?:text|content-desc)="([^"]+)"', xml_text)
|
||
(root / f"{stem}.txt").write_text("\n".join(texts), encoding="utf-8")
|
||
meta = {
|
||
"label": label,
|
||
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"command": command or [],
|
||
"page_hint": _page_hint("\n".join(texts)),
|
||
}
|
||
(root / f"{stem}.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
except Exception as exc: # 证据记录不能让开票流程中断
|
||
try:
|
||
(root / "trace_errors.log").open("a", encoding="utf-8").write(f"{label}: {exc}\n")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def is_enabled() -> bool:
|
||
return session_dir() is not None
|