autotrain/scripts/record_android_flow.py

#!/usr/bin/env python3
"""连续录制人工手机演示中的页面状态变化。

用户通过手机或 scrcpy 手动完成流程；本脚本轮询 Android XML，页面结构发生
变化时自动落盘截图、XML、可见文字和时间线。它不执行 tap / 输入等任何操作。
"""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import signal
import time
import xml.etree.ElementTree as ET
from datetime import datetime
from pathlib import Path

import device_runtime

PROJECT_ROOT = Path(__file__).resolve().parents[1]
REPLAY_ROOT = PROJECT_ROOT / "work" / "replay"
REMOTE_XML = "/sdcard/autotrain_manual_record.xml"
RUNNING = True


def _stop(_signum: int, _frame: object) -> None:
    global RUNNING
    RUNNING = False


def dump_xml() -> str:
    device_runtime.run_adb(["shell", "uiautomator", "dump", REMOTE_XML], check=False)
    result = device_runtime.run_adb_bytes(["exec-out", "cat", REMOTE_XML], check=False)
    if result.returncode != 0:
        raise RuntimeError(result.stderr.decode("utf-8", "ignore").strip() or "读取 UI XML 失败")
    return result.stdout.decode("utf-8", "ignore")


def visible_texts(xml_text: str) -> list[str]:
    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return []
    texts: 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 texts:
                texts.append(value)
    return texts


def ui_signature(xml_text: str) -> str:
    """忽略系统状态栏；保留文本、控件属性与 bounds，以识别弹窗和滚动。"""
    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return hashlib.sha256(xml_text.encode("utf-8", "ignore")).hexdigest()
    rows: list[str] = []
    keys = ("class", "resource-id", "text", "content-desc", "bounds", "clickable", "enabled", "checked")
    for node in root.iter("node"):
        attrs = node.attrib
        if attrs.get("package") == "com.android.systemui":
            continue
        rows.append("\x1f".join(attrs.get(key, "") for key in keys))
    return hashlib.sha256("\n".join(rows).encode("utf-8", "ignore")).hexdigest()


def page_hint(texts: list[str]) -> str:
    text = "\n".join(texts)
    if "证件号码后8位" in text or "请输入乘车人" 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"
    if "扫码" in text and "相册" in text:
        return "album_qr"
    return "other"


def slug(value: str) -> str:
    return re.sub(r"[^a-zA-Z0-9_-]+", "_", value)[:40] or "other"


def save_snapshot(session: Path, index: int, xml_text: str, reason: str, started_at: float) -> dict:
    texts = visible_texts(xml_text)
    hint = page_hint(texts)
    folder = session / f"{index:03d}_{slug(hint)}"
    folder.mkdir(parents=True, exist_ok=False)
    (folder / "ui.xml").write_text(xml_text, encoding="utf-8")
    (folder / "texts.txt").write_text("\n".join(texts), encoding="utf-8")
    screenshot = device_runtime.run_adb_bytes(["exec-out", "screencap", "-p"], check=False)
    if screenshot.returncode == 0:
        (folder / "screen.png").write_bytes(screenshot.stdout)
    metadata = {
        "index": index,
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "elapsed_seconds": round(time.monotonic() - started_at, 2),
        "page_hint": hint,
        "reason": reason,
        "text_count": len(texts),
        "ui_signature": ui_signature(xml_text),
    }
    (folder / "meta.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    return {"folder": folder.name, **metadata}


def record(output: Path, interval: float) -> None:
    output.mkdir(parents=True, exist_ok=False)
    started_at = time.monotonic()
    fingerprint = device_runtime.device_fingerprint()
    (output / "session.json").write_text(json.dumps({
        "started_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "device": fingerprint,
        "interval_seconds": interval,
        "mode": "manual_demonstration",
    }, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(f"session={output}", flush=True)
    timeline: list[dict] = []
    previous_signature = ""
    index = 0
    last_warning = ""
    last_warning_at = 0.0
    while RUNNING:
        try:
            xml_text = dump_xml()
            signature = ui_signature(xml_text)
            if signature != previous_signature:
                index += 1
                reason = "initial" if not previous_signature else "ui_changed"
                entry = save_snapshot(output, index, xml_text, reason, started_at)
                timeline.append(entry)
                previous_signature = signature
                print(f"snapshot={entry['folder']} page={entry['page_hint']} reason={reason}", flush=True)
        except Exception as exc:
            warning = str(exc)
            now = time.monotonic()
            if warning != last_warning or now - last_warning_at >= 10:
                print(f"record_warning={warning}", flush=True)
                last_warning = warning
                last_warning_at = now
            time.sleep(max(interval, 2.0))
            continue
        time.sleep(max(interval, 0.25))
    (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)


def main() -> int:
    parser = argparse.ArgumentParser(description="连续录制人工 Android 操作的页面变化")
    parser.add_argument("--output", default="", help="录制目录；默认 work/replay/<时间>_manual")
    parser.add_argument("--interval", type=float, default=0.75, help="轮询间隔秒数")
    args = parser.parse_args()
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
    output = Path(args.output) if args.output else REPLAY_ROOT / f"{stamp}_manual"
    signal.signal(signal.SIGINT, _stop)
    signal.signal(signal.SIGTERM, _stop)
    record(output, args.interval)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())