110 lines
4.4 KiB
Python
110 lines
4.4 KiB
Python
"""按指定 serial 录制人工手机演示；不依赖旧录制器的“唯一设备”假设。"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import threading
|
|
import time
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from . import device
|
|
|
|
|
|
REMOTE_XML = "/sdcard/autotrain-next-record.xml"
|
|
|
|
|
|
def _xml(serial: str) -> str:
|
|
device.run(serial, ["shell", "uiautomator", "dump", REMOTE_XML])
|
|
return device.run(serial, ["exec-out", "cat", REMOTE_XML]).stdout
|
|
|
|
|
|
def _texts(xml: str) -> list[str]:
|
|
try:
|
|
root = ET.fromstring(xml)
|
|
except ET.ParseError:
|
|
return []
|
|
result: list[str] = []
|
|
for node in root.iter("node"):
|
|
if node.attrib.get("package") == "com.android.systemui":
|
|
continue
|
|
for key in ("text", "content-desc"):
|
|
value = (node.attrib.get(key) or "").strip()
|
|
if value and value not in result:
|
|
result.append(value)
|
|
return result
|
|
|
|
|
|
def _signature(xml: str) -> str:
|
|
try:
|
|
root = ET.fromstring(xml)
|
|
except ET.ParseError:
|
|
return hashlib.sha256(xml.encode("utf-8", "ignore")).hexdigest()
|
|
rows = []
|
|
for node in root.iter("node"):
|
|
if node.attrib.get("package") != "com.android.systemui":
|
|
rows.append("\x1f".join(node.attrib.get(key, "") for key in ("class", "resource-id", "text", "content-desc", "bounds", "clickable")))
|
|
return hashlib.sha256("\n".join(rows).encode()).hexdigest()
|
|
|
|
|
|
def _hint(texts: list[str]) -> str:
|
|
joined = "\n".join(texts)
|
|
if "证件号码后8位" in joined or "请输入乘车人" in joined:
|
|
return "verify"
|
|
if "发票信息确认" in joined:
|
|
return "invoice_confirm"
|
|
if "发票开具成功" in joined:
|
|
return "success"
|
|
if "发票信息" in joined and ("提交" in joined or "确认" in joined):
|
|
return "invoice_info"
|
|
if "没有可开具电子发票" in joined:
|
|
return "no_ticket"
|
|
if "开具" in joined and "*" in joined:
|
|
return "list"
|
|
if "扫码" in joined and "相册" in joined:
|
|
return "album_qr"
|
|
return "other"
|
|
|
|
|
|
def _save(output: Path, index: int, serial: str, xml: str, started: float) -> dict:
|
|
texts = _texts(xml)
|
|
hint = _hint(texts)
|
|
folder = output / f"{index:03d}_{re.sub(r'[^A-Za-z0-9_-]+', '_', hint)}"
|
|
folder.mkdir()
|
|
(folder / "ui.xml").write_text(xml, encoding="utf-8")
|
|
(folder / "texts.txt").write_text("\n".join(texts), encoding="utf-8")
|
|
screen = device.run(serial, ["exec-out", "screencap", "-p"], binary=True)
|
|
(folder / "screen.png").write_bytes(screen.stdout)
|
|
entry = {"folder": folder.name, "index": index, "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "elapsed_seconds": round(time.monotonic() - started, 2), "page_hint": hint, "text_count": len(texts), "ui_signature": _signature(xml)}
|
|
(folder / "meta.json").write_text(json.dumps(entry, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
return entry
|
|
|
|
|
|
def record(output: Path, serial: str, interval: float = 0.75, stop_event: threading.Event | None = None) -> None:
|
|
output.mkdir(parents=True, exist_ok=False)
|
|
started = time.monotonic()
|
|
(output / "session.json").write_text(json.dumps({"started_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "device": device.fingerprint(serial), "interval_seconds": interval, "mode": "manual_demonstration"}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(f"session={output}", flush=True)
|
|
previous = ""
|
|
timeline: list[dict] = []
|
|
try:
|
|
while not (stop_event and stop_event.is_set()):
|
|
try:
|
|
xml = _xml(serial)
|
|
signature = _signature(xml)
|
|
if signature != previous:
|
|
entry = _save(output, len(timeline) + 1, serial, xml, started)
|
|
timeline.append(entry)
|
|
previous = signature
|
|
print(f"snapshot={entry['folder']} page={entry['page_hint']}", flush=True)
|
|
except Exception as exc:
|
|
print(f"record_warning={exc}", flush=True)
|
|
time.sleep(max(0.25, interval))
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
(output / "timeline.json").write_text(json.dumps({"stopped_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "snapshots": timeline}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(f"finished snapshots={len(timeline)}", flush=True)
|