from __future__ import annotations

import base64
from contextlib import redirect_stderr
from io import BytesIO
import importlib.util
import io
import json
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch
from xml.etree import ElementTree

from PIL import Image

from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
from cmbuyer_client.pdd.sku_reveal_spike import (
    SkuRevealSpikeCapturer,
    SkuRevealSpikeError,
    _require_safe_reveal_path,
)
from cmbuyer_client.pdd.sku_selection import SkuSelectionError, _parse_nodes


_FIXTURES = Path(__file__).with_name("fixtures")
_PRODUCT = (_FIXTURES / "product_entry_8_17_0.xml").read_text(encoding="utf-8")
_EMPTY = (_FIXTURES / "sku_panel_empty_8_17_0.xml").read_text(encoding="utf-8")
_COLOR_ONLY = (_FIXTURES / "sku_panel_color_selected_size_hidden_8_17_0.xml").read_text(encoding="utf-8")
_M_SELECTED = (_FIXTURES / "sku_panel_size_m_restored_8_17_0.xml").read_text(encoding="utf-8")


def _candidate_unselected() -> str:
    root = ElementTree.fromstring(_M_SELECTED)
    parents = {child: parent for parent in root.iter() for child in parent}
    for node in root.iter("node"):
        if node.get("text") == "已选： 黑色 CHA (纯棉） M（建议100-115）":
            node.set("text", "请选择： 尺码")
        if node.get("text") in {"S（建议80-100）", "M（建议100-115）"}:
            node.set("selected", "false")
            wrapper = parents[node]
            for child in list(wrapper):
                if child.get("class") == "android.view.View":
                    wrapper.remove(child)
    action_root = next(
        node
        for node in root.iter("node")
        if node.get("class") == "android.view.ViewGroup"
        and node.get("clickable") == "true"
        and node.get("bounds") == "[0,366][1080,2328]"
    )
    button_frame = ElementTree.SubElement(
        action_root,
        "node",
        {
            "class": "android.widget.FrameLayout",
            "package": "com.xunmeng.pinduoduo",
            "text": "",
            "content-desc": "",
            "clickable": "true",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": "[0,2181][1080,2328]",
        },
    )
    button_content = ElementTree.SubElement(
        button_frame,
        "node",
        {
            "class": "android.widget.LinearLayout",
            "package": "com.xunmeng.pinduoduo",
            "text": "",
            "content-desc": "",
            "clickable": "false",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": "[273,2181][807,2328]",
        },
    )
    ElementTree.SubElement(
        button_content,
        "node",
        {
            "class": "android.widget.TextView",
            "package": "com.xunmeng.pinduoduo",
            "text": "选择尺码后，提交订单",
            "content-desc": "",
            "clickable": "false",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": "[285,2225][795,2284]",
        },
    )
    return ElementTree.tostring(root, encoding="unicode")


def _candidate_with_color_unselected() -> str:
    root = ElementTree.fromstring(_candidate_unselected())
    target = next(
        node
        for node in root.iter("node")
        if node.get("class") == "android.view.ViewGroup"
        and node.get("content-desc") == "黑色 CHA (纯棉）"
        and node.get("bounds") == "[372,1000][684,1024]"
    )
    for node in target.iter("node"):
        node.set("selected", "false")
    return ElementTree.tostring(root, encoding="unicode")


def _candidate_with_submit_risk(kind: str) -> str:
    root = ElementTree.fromstring(_candidate_unselected())
    parents = {child: parent for parent in root.iter() for child in parent}
    submit = next(
        node
        for node in root.iter("node")
        if "提交订单" in node.get("text", "")
        and node.get("class") == "android.widget.TextView"
    )
    parent = parents[submit]
    if kind == "missing":
        parent.remove(submit)
    elif kind == "duplicate":
        parent.append(ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode")))
    elif kind == "moved_up":
        submit.set("bounds", "[285,1900][795,1959]")
    else:
        raise AssertionError(kind)
    return ElementTree.tostring(root, encoding="unicode")


def _with_clickable_overlay(package: str, class_name: str, bounds: str) -> str:
    root = ElementTree.fromstring(_COLOR_ONLY)
    container = next(root.iter("node"))
    ElementTree.SubElement(
        container,
        "node",
        {
            "package": package,
            "class": class_name,
            "text": "",
            "content-desc": "",
            "clickable": "true",
            "enabled": "true",
            "visible-to-user": "true",
            "selected": "false",
            "scrollable": "false",
            "bounds": bounds,
        },
    )
    return ElementTree.tostring(root, encoding="unicode")


def _png() -> str:
    image = Image.new("RGB", (1080, 2376), "white")
    raw = BytesIO()
    image.save(raw, format="PNG")
    return base64.b64encode(raw.getvalue()).decode("ascii")


class _FakeDevice:
    def __init__(self) -> None:
        self.hierarchy = "<hierarchy />"
        self.after_hierarchy = _candidate_unselected()
        self.calls: list[tuple[object, ...]] = []
        self.swipe_error = False

    def app_info(self, package_name: str) -> dict[str, str]:
        self.calls.append(("app_info", package_name))
        return {"versionName": "8.17.0"}

    def app_current(self) -> dict[str, str]:
        self.calls.append(("app_current",))
        return {"package": "com.xunmeng.pinduoduo"}

    def window_size(self) -> tuple[int, int]:
        self.calls.append(("window_size",))
        return 1080, 2376

    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
        self.calls.append(("jsonrpc", method, params, timeout))
        if method == "dumpWindowHierarchy":
            return self.hierarchy
        if method == "takeScreenshot":
            return _png()
        if method == "click":
            if params == [865, 2218]:
                self.hierarchy = _EMPTY
                return ""
            if params == [528, 1387]:
                self.hierarchy = _COLOR_ONLY
                return ""
            raise AssertionError(params)
        if method == "swipe":
            self.hierarchy = self.after_hierarchy
            if self.swipe_error:
                raise TimeoutError("unknown outcome")
            return ""
        raise AssertionError(method)


class _FakeAdb:
    def __init__(self, device: _FakeDevice) -> None:
        self.device = device
        self.calls: list[tuple[object, ...]] = []

    def inspect(self, serial: str) -> DeviceInspection:
        self.calls.append(("inspect", serial))
        return DeviceInspection(AdbDevice(serial=serial, state="device"), "PKG110", "16")

    def start_pdd_view_intent(self, serial: str, goods_id: str) -> object:
        self.calls.append(("intent", serial, goods_id))
        self.device.hierarchy = _PRODUCT
        return object()


def _swipes(device: _FakeDevice) -> list[tuple[object, ...]]:
    return [call for call in device.calls if call[:2] == ("jsonrpc", "swipe")]


class SkuRevealSpikeTests(unittest.TestCase):
    def _capturer(self, device: _FakeDevice) -> SkuRevealSpikeCapturer:
        now = [0.0]
        return SkuRevealSpikeCapturer(
            _FakeAdb(device),
            lambda serial: device,
            30,
            monotonic_clock=lambda: now[0],
            sleep_function=lambda seconds: now.__setitem__(0, now[0] + seconds),
        )

    def _assert_after_rejected_once(self, after_hierarchy: str, *, ambiguous: bool = False) -> None:
        device = _FakeDevice()
        device.after_hierarchy = after_hierarchy
        device.swipe_error = ambiguous
        with TemporaryDirectory() as directory:
            target = Path(directory) / "evidence"
            with self.assertRaises(SkuRevealSpikeError):
                self._capturer(device).capture(
                    "192.168.0.173:5555",
                    "937122477375",
                    target,
                )
            self.assertEqual(len(_swipes(device)), 1)
            self.assertFalse(target.exists())
            self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

    def test_success_publishes_before_after_and_exactly_one_reveal(self) -> None:
        device = _FakeDevice()
        with TemporaryDirectory() as directory:
            target = Path(directory) / "evidence"
            result = self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)

            self.assertEqual(result.output_directory, target)
            self.assertEqual(len(_swipes(device)), 1)
            self.assertFalse(any(call[1] == "pressKey" for call in device.calls if call[0] == "jsonrpc"))
            for relative in (
                "before/screenshot.png",
                "before/hierarchy.xml",
                "after/screenshot.png",
                "after/hierarchy.xml",
                "manifest.json",
            ):
                self.assertTrue((target / relative).is_file(), relative)
            manifest_text = result.manifest_path.read_text(encoding="utf-8")
            manifest = json.loads(manifest_text)
            self.assertEqual(manifest["reveal_attempts"], 1)
            self.assertEqual(manifest["rpc_outcome"], "completed")
            self.assertNotIn("192.168.0.173:5555", manifest_text)
            self.assertNotIn("gesture", manifest)
            self.assertNotIn("coordinates", manifest)
            self.assertNotIn("start", manifest)
            self.assertNotIn("end", manifest)

    def test_ambiguous_rpc_is_read_only_reconciled_without_retry(self) -> None:
        device = _FakeDevice()
        device.swipe_error = True
        with TemporaryDirectory() as directory:
            target = Path(directory) / "evidence"
            result = self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
            self.assertEqual(len(_swipes(device)), 1)
            manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
            self.assertEqual(manifest["rpc_outcome"], "ambiguous_reconciled")

    def test_selected_m_or_precondition_drift_never_publishes_or_retries(self) -> None:
        for after in (
            _M_SELECTED,
            _candidate_unselected().replace("请选择： 尺码", "已选： 尺码"),
        ):
            with self.subTest():
                device = _FakeDevice()
                device.after_hierarchy = after
                with TemporaryDirectory() as directory:
                    target = Path(directory) / "evidence"
                    with self.assertRaises(SkuRevealSpikeError):
                        self._capturer(device).capture("192.168.0.173:5555", "937122477375", target)
                    self.assertEqual(len(_swipes(device)), 1)
                    self.assertFalse(target.exists())
                    self.assertEqual(list(Path(directory).glob(".*.staging-*")), [])

        device = _FakeDevice()
        device.hierarchy = "<hierarchy />"
        invalid_color = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码")
        original_call = device.jsonrpc_call

        def drift(method: str, params: object = None, timeout: float = 10) -> str:
            value = original_call(method, params, timeout)
            if method == "click" and params == [528, 1387]:
                device.hierarchy = invalid_color
            return value

        device.jsonrpc_call = drift  # type: ignore[method-assign]
        with TemporaryDirectory() as directory:
            with self.assertRaises((SkuSelectionError, SkuRevealSpikeError)):
                self._capturer(device).capture(
                    "192.168.0.173:5555",
                    "937122477375",
                    Path(directory) / "evidence",
                )
        self.assertEqual(_swipes(device), [])

    def test_hidden_sizes_after_swipe_are_not_published_or_retried(self) -> None:
        self._assert_after_rejected_once(_COLOR_ONLY)

    def test_post_reveal_color_selected_drift_is_not_published_or_retried(self) -> None:
        self._assert_after_rejected_once(_candidate_with_color_unselected())

    def test_submit_hard_reject_zone_risk_is_not_published_or_retried(self) -> None:
        for kind in ("missing", "duplicate", "moved_up"):
            with self.subTest(kind=kind):
                self._assert_after_rejected_once(_candidate_with_submit_risk(kind))

    def test_ambiguous_rpc_with_unchanged_page_is_not_published_or_retried(self) -> None:
        self._assert_after_rejected_once(_COLOR_ONLY, ambiguous=True)

    def test_invalid_goods_and_existing_target_are_zero_action(self) -> None:
        device = _FakeDevice()
        with TemporaryDirectory() as directory:
            target = Path(directory) / "existing"
            target.mkdir()
            with self.assertRaises(SkuRevealSpikeError):
                self._capturer(device).capture("wifi", "1", Path(directory) / "new")
            with self.assertRaises(SkuRevealSpikeError):
                self._capturer(device).capture("wifi", "937122477375", target)
        self.assertEqual(device.calls, [])

    def test_before_screenshot_drift_is_rechecked_before_zero_swipe(self) -> None:
        device = _FakeDevice()
        original_call = device.jsonrpc_call

        def drift(method: str, params: object = None, timeout: float = 10) -> str:
            value = original_call(method, params, timeout)
            if method == "takeScreenshot":
                device.hierarchy = _COLOR_ONLY.replace("请选择： 尺码", "请选择： 颜色分类 尺码")
            return value

        device.jsonrpc_call = drift  # type: ignore[method-assign]
        with TemporaryDirectory() as directory:
            target = Path(directory) / "evidence"
            with self.assertRaises(SkuSelectionError):
                self._capturer(device).capture(
                    "192.168.0.173:5555",
                    "937122477375",
                    target,
                )
            self.assertFalse(target.exists())
        self.assertEqual(_swipes(device), [])

    def test_complete_reveal_segment_rejects_narrow_impostor_and_invalid_bounds(self) -> None:
        for hierarchy in (
            _with_clickable_overlay(
                "com.xunmeng.pinduoduo",
                "android.view.ViewGroup",
                "[350,1450][370,1460]",
            ),
            _with_clickable_overlay(
                "com.android.systemui",
                "android.view.ViewGroup",
                "[0,366][1080,2328]",
            ),
            _with_clickable_overlay(
                "com.xunmeng.pinduoduo",
                "android.view.ViewGroup",
                "not-a-bound",
            ),
        ):
            with self.subTest(), self.assertRaises(SkuRevealSpikeError):
                _require_safe_reveal_path(_parse_nodes(hierarchy))


class SkuRevealSpikeCliTests(unittest.TestCase):
    def test_cli_has_no_gesture_or_task_specification_parameters(self) -> None:
        script = _load_reveal_script()
        arguments = script.parse_arguments(
            [
                "--serial", "device-1",
                "--goods-id", "937122477375",
                "--output-dir", "evidence",
            ]
        )
        self.assertEqual(
            set(vars(arguments)),
            {"serial", "goods_id", "output_dir", "timeout", "adb"},
        )
        script.validate_arguments(arguments)
        for field, value in (
            ("serial", ""),
            ("goods_id", "1"),
            ("timeout", 0),
            ("timeout", float("inf")),
        ):
            with self.subTest(field=field), self.assertRaises(ValueError):
                script.validate_arguments(
                    type("Arguments", (), vars(arguments) | {field: value})()
                )

    def test_cli_failure_is_redacted(self) -> None:
        script = _load_reveal_script()
        secret = "SERIAL=192.168.0.173:5555 <hierarchy>private</hierarchy>"

        class FailingCapturer:
            def __init__(self, *args: object, **kwargs: object) -> None:
                return None

            def capture(self, *args: object, **kwargs: object) -> object:
                raise SkuRevealSpikeError(secret)

        stderr = io.StringIO()
        with patch.object(script, "SkuRevealSpikeCapturer", FailingCapturer), redirect_stderr(stderr):
            status = script.main(
                [
                    "--serial", "192.168.0.173:5555",
                    "--goods-id", "937122477375",
                    "--output-dir", "evidence",
                ]
            )
        output = stderr.getvalue()
        self.assertEqual(status, 1)
        self.assertNotIn("Traceback", output)
        self.assertNotIn("192.168.0.173:5555", output)
        self.assertNotIn("private", output)


def _load_reveal_script() -> object:
    path = Path(__file__).resolve().parents[2] / "scripts" / "capture_sku_reveal_spike.py"
    specification = importlib.util.spec_from_file_location("capture_sku_reveal_spike_test", path)
    if specification is None or specification.loader is None:
        raise RuntimeError("无法加载 T-103 reveal 取证脚本。")
    module = importlib.util.module_from_spec(specification)
    specification.loader.exec_module(module)
    return module


if __name__ == "__main__":
    unittest.main()