mcp-adb/src/mixins/apps.py

"""Apps mixin for Android ADB MCP Server.

Provides tools for app management and launching.
"""

import re
from typing import Any

from fastmcp import Context
from fastmcp.contrib.mcp_mixin import mcp_resource, mcp_tool

from ..config import is_developer_mode
from .base import ADBBaseMixin

# Common Android intent flags (hex values for am start -f)
_INTENT_FLAGS: dict[str, int] = {
    "FLAG_ACTIVITY_NEW_TASK": 0x10000000,
    "FLAG_ACTIVITY_CLEAR_TOP": 0x04000000,
    "FLAG_ACTIVITY_SINGLE_TOP": 0x20000000,
    "FLAG_ACTIVITY_NO_HISTORY": 0x40000000,
    "FLAG_ACTIVITY_CLEAR_TASK": 0x00008000,
    "FLAG_ACTIVITY_EXCLUDE_FROM_RECENTS": 0x00800000,
    "FLAG_ACTIVITY_FORWARD_RESULT": 0x02000000,
    "FLAG_ACTIVITY_MULTIPLE_TASK": 0x08000000,
}


class AppsMixin(ADBBaseMixin):
    """Mixin for Android app management.

    Provides tools for:
    - Launching apps
    - Opening URLs
    - Listing packages (developer mode)
    - Installing/uninstalling apps (developer mode)
    """

    @mcp_tool()
    async def app_launch(
        self,
        package_name: str,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Launch an app by package name.

        Starts the main activity of the specified application.

        Common package names:
        - com.android.chrome - Chrome browser
        - com.android.settings - Settings
        - com.android.vending - Play Store
        - com.google.android.gm - Gmail
        - com.google.android.apps.maps - Google Maps

        Args:
            package_name: Android package name (e.g., com.android.chrome)
            device_id: Target device

        Returns:
            Success status
        """
        result = await self.run_shell_args(
            [
                "monkey",
                "-p",
                package_name,
                "-c",
                "android.intent.category.LAUNCHER",
                "1",
            ],
            device_id,
        )
        return {
            "success": result.success,
            "action": "launch",
            "package": package_name,
            "error": result.stderr if not result.success else None,
        }

    @mcp_tool()
    async def app_open_url(
        self,
        url: str,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Open a URL in the default browser.

        Launches the default browser and navigates to the URL.
        Supports http://, https://, and other URL schemes.

        Args:
            url: URL to open (e.g., https://example.com)
            device_id: Target device

        Returns:
            Success status
        """
        result = await self.run_shell_args(
            [
                "am",
                "start",
                "-a",
                "android.intent.action.VIEW",
                "-d",
                url,
            ],
            device_id,
        )
        return {
            "success": result.success,
            "action": "open_url",
            "url": url,
            "error": result.stderr if not result.success else None,
        }

    @mcp_tool()
    async def app_close(
        self,
        package_name: str,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Force stop an app.

        Stops the application and all its background services.

        Args:
            package_name: Package name to stop
            device_id: Target device

        Returns:
            Success status
        """
        result = await self.run_shell_args(
            ["am", "force-stop", package_name], device_id
        )
        return {
            "success": result.success,
            "action": "close",
            "package": package_name,
            "error": result.stderr if not result.success else None,
        }

    @mcp_tool()
    async def app_current(
        self,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Get the currently focused app.

        Returns the package name of the app currently in foreground.

        Args:
            device_id: Target device

        Returns:
            Current app package and activity
        """
        # Use plain "dumpsys window" — the focus fields live at the
        # top level, not inside the "windows" subsection on many devices
        result = await self.run_shell_args(["dumpsys", "window"], device_id)

        if result.success:
            package = None
            activity = None
            for line in result.stdout.split("\n"):
                if "mFocusedApp" in line or "mCurrentFocus" in line:
                    # Match both formats:
                    #   mFocusedApp=ActivityRecord{... com.pkg/.Act t123}
                    #   mCurrentFocus=Window{... com.pkg/com.pkg.Act}
                    match = re.search(r"([\w.]+)/([\w.]*\.?\w+)", line)
                    if match:
                        package = match.group(1)
                        activity = match.group(2)
                        break
            return {
                "success": True,
                "package": package,
                "activity": activity,
                "raw": result.stdout[:500] if not package else None,
            }

        return {
            "success": False,
            "error": result.stderr,
        }

    # === Developer Mode Tools ===

    @mcp_tool(
        tags={"developer"},
        annotations={"requires": "developer_mode"},
    )
    async def app_list_packages(
        self,
        filter_text: str | None = None,
        system_only: bool = False,
        third_party_only: bool = False,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """List installed packages.

        [DEVELOPER MODE] Retrieves all installed application packages.

        Args:
            filter_text: Filter packages containing this text
            system_only: Only show system packages
            third_party_only: Only show third-party (user installed) packages
            device_id: Target device

        Returns:
            List of package names
        """
        if not is_developer_mode():
            return {
                "success": False,
                "error": "Developer mode required",
            }

        cmd = ["pm", "list", "packages"]
        if system_only:
            cmd.append("-s")
        elif third_party_only:
            cmd.append("-3")

        result = await self.run_shell_args(cmd, device_id)

        if result.success:
            packages = []
            for line in result.stdout.split("\n"):
                if line.startswith("package:"):
                    pkg = line.replace("package:", "").strip()
                    if filter_text is None or filter_text.lower() in pkg.lower():
                        packages.append(pkg)

            return {
                "success": True,
                "packages": sorted(packages),
                "count": len(packages),
            }

        return {
            "success": False,
            "error": result.stderr,
        }

    @mcp_tool(
        tags={"developer"},
        annotations={"requires": "developer_mode"},
    )
    async def app_install(
        self,
        apk_path: str,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Install an APK file.

        [DEVELOPER MODE] Installs an APK from the host machine to the device.

        Args:
            apk_path: Path to APK file on host machine
            device_id: Target device

        Returns:
            Installation result
        """
        if not is_developer_mode():
            return {
                "success": False,
                "error": "Developer mode required",
            }

        result = await self.run_adb(["install", "-r", apk_path], device_id)
        return {
            "success": result.success,
            "action": "install",
            "apk": apk_path,
            "output": result.stdout,
            "error": result.stderr if not result.success else None,
        }

    @mcp_tool(
        tags={"developer"},
        annotations={"requires": "developer_mode"},
    )
    async def app_uninstall(
        self,
        ctx: Context,
        package_name: str,
        keep_data: bool = False,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Uninstall an app.

        [DEVELOPER MODE] Removes an application from the device.
        Requires user confirmation before proceeding.

        Args:
            ctx: MCP context for elicitation/logging
            package_name: Package to uninstall
            keep_data: Keep app data after uninstall
            device_id: Target device

        Returns:
            Uninstall result
        """
        if not is_developer_mode():
            return {
                "success": False,
                "error": "Developer mode required",
            }

        # Elicit confirmation
        await ctx.warning(f"Uninstall requested: {package_name}")

        data_note = " (keeping app data)" if keep_data else " (all data will be lost)"
        confirmation = await ctx.elicit(
            f"Are you sure you want to uninstall '{package_name}'?{data_note}",
            ["Yes, uninstall", "Cancel"],
        )

        if confirmation.action != "accept" or confirmation.content == "Cancel":
            await ctx.info("Uninstall cancelled by user")
            return {
                "success": False,
                "cancelled": True,
                "message": "Uninstall cancelled by user",
            }

        await ctx.info(f"Uninstalling {package_name}...")

        cmd = ["uninstall"]
        if keep_data:
            cmd.append("-k")
        cmd.append(package_name)

        result = await self.run_adb(cmd, device_id)

        if result.success:
            await ctx.info(f"Successfully uninstalled {package_name}")
        else:
            await ctx.error(f"Uninstall failed: {result.stderr}")

        return {
            "success": result.success,
            "action": "uninstall",
            "package": package_name,
            "kept_data": keep_data,
            "error": result.stderr if not result.success else None,
        }

    @mcp_tool(
        tags={"developer"},
        annotations={"requires": "developer_mode"},
    )
    async def app_clear_data(
        self,
        ctx: Context,
        package_name: str,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Clear app data and cache.

        [DEVELOPER MODE] Clears all data for an application (like a fresh
        install). Requires user confirmation before proceeding.

        Args:
            ctx: MCP context for elicitation/logging
            package_name: Package to clear
            device_id: Target device

        Returns:
            Clear result
        """
        if not is_developer_mode():
            return {
                "success": False,
                "error": "Developer mode required",
            }

        # Elicit confirmation
        await ctx.warning(f"Clear data requested: {package_name}")

        confirmation = await ctx.elicit(
            f"Are you sure you want to clear ALL data for "
            f"'{package_name}'? This includes login state, settings, "
            "saved files, and cache. The app will be reset to a "
            "fresh install state.",
            ["Yes, clear all data", "Cancel"],
        )

        if confirmation.action != "accept" or confirmation.content == "Cancel":
            await ctx.info("Clear data cancelled by user")
            return {
                "success": False,
                "cancelled": True,
                "message": "Clear data cancelled by user",
            }

        await ctx.info(f"Clearing data for {package_name}...")

        result = await self.run_shell_args(["pm", "clear", package_name], device_id)

        if result.success:
            await ctx.info(f"Successfully cleared data for {package_name}")
        else:
            await ctx.error(f"Clear data failed: {result.stderr}")

        return {
            "success": result.success,
            "action": "clear_data",
            "package": package_name,
            "error": result.stderr if not result.success else None,
        }

    @mcp_tool(
        tags={"developer"},
        annotations={"requires": "developer_mode"},
    )
    async def activity_start(
        self,
        component: str,
        action: str | None = None,
        data_uri: str | None = None,
        extras: dict[str, str] | None = None,
        flags: list[str] | None = None,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Start a specific activity with intent.

        [DEVELOPER MODE] Launch an activity with full intent control.
        More powerful than app_launch for deep linking and testing.

        Args:
            component: Activity component (e.g., "com.example/.MainActivity")
            action: Intent action (e.g., "android.intent.action.VIEW")
            data_uri: Data URI to pass to activity
            extras: Extra key-value pairs to include in intent
            flags: Intent flag names (e.g., ["FLAG_ACTIVITY_NEW_TASK"])
                   or hex values (e.g., ["0x10000000"])
            device_id: Target device

        Returns:
            Activity start result

        Examples:
            Start specific activity:
                component="com.example/.MainActivity"
            Deep link with data:
                component="com.example/.DeepLinkActivity"
                data_uri="myapp://product/123"
        """
        if not is_developer_mode():
            return {
                "success": False,
                "error": "Developer mode required",
            }

        cmd_args = ["am", "start"]

        if action:
            cmd_args.extend(["-a", action])

        if data_uri:
            cmd_args.extend(["-d", data_uri])

        # Resolve flag names to integer values for am start -f
        if flags:
            combined_flags = 0
            for flag in flags:
                flag_clean = flag.strip()
                if flag_clean.startswith("0x"):
                    combined_flags |= int(flag_clean, 16)
                elif flag_clean in _INTENT_FLAGS:
                    combined_flags |= _INTENT_FLAGS[flag_clean]
                elif flag_clean.isdigit():
                    combined_flags |= int(flag_clean)
            if combined_flags:
                cmd_args.extend(["-f", str(combined_flags)])

        if extras:
            for key, value in extras.items():
                if value.lower() in ("true", "false"):
                    cmd_args.extend(["--ez", key, value.lower()])
                elif re.match(r"^-?\d+$", value):
                    cmd_args.extend(["--ei", key, value])
                else:
                    cmd_args.extend(["--es", key, value])

        cmd_args.extend(["-n", component])

        result = await self.run_shell_args(cmd_args, device_id)

        return {
            "success": result.success,
            "action": "activity_start",
            "component": component,
            "intent_action": action,
            "data_uri": data_uri,
            "output": result.stdout,
            "error": result.stderr if not result.success else None,
        }

    @mcp_tool(
        tags={"developer"},
        annotations={"requires": "developer_mode"},
    )
    async def broadcast_send(
        self,
        action: str,
        extras: dict[str, str] | None = None,
        package: str | None = None,
        device_id: str | None = None,
    ) -> dict[str, Any]:
        """Send a broadcast intent.

        [DEVELOPER MODE] Sends a broadcast that can be received by
        BroadcastReceivers. Useful for testing and triggering app behavior.

        Args:
            action: Broadcast action (e.g., "com.example.MY_ACTION")
            extras: Extra key-value pairs to include
            package: Limit to specific package (optional)
            device_id: Target device

        Returns:
            Broadcast result

        Common system broadcasts (for testing receivers):
            - android.intent.action.AIRPLANE_MODE
            - android.intent.action.BATTERY_LOW
            - android.net.conn.CONNECTIVITY_CHANGE
        """
        if not is_developer_mode():
            return {
                "success": False,
                "error": "Developer mode required",
            }

        cmd_args = ["am", "broadcast", "-a", action]

        if package:
            cmd_args.extend(["-p", package])

        if extras:
            for key, value in extras.items():
                if value.lower() in ("true", "false"):
                    cmd_args.extend(["--ez", key, value.lower()])
                elif re.match(r"^-?\d+$", value):
                    cmd_args.extend(["--ei", key, value])
                else:
                    cmd_args.extend(["--es", key, value])

        result = await self.run_shell_args(cmd_args, device_id)

        return {
            "success": result.success,
            "action": "broadcast_send",
            "broadcast_action": action,
            "package": package,
            "output": result.stdout,
            "error": result.stderr if not result.success else None,
        }

    # === Resources ===

    @mcp_resource(uri="adb://apps/current")
    async def resource_current_app(self) -> dict[str, Any]:
        """Resource: Get currently focused app."""
        return await self.app_current()