#!/usr/bin/env python3
"""Агент Джарвиса для компьютера (Windows / macOS / Linux).

Подключается к серверу Джарвиса и выполняет действия на этом компьютере:
папки и файлы (только внутри домашней папки), открытие сайтов и программ,
снимок экрана.

    python jarvis_agent.py --server https://jarvis.example.com --token ВАШ_ТОКЕН
    python jarvis_agent.py --check          # проверить разрешения ОС

Настройки можно положить в agent.env рядом со скриптом:
    JARVIS_SERVER=https://jarvis.example.com
    JARVIS_TOKEN=...
"""
from __future__ import annotations

import argparse
import asyncio
import base64
import getpass
import io
import json
import logging
import os
import platform
import shutil
import socket
import subprocess
import sys
import webbrowser
from pathlib import Path
from urllib.parse import urlparse

VERSION = "1.1.0"
MAX_READ_CHARS = 20000
MAX_WRITE_CHARS = 2_000_000
SYSTEM = platform.system()  # Windows | Darwin | Linux

log = logging.getLogger("jarvis-agent")


# ---------------------------------------------------------------------------
#  Пути
# ---------------------------------------------------------------------------
def _windows_known_folder(csidl: int) -> Path | None:
    try:
        import ctypes
        from ctypes import wintypes

        buf = ctypes.create_unicode_buffer(wintypes.MAX_PATH)
        if ctypes.windll.shell32.SHGetFolderPathW(None, csidl, None, 0, buf) == 0:  # type: ignore[attr-defined]
            return Path(buf.value)
    except Exception:  # noqa: BLE001
        pass
    return None


def _xdg_dir(name: str) -> Path | None:
    if shutil.which("xdg-user-dir"):
        try:
            out = subprocess.run(["xdg-user-dir", name], capture_output=True, text=True, timeout=3).stdout.strip()
            if out and Path(out) != Path.home():
                return Path(out)
        except Exception:  # noqa: BLE001
            pass
    return None


def known_folder(kind: str) -> Path:
    home = Path.home()
    csidl = {"desktop": 0x10, "documents": 0x05, "pictures": 0x27, "music": 0x0D, "videos": 0x0E}
    english = {"desktop": "Desktop", "documents": "Documents", "downloads": "Downloads",
               "pictures": "Pictures", "music": "Music", "videos": "Videos"}
    russian = {"desktop": "Рабочий стол", "documents": "Документы", "downloads": "Загрузки",
               "pictures": "Изображения", "music": "Музыка", "videos": "Видео"}
    if kind == "home":
        return home
    if SYSTEM == "Windows" and kind in csidl:
        found = _windows_known_folder(csidl[kind])
        if found and found.exists():
            return found
    if SYSTEM == "Linux":
        found = _xdg_dir(kind.upper())
        if found:
            return found
    candidates = [home / english[kind], home / "OneDrive" / english[kind], home / russian[kind],
                  home / "OneDrive" / russian[kind]]
    for path in candidates:
        if path.exists():
            return path
    return candidates[0]


ALIASES = {
    "desktop": "desktop", "рабочий стол": "desktop", "рабочийстол": "desktop", "стол": "desktop",
    "documents": "documents", "документы": "documents", "мои документы": "documents",
    "downloads": "downloads", "загрузки": "downloads",
    "pictures": "pictures", "изображения": "pictures", "картинки": "pictures",
    "music": "music", "музыка": "music",
    "videos": "videos", "видео": "videos",
    "home": "home", "~": "home", "домашняя папка": "home",
}


class Sandbox:
    """Все пути — только внутри корня (по умолчанию домашняя папка пользователя)."""

    def __init__(self, root: Path | None = None):
        self.root = (root or Path.home()).expanduser().resolve()

    def resolve(self, raw: str) -> Path:
        raw = (raw or "").strip().strip('"').strip("'").replace("\\", "/")
        if not raw:
            raise ValueError("Не указан путь.")
        head, _, tail = raw.partition("/")
        alias = ALIASES.get(head.strip().lower())
        if alias:
            base = known_folder(alias)
            path = base / tail if tail else base
        else:
            path = Path(raw).expanduser()
            if not path.is_absolute():
                path = self.root / path
        path = path.resolve()
        if path != self.root and self.root not in path.parents:
            raise PermissionError(f"Доступ разрешён только внутри {self.root}")
        return path


# ---------------------------------------------------------------------------
#  Действия
# ---------------------------------------------------------------------------
def _open_with_system(target: str) -> None:
    if SYSTEM == "Windows":
        os.startfile(target)  # type: ignore[attr-defined]  # noqa: S606
    elif SYSTEM == "Darwin":
        subprocess.Popen(["open", target])  # noqa: S603,S607
    else:
        subprocess.Popen(["xdg-open", target], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)  # noqa: S603,S607


class Actions:
    def __init__(self, sandbox: Sandbox):
        self.sandbox = sandbox

    def names(self) -> list[str]:
        return ["open_url", "screenshot", "create_folder", "list_dir", "read_file", "write_file",
                "open_path", "system_info"]

    def system_info(self) -> dict:
        return {
            "platform": f"{SYSTEM} {platform.release()}",
            "hostname": socket.gethostname(),
            "user": getpass.getuser(),
            "home": str(Path.home()),
            "desktop": str(known_folder("desktop")),
        }

    def open_url(self, url: str) -> dict:
        parsed = urlparse(url or "")
        if parsed.scheme not in ("http", "https") or not parsed.netloc:
            raise ValueError("Можно открывать только http(s)-ссылки.")
        if not webbrowser.open(url, new=2):
            _open_with_system(url)
        return {"opened": url}

    def screenshot(self) -> dict:
        try:
            import mss  # type: ignore
            from PIL import Image  # type: ignore
        except ImportError as exc:
            raise RuntimeError("Установите зависимости агента: pip install -r requirements.txt") from exc
        with mss.mss() as sct:
            monitor = sct.monitors[1] if len(sct.monitors) > 1 else sct.monitors[0]
            shot = sct.grab(monitor)
            img = Image.frombytes("RGB", shot.size, shot.bgra, "raw", "BGRX")
        img.thumbnail((1600, 1600))
        buf = io.BytesIO()
        img.save(buf, format="JPEG", quality=80)
        return {"image": base64.b64encode(buf.getvalue()).decode(), "width": img.width, "height": img.height}

    def create_folder(self, path: str) -> dict:
        target = self.sandbox.resolve(path)
        existed = target.exists()
        if existed and not target.is_dir():
            raise FileExistsError(f"Уже есть файл с таким именем: {target}")
        target.mkdir(parents=True, exist_ok=True)
        return {"path": str(target), "created": not existed, "already_existed": existed}

    def list_dir(self, path: str = "home", limit: int = 200) -> dict:
        target = self.sandbox.resolve(path)
        if not target.is_dir():
            raise NotADirectoryError(f"Это не папка: {target}")
        items = []
        for entry in sorted(target.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower()))[:limit]:
            if entry.name.startswith("."):
                continue
            try:
                size = entry.stat().st_size if entry.is_file() else None
            except OSError:
                size = None
            items.append({"name": entry.name, "type": "folder" if entry.is_dir() else "file", "size": size})
        return {"path": str(target), "items": items}

    def read_file(self, path: str) -> dict:
        target = self.sandbox.resolve(path)
        if not target.is_file():
            raise FileNotFoundError(f"Файл не найден: {target}")
        if target.suffix.lower() == ".docx":
            text = _read_docx(target)
        else:
            text = target.read_text(encoding="utf-8", errors="replace")
        return {"path": str(target), "content": text[:MAX_READ_CHARS], "truncated": len(text) > MAX_READ_CHARS}

    def write_file(self, path: str, content: str, mode: str = "create") -> dict:
        target = self.sandbox.resolve(path)
        if len(content or "") > MAX_WRITE_CHARS:
            raise ValueError("Слишком большой текст.")
        if mode not in ("create", "overwrite", "append"):
            mode = "create"
        if mode == "create" and target.exists():
            raise FileExistsError(f"Файл уже существует: {target}. Используйте mode=overwrite или append.")
        target.parent.mkdir(parents=True, exist_ok=True)
        with target.open("a" if mode == "append" else "w", encoding="utf-8") as fh:
            fh.write(content or "")
        return {"path": str(target), "mode": mode, "chars": len(content or "")}

    def open_path(self, path: str) -> dict:
        target = self.sandbox.resolve(path)
        if not target.exists():
            raise FileNotFoundError(f"Не найдено: {target}")
        _open_with_system(str(target))
        return {"opened": str(target)}

    def run(self, name: str, args: dict) -> dict:
        if name not in self.names():
            raise ValueError(f"Неизвестное действие: {name}")
        return getattr(self, name)(**(args or {}))


def _read_docx(path: Path) -> str:
    import re
    import zipfile

    with zipfile.ZipFile(path) as zf:
        xml = zf.read("word/document.xml").decode("utf-8", errors="replace")
    xml = re.sub(r"</w:p>", "\n", xml)
    return re.sub(r"<[^>]+>", "", xml)


# ---------------------------------------------------------------------------
#  Проверка разрешений ОС
# ---------------------------------------------------------------------------
MAC_PANES = {
    "screen": "x-apple.systempreferences:com.apple.preference.security?Privacy_ScreenCapture",
    "files": "x-apple.systempreferences:com.apple.preference.security?Privacy_FilesAndFolders",
    "full_disk": "x-apple.systempreferences:com.apple.preference.security?Privacy_AllFiles",
    "automation": "x-apple.systempreferences:com.apple.preference.security?Privacy_Automation",
}


def check_permissions(actions: Actions, open_settings: bool = True) -> bool:
    ok = True
    info = actions.system_info()
    print(f"Система: {info['platform']}, пользователь {info['user']}")
    print(f"Рабочий стол: {info['desktop']}")

    # 1. Файлы: создать и удалить тестовую папку на рабочем столе.
    probe = Path(info["desktop"]) / ".jarvis_access_check"
    try:
        probe.mkdir(parents=True, exist_ok=True)
        (probe / "ok.txt").write_text("ok", encoding="utf-8")
        shutil.rmtree(probe)
        print("✅ Файлы: запись на рабочий стол работает")
    except Exception as exc:  # noqa: BLE001
        ok = False
        print(f"❌ Файлы: нет доступа к рабочему столу ({exc})")
        if SYSTEM == "Darwin":
            print("   macOS: Настройки → Конфиденциальность и безопасность → Файлы и папки → "
                  "разрешите Терминалу/Python доступ к «Рабочий стол» и «Документы».")
            if open_settings:
                subprocess.run(["open", MAC_PANES["files"]], check=False)

    # 2. Экран.
    try:
        shot = actions.screenshot()
        print(f"✅ Экран: снимок {shot['width']}×{shot['height']}")
        if SYSTEM == "Darwin":
            print("   macOS: если на снимке только обои без окон — включите Терминал/Python в "
                  "Настройки → Конфиденциальность и безопасность → Запись экрана и перезапустите агента.")
            if open_settings:
                subprocess.run(["open", MAC_PANES["screen"]], check=False)
    except Exception as exc:  # noqa: BLE001
        ok = False
        print(f"❌ Экран: {exc}")
        if SYSTEM == "Linux" and os.environ.get("XDG_SESSION_TYPE") == "wayland":
            print("   Linux/Wayland: снимок экрана через mss работает только в X11-сессии (или XWayland).")

    # 3. Браузер.
    browser = None
    try:
        browser = webbrowser.get().name
    except webbrowser.Error:
        pass
    if browser or SYSTEM in ("Windows", "Darwin"):
        print(f"✅ Браузер: {browser or 'системный по умолчанию'}")
    else:
        ok = False
        print("❌ Браузер: не найден браузер по умолчанию (установите xdg-utils)")

    print("ℹ️  Микрофон выдаётся в самом браузере: значок замка в адресной строке → Микрофон → Разрешить.")
    return ok


# ---------------------------------------------------------------------------
#  Связь с сервером
# ---------------------------------------------------------------------------
def ws_url(server: str, token: str) -> str:
    server = server.strip().rstrip("/")
    if server.startswith("https://"):
        server = "wss://" + server[len("https://"):]
    elif server.startswith("http://"):
        server = "ws://" + server[len("http://"):]
    elif not server.startswith(("ws://", "wss://")):
        server = "wss://" + server
    if not server.endswith("/agent"):
        server += "/agent"
    from urllib.parse import quote

    return f"{server}?token={quote(token)}"


def ssl_context(url: str):
    """Системные сертификаты + certifi.

    Python с python.org на macOS не видит сертификаты Связки ключей и падает с
    CERTIFICATE_VERIFY_FAILED; certifi закрывает эту проблему на всех ОС.
    """
    if not url.startswith("wss://"):
        return None
    import ssl

    ctx = ssl.create_default_context()
    try:
        import certifi  # type: ignore

        ctx.load_verify_locations(cafile=certifi.where())
    except ImportError:
        log.warning("certifi не установлен: pip install -r requirements.txt")
    return ctx


def _connect(url: str):
    try:
        from websockets.asyncio.client import connect  # websockets >= 13
    except ImportError:  # pragma: no cover
        from websockets import connect  # type: ignore
    kwargs = {"max_size": 32 * 1024 * 1024, "ping_interval": 20, "ping_timeout": 20, "open_timeout": 15}
    ctx = ssl_context(url)
    if ctx is not None:
        kwargs["ssl"] = ctx
    return connect(url, **kwargs)


async def handle_call(ws, actions: Actions, msg: dict) -> None:
    name, args = msg.get("name", ""), msg.get("args") or {}
    shown = {k: (v[:60] + "…" if isinstance(v, str) and len(v) > 60 else v) for k, v in args.items()}
    log.info("→ %s %s", name, shown)
    try:
        result = await asyncio.to_thread(actions.run, name, args)
        reply = {"type": "result", "id": msg.get("id"), "ok": True, "result": result}
    except Exception as exc:  # noqa: BLE001
        log.warning("✗ %s: %s", name, exc)
        reply = {"type": "result", "id": msg.get("id"), "ok": False, "error": f"{exc.__class__.__name__}: {exc}"}
    await ws.send(json.dumps(reply, ensure_ascii=False))


def ui_url(server: str, token: str) -> str:
    from urllib.parse import quote

    base = server.strip().rstrip("/")
    if base.startswith("wss://"):
        base = "https://" + base[len("wss://"):]
    elif base.startswith("ws://"):
        base = "http://" + base[len("ws://"):]
    elif not base.startswith(("http://", "https://")):
        base = "https://" + base
    if base.endswith("/agent"):
        base = base[: -len("/agent")]
    return f"{base}/?token={quote(token)}" if token else f"{base}/"


async def run_agent(server: str, token: str, actions: Actions, open_ui: bool = False) -> None:
    url = ws_url(server, token)
    delay = 2
    while True:
        try:
            async with _connect(url) as ws:
                hello = {"type": "hello", "version": VERSION, "tools": actions.names(), **actions.system_info()}
                await ws.send(json.dumps(hello, ensure_ascii=False))
                async for raw in ws:
                    msg = json.loads(raw)
                    if msg.get("type") == "auth_error":
                        log.error("Сервер отклонил токен. Проверьте --token (JARVIS_TOKEN на сервере).")
                        return
                    if msg.get("type") == "replaced":
                        log.warning("Подключился другой агент Джарвиса с этим токеном — этот останавливается.")
                        return
                    if msg.get("type") == "welcome":
                        log.info("✅ Подключено к %s. Джарвис может работать с этим компьютером.", server)
                        delay = 2
                        if open_ui:
                            open_ui = False
                            webbrowser.open(ui_url(server, token), new=2)
                    elif msg.get("type") == "call":
                        asyncio.create_task(handle_call(ws, actions, msg))
            log.warning("Соединение закрыто сервером")
        except (OSError, asyncio.TimeoutError) as exc:
            log.warning("Нет связи с сервером (%s)", exc)
        except Exception as exc:  # noqa: BLE001
            log.warning("Соединение прервано: %s", exc)
        log.info("Переподключение через %s с…", delay)
        await asyncio.sleep(delay)
        delay = min(delay * 2, 30)


# ---------------------------------------------------------------------------
#  Автозапуск при входе в систему
# ---------------------------------------------------------------------------
LAUNCH_LABEL = "com.jarvis.agent"


def _mac_plist_path() -> Path:
    return Path.home() / "Library" / "LaunchAgents" / f"{LAUNCH_LABEL}.plist"


def _windows_startup_file() -> Path:
    appdata = Path(os.environ.get("APPDATA") or Path.home() / "AppData" / "Roaming")
    return appdata / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Startup" / "Jarvis Agent.vbs"


def _linux_autostart_file() -> Path:
    return Path.home() / ".config" / "autostart" / "jarvis-agent.desktop"


def _save_agent_env(here: Path, server: str, token: str) -> Path:
    env = here / "agent.env"
    env.write_text(f"JARVIS_SERVER={server}\nJARVIS_TOKEN={token}\n", encoding="utf-8")
    try:
        env.chmod(0o600)
    except OSError:
        pass
    return env


def install_autostart(server: str, token: str) -> str:
    """Регистрирует агента на запуск при входе в систему (без окна терминала)."""
    here = Path(__file__).resolve().parent
    script = Path(__file__).resolve()
    _save_agent_env(here, server, token)  # токен — в файле с правами 600, а не в аргументах процесса
    python = Path(sys.executable)

    if SYSTEM == "Darwin":
        import plistlib

        env = {"PYTHONUNBUFFERED": "1"}
        try:
            import certifi  # type: ignore

            env["SSL_CERT_FILE"] = certifi.where()
        except ImportError:
            pass
        logs = Path.home() / "Library" / "Logs" / "jarvis-agent.log"
        plist = {
            "Label": LAUNCH_LABEL,
            "ProgramArguments": [str(python), str(script), "--open-ui"],
            "WorkingDirectory": str(here),
            "EnvironmentVariables": env,
            "RunAtLoad": True,
            "KeepAlive": {"SuccessfulExit": False},
            "ProcessType": "Interactive",
            "LimitLoadToSessionType": "Aqua",
            "StandardOutPath": str(logs),
            "StandardErrorPath": str(logs),
        }
        path = _mac_plist_path()
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("wb") as fh:
            plistlib.dump(plist, fh)
        domain = f"gui/{os.getuid()}"  # type: ignore[attr-defined]
        subprocess.run(["launchctl", "bootout", domain, str(path)], capture_output=True, check=False)
        result = subprocess.run(["launchctl", "bootstrap", domain, str(path)], capture_output=True, text=True, check=False)
        if result.returncode != 0:
            subprocess.run(["launchctl", "load", "-w", str(path)], capture_output=True, check=False)
        return f"{path} (логи: {logs})"

    if SYSTEM == "Windows":
        pythonw = python.with_name("pythonw.exe")
        exe = pythonw if pythonw.exists() else python
        path = _windows_startup_file()
        path.parent.mkdir(parents=True, exist_ok=True)
        cmd = f'"{exe}" "{script}" --open-ui'
        vbs = (
            'Set sh = CreateObject("WScript.Shell")\r\n'
            f'sh.CurrentDirectory = "{here}"\r\n'
            f'sh.Run "{cmd.replace(chr(34), chr(34) * 2)}", 0, False\r\n'
        )
        path.write_text(vbs, encoding="utf-16")
        subprocess.Popen(["wscript", str(path)])  # noqa: S603,S607 — запускаем сразу
        return str(path)

    path = _linux_autostart_file()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        "[Desktop Entry]\nType=Application\nName=Jarvis Agent\n"
        f'Exec="{python}" "{script}" --open-ui\nPath={here}\nX-GNOME-Autostart-enabled=true\n',
        encoding="utf-8",
    )
    subprocess.Popen([str(python), str(script), "--open-ui"], cwd=here,  # noqa: S603
                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
    return str(path)


def uninstall_autostart() -> str:
    if SYSTEM == "Darwin":
        path = _mac_plist_path()
        subprocess.run(["launchctl", "bootout", f"gui/{os.getuid()}", str(path)],  # type: ignore[attr-defined]
                       capture_output=True, check=False)
    elif SYSTEM == "Windows":
        path = _windows_startup_file()
    else:
        path = _linux_autostart_file()
    if path.exists():
        path.unlink()
        return f"удалён {path}"
    return "автозапуск не был установлен"


def _load_env_file(path: Path) -> None:
    if not path.is_file():
        return
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line and not line.startswith("#") and "=" in line:
            key, _, value = line.partition("=")
            os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))


def main(argv: list[str] | None = None) -> int:
    here = Path(__file__).resolve().parent
    _load_env_file(here / "agent.env")
    parser = argparse.ArgumentParser(description="Агент Джарвиса для компьютера")
    parser.add_argument("--server", default=os.getenv("JARVIS_SERVER", ""), help="https://jarvis.example.com")
    parser.add_argument("--token", default=os.getenv("JARVIS_TOKEN", ""), help="JARVIS_TOKEN с сервера")
    parser.add_argument("--root", default=os.getenv("JARVIS_ROOT", ""), help="Корень доступа (по умолчанию домашняя папка)")
    parser.add_argument("--check", action="store_true", help="Проверить разрешения ОС и выйти")
    parser.add_argument("--no-settings", action="store_true", help="Не открывать окна настроек macOS при проверке")
    parser.add_argument("--install-autostart", action="store_true",
                        help="Запускать агента автоматически при входе в систему")
    parser.add_argument("--uninstall-autostart", action="store_true", help="Убрать автозапуск")
    parser.add_argument("--open-ui", action="store_true", help="После подключения открыть окно Джарвиса в браузере")
    args = parser.parse_args(argv)

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", datefmt="%H:%M:%S")
    actions = Actions(Sandbox(Path(args.root) if args.root else None))

    if args.uninstall_autostart:
        print("Автозапуск:", uninstall_autostart())
        return 0
    if args.check:
        return 0 if check_permissions(actions, open_settings=not args.no_settings) else 1
    if not args.server:
        parser.error("укажите --server (адрес Джарвиса) или JARVIS_SERVER в agent.env")
    if args.install_autostart:
        where = install_autostart(args.server, args.token)
        print(f"✅ Готово: агент Джарвиса будет запускаться сам при входе в систему и уже запущен в фоне.\n"
              f"   {where}\n   Окно терминала можно закрыть. Убрать автозапуск: --uninstall-autostart")
        return 0

    marker = here / ".permissions_checked"
    if not marker.exists() and sys.stdin is not None and sys.stdin.isatty():
        print("Первый запуск: проверяю разрешения ОС…")
        check_permissions(actions, open_settings=True)
        marker.write_text("ok", encoding="utf-8")
        print()

    log.info("Доступ к файлам ограничен папкой %s", actions.sandbox.root)
    try:
        asyncio.run(run_agent(args.server, args.token, actions, open_ui=args.open_ui))
    except KeyboardInterrupt:
        pass
    return 0


if __name__ == "__main__":
    sys.exit(main())