"""Playwright-backed browser automation tools for Cowork.

The dependency is optional. If Playwright or its browser binaries are not installed, the
tools return a clear setup error instead of breaking engine construction.

Implementation note:
  The Playwright browser runs in a separate Python subprocess (browser_worker.py)
  because PyInstaller-packaged Python is incompatible with playwright's C extensions.
  The controller communicates with the worker via stdin/stdout JSON lines.
"""

from __future__ import annotations

import json
import os
import subprocess
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any, Callable, Optional

import aisuite as ai

from ..web.guard import check_url


def _find_tools_dir() -> Path:
    """查找 tools 目录（从当前模块路径向上找，或用环境变量）。"""
    # 1. 环境变量优先
    env_path = os.environ.get("OPENMESH_TOOLS_DIR", "")
    if env_path and Path(env_path).exists():
        return Path(env_path)

    # 2. 从模块路径向上找（打包后 coworker 在 _internal/ 里，tools 在上层）
    try:
        here = Path(__file__).resolve().parent
        for p in [here, here.parent, here.parent.parent]:
            candidate = p / "tools"
            if candidate.exists():
                return candidate
    except Exception:
        pass

    return Path("tools")


def _find_worker_script(tools_dir: Path) -> Path:
    """查找 browser_worker.py 脚本。

    查找顺序：
    1. 环境变量 BROWSER_WORKER_PATH
    2. tools/python/browser_worker.py（运行时预放置的位置）
    3. 当前模块同目录（开发环境，PYZ 未打包时）
    """
    # 1. 环境变量
    env_path = os.environ.get("BROWSER_WORKER_PATH", "")
    if env_path and Path(env_path).exists():
        return Path(env_path)

    # 2. tools/python/browser_worker.py（打包后放这里最稳）
    candidate = tools_dir / "python" / "browser_worker.py"
    if candidate.exists():
        return candidate

    # 3. 当前模块同目录
    here = Path(__file__).resolve().parent
    candidate = here / "browser_worker.py"
    if candidate.exists():
        return candidate

    # 4. 兜底返回同目录路径（报错时能看到在哪找的）
    return here / "browser_worker.py"


_TOOLS_DIR = _find_tools_dir()
_WORKER_SCRIPT = _find_worker_script(_TOOLS_DIR)


def _meta(
    name: str, *, approval: bool = False, capabilities: Optional[list[str]] = None
):
    return ai.ToolMetadata(
        name=name,
        category="connector",
        risk_level="medium" if approval else "low",
        capabilities=capabilities or ["browser"],
        requires_approval=approval,
    )


def _schema(
    name: str, description: str, properties: dict[str, Any], required: list[str]
) -> dict[str, Any]:
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }


def _attach(fn: Callable[..., Any], schema: dict[str, Any], *, approval: bool = True):
    from .tool_defs import approval_for_tool

    name = schema["function"]["name"]
    # §36: the tool registry's read/write kind wins for registered tools — reads never gate.
    approval = approval_for_tool(name, default=approval)
    fn.__coworker_schema__ = schema
    fn.__aisuite_tool_metadata__ = _meta(name, approval=approval)
    fn.__doc__ = schema["function"]["description"]
    return fn


class _BrowserController:
    """浏览器控制器 — 通过子进程 worker 运行 Playwright。

    为什么用子进程：
      后端是 PyInstaller 打包的 Python，playwright 的 C 扩展在里面会崩；
      tools/python/ 是原生 Python，playwright + .venv 里的包能正常运行。
    """

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._executor = ThreadPoolExecutor(
            max_workers=1, thread_name_prefix="coworker-browser"
        )
        self._proc: Optional[subprocess.Popen] = None
        self._request_id = 0
        self._error: Optional[str] = None
        self._state: dict[str, Any] = {
            "open": False,
            "url": "",
            "title": "",
            "status": "closed",
            "last_action": "",
            "last_result": "",
            "last_error": "",
            "screenshot_data_url": "",
            "updated_at": None,
            "controls": [],
        }

    def _touch(self, **changes: Any) -> None:
        self._state.update(changes)
        self._state["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

    def _ensure_worker(self) -> Optional[dict[str, Any]]:
        """确保 worker 子进程已启动。返回 None 表示成功，返回 dict 表示错误。"""
        if self._proc is not None and self._proc.poll() is None:
            return None

        self._error = None
        try:
            python_exe = _TOOLS_DIR / "python" / "python.exe"
            if not python_exe.exists():
                return {
                    "error": "Browser automation requires tools/python/python.exe (not found).",
                    "details": f"Expected at: {python_exe}",
                }

            env = os.environ.copy()
            env["OPENMESH_TOOLS_DIR"] = str(_TOOLS_DIR)
            env["PYTHONIOENCODING"] = "utf-8"
            env["PYTHONUNBUFFERED"] = "1"

            self._proc = subprocess.Popen(
                [str(python_exe), str(_WORKER_SCRIPT)],
                stdin=subprocess.PIPE,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                env=env,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            )

            # 等待 ready 信号
            ready_line = self._proc.stdout.readline()
            if not ready_line:
                stderr = self._proc.stderr.read().decode("utf-8", errors="replace")
                return {
                    "error": "Browser worker failed to start.",
                    "details": stderr[:500],
                }
            try:
                ready = json.loads(ready_line.decode("utf-8").strip())
                if ready.get("status") != "ready":
                    return {
                        "error": "Browser worker did not report ready.",
                        "details": str(ready),
                    }
            except json.JSONDecodeError:
                return {
                    "error": "Browser worker sent invalid ready signal.",
                    "details": ready_line.decode("utf-8", errors="replace")[:200],
                }

            return None
        except Exception as exc:
            self._error = str(exc)
            return {
                "error": (
                    "Interactive browser automation requires Playwright. "
                    "Make sure tools/python/ and tools/playwright/chromium-1234/ are present."
                ),
                "details": str(exc),
            }

    def _send_command(self, action: str, params: Optional[dict] = None) -> dict[str, Any]:
        """向 worker 发送一条命令并等待响应。"""
        with self._lock:
            err = self._ensure_worker()
            if err:
                return err

            self._request_id += 1
            req_id = self._request_id
            req = json.dumps(
                {"id": req_id, "action": action, "params": params or {}},
                ensure_ascii=False,
            ) + "\n"

            try:
                self._proc.stdin.write(req.encode("utf-8"))
                self._proc.stdin.flush()
            except Exception as exc:
                self._kill_worker()
                return {"error": f"failed to send to browser worker: {exc}"}

            try:
                line = self._proc.stdout.readline()
                if not line:
                    stderr = self._proc.stderr.read().decode("utf-8", errors="replace")
                    self._kill_worker()
                    return {
                        "error": "browser worker exited unexpectedly",
                        "details": stderr[:500],
                    }
                resp = json.loads(line.decode("utf-8").strip())
                if resp.get("id") != req_id:
                    return {"error": f"response id mismatch: {resp.get('id')} vs {req_id}"}
                if "error" in resp:
                    self._touch(last_action=action, last_result="error", last_error=resp["error"])
                    return {"error": resp["error"]}
                self._touch(last_action=action, last_result="ok", last_error="")
                # 同步状态
                if "url" in resp:
                    self._state["url"] = resp.get("url", "")
                if "title" in resp:
                    self._state["title"] = resp.get("title", "")
                if "controls" in resp:
                    self._state["controls"] = resp.get("controls", [])
                if "screenshot_data_url" in resp:
                    self._state["screenshot_data_url"] = resp["screenshot_data_url"]
                if action == "close":
                    self._state["open"] = False
                    self._state["status"] = "closed"
                    self._kill_worker()
                elif action in ("open_url", "read_page", "click", "type", "select", "wait"):
                    self._state["open"] = True
                    self._state["status"] = "open"
                return resp
            except Exception as exc:
                self._kill_worker()
                return {"error": f"failed to read browser worker response: {exc}"}

    def _kill_worker(self) -> None:
        try:
            if self._proc and self._proc.poll() is None:
                self._proc.terminate()
                self._proc.wait(timeout=3)
        except Exception:
            try:
                if self._proc and self._proc.poll() is None:
                    self._proc.kill()
            except Exception:
                pass
        finally:
            self._proc = None

    def _submit(self, fn: Callable[[], dict[str, Any]]) -> dict[str, Any]:
        return self._executor.submit(fn).result()

    def close(self) -> dict[str, Any]:
        def _do():
            with self._lock:
                if self._proc is None:
                    return {"ok": True}
                return self._send_command("close")
        return self._submit(_do)

    def state(self) -> dict[str, Any]:
        def _do():
            with self._lock:
                if self._proc is None:
                    return dict(self._state)
                resp = self._send_command("state")
                if "error" in resp:
                    return dict(self._state)
                return {**dict(self._state), **{k: v for k, v in resp.items() if k != "id"}}
        return self._submit(_do)

    def screenshot(self) -> dict[str, Any]:
        def _do():
            with self._lock:
                resp = self._send_command("screenshot")
                if "error" in resp:
                    return resp
                return {"ok": True, **dict(self._state)}
        return self._submit(_do)

    def call(self, action: str, fn: Callable[[Any], dict[str, Any]]) -> dict[str, Any]:
        """兼容旧的 call 接口 — 直接通过 worker 执行 action。

        注意：fn 参数被忽略，所有逻辑在 worker 里实现。
        action 名直接映射到 worker 的 action 名。
        """
        def _do():
            with self._lock:
                # 从 fn 的闭包或通过 action 名构造 params
                # 为了兼容旧代码，这里用 action 名直接发命令
                # 参数通过额外机制传递（见下方每个工具函数的实现）
                return self._send_command(action)
        return self._submit(_do)


_BROWSER = _BrowserController()


def browser_state() -> dict[str, Any]:
    return _BROWSER.state()


def browser_take_screenshot() -> dict[str, Any]:
    return _BROWSER.screenshot()


def browser_close_session() -> dict[str, Any]:
    return _BROWSER.close()


_BROWSER = _BrowserController()


def browser_state() -> dict[str, Any]:
    return _BROWSER.state()


def browser_take_screenshot() -> dict[str, Any]:
    return _BROWSER.screenshot()


def browser_close_session() -> dict[str, Any]:
    return _BROWSER.close()


def _worker_call(action: str, params: Optional[dict] = None) -> dict[str, Any]:
    """通过 worker 子进程执行浏览器操作。"""
    # 用 _send_command，但它是私有方法——通过 _BROWSER 的内部机制调用
    return _BROWSER._send_command(action, params)


def make_browser_automation_tools(
    *, roots: Optional[list[Any]] = None
) -> list[Callable[..., Any]]:
    tools: list[Callable[..., Any]] = []

    def _readable_source(raw: str) -> tuple[Any, dict[str, Any] | None]:
        """A local file to upload, resolved inside a granted root (OPE-122)."""
        allowed = [r.path for r in (roots or [])]
        if not allowed:
            return None, {"error": "no session directory is available to upload from"}
        path = Path(str(raw)).expanduser().resolve()
        if not any(path.is_relative_to(root) for root in allowed):
            return None, {"error": f"{path} is outside the session's directories"}
        return path, None

    def _writable_target(raw: str) -> tuple[Any, dict[str, Any] | None]:
        """Where a screenshot may land: inside a WRITABLE granted root."""
        writable = [r.path for r in (roots or []) if r.writable]
        if not writable:
            return None, {"error": "no writable session directory for the screenshot"}
        path = Path(str(raw)).expanduser().resolve()
        if not any(path.is_relative_to(root) for root in writable):
            return None, {
                "error": f"{path} is outside the session's writable directories"
            }
        return path, None

    def browser_open_url(
        url: str, wait_until: str = "domcontentloaded"
    ) -> dict[str, Any]:
        if not url.lower().startswith(("http://", "https://")):
            return {"error": "url must start with http:// or https://"}
        # URL guard (same as web_fetch — defense in depth)
        blocked = check_url(url)
        if blocked:
            return {"error": blocked}
        return _worker_call("open_url", {"url": url, "wait_until": wait_until})

    browser_open_url.__name__ = "browser_open_url"
    tools.append(
        _attach(
            browser_open_url,
            _schema(
                "browser_open_url",
                "Open a URL in the local Playwright browser session.",
                {"url": {"type": "string"}, "wait_until": {"type": "string"}},
                ["url"],
            ),
            approval=True,
        )
    )

    def browser_read_page(max_chars: int = 20000) -> dict[str, Any]:
        return _worker_call("read_page", {"max_chars": max_chars})

    browser_read_page.__name__ = "browser_read_page"
    tools.append(
        _attach(
            browser_read_page,
            _schema(
                "browser_read_page",
                "Read the current page: its text plus visible controls and selector "
                "hints (for browser_click/browser_type). Not an image — use "
                "browser_screenshot for pixels.",
                {"max_chars": {"type": "integer"}},
                [],
            ),
            approval=True,
        )
    )

    def browser_click(target: str) -> dict[str, Any]:
        return _worker_call("click", {"target": target})

    browser_click.__name__ = "browser_click"
    tools.append(
        _attach(
            browser_click,
            _schema(
                "browser_click",
                "Click a visible page element by CSS selector, text=label, role=button:Name, or text fallback. Requires approval.",
                {"target": {"type": "string"}},
                ["target"],
            ),
            approval=True,
        )
    )

    def browser_type(target: str, text: str, clear: bool = True) -> dict[str, Any]:
        return _worker_call("type", {"target": target, "text": text, "clear": clear})

    browser_type.__name__ = "browser_type"
    tools.append(
        _attach(
            browser_type,
            _schema(
                "browser_type",
                "Fill or type into an input, textarea, or editable element. Requires approval.",
                {
                    "target": {"type": "string"},
                    "text": {"type": "string"},
                    "clear": {"type": "boolean"},
                },
                ["target", "text"],
            ),
            approval=True,
        )
    )

    def browser_select(target: str, value: str) -> dict[str, Any]:
        return _worker_call("select", {"target": target, "value": value})

    browser_select.__name__ = "browser_select"
    tools.append(
        _attach(
            browser_select,
            _schema(
                "browser_select",
                "Select an option in a dropdown by selector and option value/label. Requires approval.",
                {"target": {"type": "string"}, "value": {"type": "string"}},
                ["target", "value"],
            ),
            approval=True,
        )
    )

    def browser_upload_file(target: str, path: str) -> dict[str, Any]:
        file_path, err = _readable_source(path)
        if err:
            return err
        if not file_path.exists():
            return {"error": f"file not found: {file_path}"}
        return _worker_call("upload_file", {"target": target, "path": str(file_path)})

    browser_upload_file.__name__ = "browser_upload_file"
    tools.append(
        _attach(
            browser_upload_file,
            _schema(
                "browser_upload_file",
                "Upload a local file through a file input. Requires approval.",
                {"target": {"type": "string"}, "path": {"type": "string"}},
                ["target", "path"],
            ),
            approval=True,
        )
    )

    def browser_wait(milliseconds: int = 1000, target: str = "") -> dict[str, Any]:
        return _worker_call(
            "wait", {"milliseconds": milliseconds, "target": target}
        )

    browser_wait.__name__ = "browser_wait"
    tools.append(
        _attach(
            browser_wait,
            _schema(
                "browser_wait",
                "Wait for a duration or for a target element to appear.",
                {"milliseconds": {"type": "integer"}, "target": {"type": "string"}},
                [],
            ),
            approval=True,
        )
    )

    def browser_screenshot(path: str = "") -> dict[str, Any]:
        params = {}
        if path:
            _target, target_err = _writable_target(path)
            if target_err:
                return target_err
            params["path"] = str(_target)
        return _worker_call("screenshot", params)

    browser_screenshot.__name__ = "browser_screenshot"
    tools.append(
        _attach(
            browser_screenshot,
            _schema(
                "browser_screenshot",
                "Save a full-page screenshot of the current browser page and return the local path.",
                {"path": {"type": "string"}},
                [],
            ),
            approval=True,
        )
    )

    def browser_close() -> dict[str, Any]:
        return browser_close_session()

    browser_close.__name__ = "browser_close"
    tools.append(
        _attach(
            browser_close,
            _schema(
                "browser_close",
                "Close the local Playwright browser session.",
                {},
                [],
            ),
            approval=True,
        )
    )

    return tools