"""测量同一 Android 设备冷/热启动打开 PDD 商品页的耗时。

本工具只启动或停止 PDD、打开商品链接并读取控件树；不点击规格、下单或付款。
输出只包含运行模式、稳定阶段名和毫秒值，不输出设备号、URL 或控件树。
"""

from __future__ import annotations

import argparse
import json
import math
import statistics
import subprocess
import time
import xml.etree.ElementTree as ET
from collections.abc import Callable
from typing import Any


PDD_PACKAGE_NAME = "com.xunmeng.pinduoduo"
READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
STOP_MARKERS = ("手机号登录", "请完成验证", "安全验证", "拖动滑块")


def elapsed_ms(started_at: float) -> int:
    return max(0, round((time.monotonic() - started_at) * 1000))


def measure_call(operation: str, call: Callable[[], Any]) -> tuple[Any, int]:
    started_at = time.monotonic()
    value = call()
    return value, elapsed_ms(started_at)


def labels(xml_data: str) -> list[str]:
    root = ET.fromstring(xml_data)
    values = []
    for node in root.iter("node"):
        value = str(node.get("text") or node.get("content-desc") or "").strip()
        if value:
            values.append(value)
    return values


def wait_goods_ready(device: Any, first_xml: str, timeout: float) -> None:
    deadline = time.monotonic() + timeout
    xml_data = first_xml
    while time.monotonic() < deadline:
        current_labels = labels(xml_data)
        combined = " ".join(current_labels)
        if any(marker in combined for marker in STOP_MARKERS):
            raise RuntimeError("PDD 出现登录或安全验证，测量已停止")
        loading = "加载中" in combined or "正在加载" in combined
        if not loading and any(marker in combined for marker in READY_MARKERS):
            return
        time.sleep(0.25)
        xml_data = str(device.dump_hierarchy())
    raise TimeoutError("等待 PDD 商品页就绪超时")


def measure_once(serial: str, goods_url: str, cold: bool) -> dict[str, int]:
    import uiautomator2 as u2

    if cold:
        subprocess.run(
            ["adb", "-s", serial, "shell", "am", "force-stop", PDD_PACKAGE_NAME],
            check=True,
            capture_output=True,
            text=True,
        )
        time.sleep(1)
    else:
        warm_device = u2.connect(serial)
        warm_device.app_start(PDD_PACKAGE_NAME)
        if not warm_device.app_wait(PDD_PACKAGE_NAME, timeout=10):
            raise RuntimeError("PDD 热启动准备失败")

    total_started_at = time.monotonic()
    result: dict[str, int] = {}
    _, result["adb_device_check"] = measure_call(
        "adb_device_check",
        lambda: subprocess.run(
            ["adb", "-s", serial, "get-state"],
            check=True,
            capture_output=True,
            text=True,
        ),
    )
    device, result["uiautomator2_connect"] = measure_call(
        "uiautomator2_connect", lambda: u2.connect(serial)
    )
    current, result["first_app_current"] = measure_call(
        "first_app_current", device.app_current
    )
    if current.get("package") != PDD_PACKAGE_NAME:
        started_at = time.monotonic()
        device.app_start(PDD_PACKAGE_NAME)
        if not device.app_wait(PDD_PACKAGE_NAME, timeout=10):
            raise RuntimeError("PDD 冷启动失败")
        result["pdd_start_or_wait"] = elapsed_ms(started_at)
    else:
        result["pdd_start_or_wait"] = 0

    _, result["open_url"] = measure_call(
        "open_url", lambda: device.open_url(goods_url)
    )
    first_xml, result["first_dump_hierarchy"] = measure_call(
        "first_dump_hierarchy", lambda: str(device.dump_hierarchy())
    )
    ready_started_at = time.monotonic()
    wait_goods_ready(device, first_xml, timeout=30)
    result["goods_page_ready"] = elapsed_ms(ready_started_at)
    result["end_to_end_total"] = elapsed_ms(total_started_at)
    return result


def percentile_95(values: list[int]) -> int:
    ordered = sorted(values)
    return ordered[max(0, math.ceil(len(ordered) * 0.95) - 1)]


def summarize(rows: list[dict[str, int]]) -> dict[str, dict[str, int]]:
    return {
        operation: {
            "median_ms": round(statistics.median(row[operation] for row in rows)),
            "p95_ms": percentile_95([row[operation] for row in rows]),
        }
        for operation in rows[0]
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("serial")
    parser.add_argument("goods_url")
    parser.add_argument("--runs", type=int, default=5)
    args = parser.parse_args()
    if args.runs < 1:
        parser.error("--runs 必须大于 0")

    output = {}
    for mode, cold in (("cold", True), ("hot", False)):
        rows = [
            measure_once(args.serial, args.goods_url, cold)
            for _ in range(args.runs)
        ]
        output[mode] = {"runs": args.runs, "stages": summarize(rows)}
    print(json.dumps(output, ensure_ascii=False, indent=2))
    return 0


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