# utils/context_manager.py - 上下文管理器（集成对话持久化和Token统计）

import os
import json
import base64
import mimetypes
import io
import uuid
import platform
import shutil
import subprocess
from copy import deepcopy
from typing import Dict, List, Optional, Any
from pathlib import Path
from datetime import datetime
try:
    from config import (
        MAX_CONTEXT_SIZE,
        DATA_DIR,
        PROMPTS_DIR,
        TERMINAL_SANDBOX_MOUNT_PATH,
        TERMINAL_SANDBOX_CPUS,
        TERMINAL_SANDBOX_MEMORY,
        PROJECT_MAX_STORAGE_MB,
        TERMINAL_SANDBOX_MODE,
        LINUX_SAFETY,
    )
    from config.model_profiles import (
        get_model_prompt_replacements,
        get_registered_model_keys,
        model_supports_image,
        model_supports_video,
    )
except ImportError:
    import sys
    from pathlib import Path
    project_root = Path(__file__).resolve().parents[1]
    if str(project_root) not in sys.path:
        sys.path.insert(0, str(project_root))
    from config import (
        MAX_CONTEXT_SIZE,
        DATA_DIR,
        PROMPTS_DIR,
        TERMINAL_SANDBOX_MOUNT_PATH,
        TERMINAL_SANDBOX_CPUS,
        TERMINAL_SANDBOX_MEMORY,
        PROJECT_MAX_STORAGE_MB,
        TERMINAL_SANDBOX_MODE,
        LINUX_SAFETY,
    )
    from config.model_profiles import (
        get_model_prompt_replacements,
        get_registered_model_keys,
        model_supports_image,
        model_supports_video,
    )
from utils.conversation_manager import ConversationManager
from utils.host_workspace_debug import write_host_workspace_debug
from utils.media_store import MediaStore
from utils.token_usage import normalize_usage_payload

AUTO_SHALLOW_PLACEHOLDER = "过早的工具结果已经被自动压缩"
AUTO_SHALLOW_TOOL_WHITELIST = {
    "write_file",
    "read_file",
    "edit_file",
    "terminal_input",
    "terminal_snapshot",
    "web_search",
    "extract_webpage",
    "run_command",
    "view_image",
    "view_video",
}


class RuntimeMixin:
    """ContextManager runtime mixin 能力 mixin。"""

    def _run_command(self, cmd: List[str], *, timeout: float = 1.5, cwd: Optional[Path] = None) -> str:
        """运行命令并返回标准输出 / Run command and return stdout."""
        try:
            completed = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout,
                cwd=str(cwd) if cwd else None,
            )
        except (OSError, subprocess.TimeoutExpired):
            return ""
        if completed.returncode != 0:
            return ""
        return (completed.stdout or "").strip()

    def _read_first_line(self, path: Path) -> str:
        """读取文件首行并去除空白 / Read first line and strip."""
        try:
            with path.open("r", encoding="utf-8", errors="ignore") as fh:
                return fh.readline().strip()
        except OSError:
            return ""

    def _read_os_release_pretty(self) -> str:
        """读取 Linux 发行版信息 / Read Linux distro from os-release."""
        path = Path("/etc/os-release")
        if not path.exists():
            return ""
        try:
            content = path.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            return ""
        for line in content.splitlines():
            if line.startswith("PRETTY_NAME="):
                value = line.split("=", 1)[1].strip().strip('"')
                return value
        return ""

    def _get_os_description(self) -> str:
        """获取 OS 描述 / Get OS description."""
        system = platform.system()
        if system == "Darwin":
            version = platform.mac_ver()[0] or platform.release()
            return f"macOS {version}".strip()
        if system == "Windows":
            release, version, _csd, _ptype = platform.win32_ver()
            if release and version and version not in release:
                return f"Windows {release} ({version})".strip()
            return f"Windows {release or version or platform.release()}".strip()
        if system == "Linux":
            pretty = self._read_os_release_pretty()
            if pretty:
                return f"Linux {pretty}".strip()
            version = platform.release() or platform.version()
            return f"Linux {version}".strip()
        version = platform.release() or platform.version()
        name = system or "Unknown"
        return f"{name} {version}".strip()

    def _parse_wmic_model(self, output: str) -> str:
        """解析 WMIC 输出 / Parse WMIC output."""
        if not output:
            return ""
        lines = [line.strip() for line in output.splitlines() if line.strip()]
        for line in lines:
            if line.lower() == "model":
                continue
            return line
        return ""

    def _get_device_model(self) -> str:
        """获取设备型号 / Get device model."""
        system = platform.system()
        if system == "Darwin":
            return self._run_command(["sysctl", "-n", "hw.model"])
        if system == "Windows":
            model = self._run_command(
                ["powershell", "-NoProfile", "-Command", "(Get-CimInstance -ClassName Win32_ComputerSystem).Model"]
            )
            if model:
                return model
            output = self._run_command(["wmic", "computersystem", "get", "model"])
            return self._parse_wmic_model(output)
        if system == "Linux":
            product = self._read_first_line(Path("/sys/devices/virtual/dmi/id/product_name"))
            vendor = self._read_first_line(Path("/sys/devices/virtual/dmi/id/sys_vendor"))
            if vendor and product and vendor not in product:
                return f"{vendor} {product}".strip()
            return product or vendor
        return ""

    def _get_python_info(self) -> str:
        """获取 Python 版本与可用命令 / Get Python version and commands."""
        version = platform.python_version()
        commands: List[str] = []
        if shutil.which("python"):
            commands.append("python")
        if shutil.which("python3"):
            commands.append("python3")
        if commands:
            return f"{version} ({', '.join(commands)})"
        return f"{version} (未在PATH)"

    def _get_node_info(self) -> str:
        """获取 Node 版本信息 / Get Node version info."""
        node_cmd = None
        if shutil.which("node"):
            node_cmd = "node"
        elif shutil.which("nodejs"):
            node_cmd = "nodejs"
        if not node_cmd:
            return "nodejs 未安装"
        version = self._run_command([node_cmd, "-v"])
        if version:
            if node_cmd == "nodejs":
                return f"{version} (nodejs)"
            return version
        return f"{node_cmd} 可用"

    def _get_git_info(self) -> str:
        """获取 Git 分支与状态 / Get git branch and status."""
        if not shutil.which("git"):
            return "无git环境"
        cwd = self.project_path if self.project_path.exists() else None
        if not cwd:
            return "未初始化"
        inside = self._run_command(["git", "rev-parse", "--is-inside-work-tree"], cwd=cwd)
        if inside.strip() != "true":
            return "未初始化"
        branch = self._run_command(["git", "rev-parse", "--abbrev-ref", "HEAD"], cwd=cwd)
        if not branch or branch == "HEAD":
            sha = self._run_command(["git", "rev-parse", "--short", "HEAD"], cwd=cwd)
            branch = f"detached@{sha}" if sha else "detached"
        status = self._run_command(["git", "status", "--porcelain"], cwd=cwd)
        dirty = bool(status.strip())
        return f"{branch} ({'dirty' if dirty else 'clean'})"

    def _get_host_runtime_cache(self) -> Dict[str, str]:
        """获取宿主机固定信息 / Get cached host info."""
        if self._host_runtime_cache:
            return self._host_runtime_cache
        os_desc = self._get_os_description() or "unknown"
        arch = platform.machine() or platform.processor() or "unknown"
        model = self._get_device_model() or "unknown"
        python_info = self._get_python_info() or "unknown"
        node_info = self._get_node_info() or "unknown"
        git_info = self._get_git_info() or "unknown"
        self._host_runtime_cache = {
            "os": os_desc,
            "arch": arch,
            "model": model,
            "python": python_info,
            "node": node_info,
            "git": git_info,
        }
        return self._host_runtime_cache

    def _build_host_runtime_environment(self) -> str:
        """构建宿主机运行环境提示 / Build host runtime environment text."""
        base = self._get_host_runtime_cache()
        lines = [
            "宿主机模式",
            f"  OS: {base.get('os', 'unknown')} | Arch: {base.get('arch', 'unknown')} | Model: {base.get('model', 'unknown')}",
            f"  Python: {base.get('python', 'unknown')}",
            f"  Node: {base.get('node', 'unknown')}",
            f"  Git: {base.get('git', 'unknown')}",
        ]
        return "\n".join(lines)
