"""Base ADB client with shared connection management and commands.

Subclasses add device-specific polling (media sessions, wake state, etc.).
"""

import contextlib
import logging
import re

from .media_session import MediaSessionInfo, parse_all_media_sessions, parse_media_session

_LOGGER = logging.getLogger(__name__)

# Standard Android media key codes
MEDIA_KEYCODES = {
    "play": 126,
    "pause": 127,
    "play-pause": 85,
    "stop": 86,
    "next": 87,
    "previous": 88,
}

DEVICE_KEYCODES = {
    "power": 26,
    "sleep": 223,
    "wakeup": 224,
    "volume_up": 24,
    "volume_down": 25,
    "mute": 164,
}

ALL_KEYCODES = {**MEDIA_KEYCODES, **DEVICE_KEYCODES}

# Recognized non-awake mWakefulness= values (see Android PowerManagerInternal).
# Anything else is "unknown" — never silently reported as awake or asleep.
_WAKEFULNESS_ASLEEP_VALUES = {"Asleep", "Dozing", "Dozing-suspend"}


def parse_wakefulness(text: str | None) -> bool | None:
    """Parse `dumpsys power` output for the exact ``mWakefulness=`` value (R5).

    Exact-value parse, no substring matching: only the ``mWakefulness=``
    assignment counts — ``mWakefulnessChanging=false`` and
    ``mWakefulnessSetting=...`` are ignored. Unknown values (e.g. a future
    firmware's ``mWakefulness=Dozing-ext``) return None so callers treat it
    as "unknown", never as awake or asleep.
    """
    if not text:
        return None
    for line in text.splitlines():
        match = re.search(r"(?<![A-Za-z])mWakefulness=([^\s,}]+)", line)
        if match:
            value = match.group(1)
            if value == "Awake":
                return True
            if value in _WAKEFULNESS_ASLEEP_VALUES:
                return False
            return None
    return None


class AdbClientBase:
    """Base ADB client with connection management and common commands."""

    def __init__(self, host: str, port: int, key_path: str):
        self._host = host
        self._port = port
        self._key_path = key_path
        self._device = None
        self._connected = False

        # Media session state
        self._last_info: MediaSessionInfo | None = None
        self._last_dump: str | None = None
        self._all_sessions: list[MediaSessionInfo] = []


    @property
    def connected(self) -> bool:
        return self._connected

    @property
    def last_info(self) -> MediaSessionInfo | None:
        return self._last_info

    @property
    def all_sessions(self) -> list[MediaSessionInfo]:
        return self._all_sessions

    def clear_info(self) -> None:
        """Clear the cached media session info."""
        self._last_info = None
        self._last_dump = None
        self._all_sessions = []

    async def connect(self) -> bool:
        """Connect to the device via ADB."""
        try:
            import asyncio

            from adb_shell.adb_device_async import AdbDeviceTcpAsync
            from adb_shell.auth.sign_pythonrsa import PythonRSASigner

            # FromRSAKeyPath does blocking file I/O — run it in the default
            # executor so we don't stall Home Assistant's event loop.
            loop = asyncio.get_running_loop()
            signer = await loop.run_in_executor(
                None, PythonRSASigner.FromRSAKeyPath, self._key_path
            )
            self._device = AdbDeviceTcpAsync(self._host, self._port)
            await self._device.connect(rsa_keys=[signer], auth_timeout_s=10)
            self._connected = True
            _LOGGER.info("ADB connected to %s:%d", self._host, self._port)
            return True
        except Exception as err:
            _LOGGER.debug("ADB connection failed: %s", err)
            self._connected = False
            self._device = None
            return False

    async def disconnect(self) -> None:
        """Disconnect from the device."""
        if self._device:
            with contextlib.suppress(Exception):
                await self._device.close()
        self._device = None
        self._connected = False

    async def shell(self, cmd: str) -> str | None:
        """Run a shell command, reconnecting if needed."""
        if (not self._connected or not self._device) and not await self.connect():
            return None
        try:
            return await self._device.shell(cmd)
        except Exception as err:
            _LOGGER.debug("ADB shell failed: %s", err)
            self._connected = False
            self._device = None
            return None

    # -- Common commands --

    async def send_keyevent(self, command: str) -> bool:
        """Send a key event via ADB input keyevent."""
        keycode = ALL_KEYCODES.get(command)
        if keycode is None:
            _LOGGER.warning("Unknown keyevent command: %s", command)
            return False
        result = await self.shell(f"input keyevent {keycode}")
        if result is not None:
            _LOGGER.debug("ADB keyevent %d (%s)", keycode, command)
        return result is not None

    async def send_media_command(self, command: str) -> bool:
        """Send a media command via ADB input keyevent."""
        keycode = MEDIA_KEYCODES.get(command)
        if keycode is None:
            _LOGGER.warning("Unknown media command: %s", command)
            return False
        result = await self.shell(f"input keyevent {keycode}")
        if result is not None:
            _LOGGER.debug("ADB keyevent %d (%s)", keycode, command)
            return True
        return False

    async def send_digit_keyevent(self, digit: str) -> bool:
        """Send a channel digit via `input keyevent` (KEYCODE_0..9 = 7..16)."""
        if not digit.isdigit() or len(digit) != 1:
            _LOGGER.warning("Not a single digit: %s", digit)
            return False
        result = await self.shell(f"input keyevent {7 + int(digit)}")
        if result is not None:
            _LOGGER.debug("ADB digit keyevent %d (%s)", 7 + int(digit), digit)
        return result is not None

    async def send_enter_keyevent(self) -> bool:
        """Send ENTER via `input keyevent` (KEYCODE_ENTER = 66) — the ENTER
        fallback for the §7 channel sequence."""
        result = await self.shell("input keyevent 66")
        if result is not None:
            _LOGGER.debug("ADB ENTER keyevent (66)")
        return result is not None

    async def bring_to_foreground(self, package: str) -> bool:
        """Bring an app to the foreground using monkey command."""
        result = await self.shell(
            f"monkey -p {package} -c android.intent.category.LAUNCHER 1 2>/dev/null"
        )
        if result is not None:
            _LOGGER.debug("Brought %s to foreground via monkey", package)
        return result is not None

    async def is_awake(self) -> bool | None:
        """Check if the device screen is on / awake via ADB.

        Exact-value mWakefulness= parse (R5); None means "unknown".
        """
        result = await self.shell("dumpsys power 2>/dev/null | grep mWakefulness")
        return parse_wakefulness(result)

    # -- Media session polling --

    async def poll_media_sessions(
        self, foreground_package: str | None = None, head_lines: int | None = 200
    ) -> MediaSessionInfo | None:
        """Query media sessions. Stores all sessions and returns the best match.

        Args:
            foreground_package: Prefer the session of the foreground app.
            head_lines: Truncate the dump to N lines (the historical default
                of 200). Pass None for the full dump — the ShieldConnection
                manager does so, since the cut silently drops sessions below
                it (plan blocker B4).
        """
        command = "dumpsys media_session 2>/dev/null"
        if head_lines is not None:
            command += f" | head -{head_lines}"
        dump = await self.shell(command)
        self._last_dump = dump if dump is not None else None
        if dump is None:
            return None
        self._all_sessions = parse_all_media_sessions(dump)
        self._last_info = parse_media_session(dump, foreground_package)
        return self._last_info

    @property
    def last_dump(self) -> str | None:
        """Raw output of the last media-session query (or None if it failed).

        Shared consumers (the ShieldConnectionManager) parse the "Audio
        playback" section from this same dump — no extra round trip.
        """
        return self._last_dump

    # -- Health check --

    async def health_check(self) -> bool:
        """Verify the connection is actually usable with a cheap shell round trip.

        Returns True only when the device answers the trivial ``echo ok`` with
        the expected body — a stronger signal than the ``connected`` flag,
        which can lag behind reality (the watchdog uses this to fire a
        reconnect instead of trusting a dead socket).
        """
        result = await self.shell("echo ok")
        return result is not None and result.strip() == "ok"

    # -- Running apps (recent visible tasks) --

    async def poll_running_apps(self) -> list[dict]:
        """Get running apps from Android's recent tasks.

        Returns a list of dicts with 'package' and 'foreground' keys,
        ordered by recency (most recent first). Only includes visible
        user apps, filtering out system UI, home, and invisible tasks.
        """
        result = await self.shell(
            "dumpsys activity recents 2>/dev/null "
            "| grep -E 'Recent #|realActivity|visible='"
        )
        if not result:
            return []

        apps = []
        seen = set()

        # System packages to exclude
        excluded = {
            "com.android.systemui",
            "com.android.tv.settings",
            "com.nvidia.diagtools",
            "android",
        }

        for line in result.split("\n"):
            stripped = line.strip()

            # Match recent task lines with visibility and package info
            if "Recent #" in stripped and "visible=true" in stripped:
                # Extract package from A=uid:package pattern
                match = re.search(r"A=\d+:(\S+)", stripped)
                if not match:
                    # Try I=package/activity pattern
                    match = re.search(r"I=(\S+)/", stripped)
                if match:
                    pkg = match.group(1)
                    if pkg not in seen and pkg not in excluded and not pkg.startswith("."):
                        seen.add(pkg)
                        apps.append({"package": pkg})

        return apps