start.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
跨平台启动入口。

设计目标：
- 与 start.bat 保持一致的核心启动行为
- 为 macOS / Linux 提供可运行的一键入口
- Windows 上也可使用，适合不想依赖批处理脚本的环境
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import socket
import select
import socketserver
import subprocess
import sys
import tempfile
import threading
import time
import urllib.request
import venv
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit


PROJECT_DIR = Path(__file__).resolve().parent
VENV_DIR = PROJECT_DIR / "venv"
REQ_HASH_FILE = VENV_DIR / ".req_hash"
REQUIREMENTS_FILE = PROJECT_DIR / "requirements.txt"
DEFAULT_PIP_MIRROR = "https://pypi.tuna.tsinghua.edu.cn/simple"
DEFAULT_GITHUB_REPO = "lumingya/universal-web-api"
DEFAULT_PYTHON_INSTALL_VERSION = "3.13.6"
_PYTHON_PROXY_SCHEMES = frozenset({
    "http",
    "https",
    "socks4",
    "socks4a",
    "socks5",
    "socks5h",
})

ENV_DEFAULTS = {
    "APP_HOST": "127.0.0.1",
    "APP_PORT": "8199",
    "BROWSER_PORT": "9222",
    "AUTO_UPDATE_ENABLED": "true",
    "GITHUB_REPO": DEFAULT_GITHUB_REPO,
    "PYTHON_INSTALL_VERSION": DEFAULT_PYTHON_INSTALL_VERSION,
    "PROXY_ENABLED": "false",
    "PROXY_ADDRESS": "",
    "PROXY_BYPASS": "localhost,127.0.0.1",
    "PIP_MIRROR_URL": DEFAULT_PIP_MIRROR,
    "BROWSER_PROFILE_DIR": "",
    "BROWSER_PROFILE_NAME": "",
    # Opt-in because background freezing can delay site-specific periodic commands.
    # Set BROWSER_MEMORY_SAVER=true after validating the target sites.
    "BROWSER_MEMORY_SAVER": "false",
    "PROFILE_CLEAN_ENABLED": "false",
    "SCHEDULED_RESTART_ENABLED": "false",
    "SCHEDULED_RESTART_INTERVAL_SECONDS": "10800",
    "SCHEDULED_RESTART_DRAIN_TIMEOUT_SECONDS": "1800",
    "SCHEDULED_RESTART_TAB_STATE_POLICY": "preserve",
    "AUTO_OPEN_BROWSER": "true",
}

REQUIRED_PROJECT_FILES = [
    Path("main.py"),
    Path("app") / "core" / "browser.py",
    Path("app") / "services" / "config_engine.py",
]


def _log(message: str = "") -> None:
    print(message, flush=True)


def _section(title: str) -> None:
    _log(f"[STEP] {title}")
    _log("----------------------------------------")


def _env_flag(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return str(raw).strip().lower() in ("1", "true", "yes", "on")


def _load_env_file(path: Path) -> None:
    if not path.exists():
        if path.name == ".env":
            _log("[WARN] 未找到 .env 文件，使用默认配置")
        return

    _log(f"[INFO] 读取 {path.name} 配置文件...")
    for raw_line in path.read_text(encoding="utf-8-sig").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().strip('"').strip("'")
        if key:
            os.environ[key] = value
    _log("[OK] 配置加载完成")


def _apply_env_defaults() -> None:
    for key, value in ENV_DEFAULTS.items():
        os.environ.setdefault(key, value)


def _normalize_python_proxy_url(value: str) -> str:
    """Convert a browser proxy value into a requests-compatible proxy URL."""
    raw = str(value or "").strip()
    if not raw:
        return ""
    if "://" not in raw:
        raw = f"http://{raw}"

    try:
        parsed = urlsplit(raw)
        scheme = str(parsed.scheme or "").lower()
        if scheme not in _PYTHON_PROXY_SCHEMES or not parsed.hostname:
            return ""
        # Accessing .port validates malformed values such as :not-a-port.
        _ = parsed.port
    except (TypeError, ValueError):
        return ""

    # requests must resolve R2 hostnames through the SOCKS proxy.  `socks5`
    # resolves them locally, which reintroduces the fake-DNS/direct-DNS issue.
    if scheme == "socks5":
        scheme = "socks5h"

    return urlunsplit((scheme, parsed.netloc, parsed.path, parsed.query, ""))


def _merged_no_proxy(*values: str) -> str:
    entries = []
    seen = set()
    for raw_value in values:
        for entry in str(raw_value or "").split(","):
            normalized = entry.strip()
            key = normalized.lower()
            if normalized and key not in seen:
                seen.add(key)
                entries.append(normalized)
    return ",".join(entries)


def _replace_env_value(env: dict[str, str], key: str, value: str) -> None:
    """Replace an environment key irrespective of its original casing."""
    for existing_key in tuple(env):
        if existing_key.lower() == key.lower():
            env.pop(existing_key, None)
    env[key] = value


def _build_service_env() -> dict[str, str]:
    """Build the child service environment without mutating the launcher shell."""
    env = os.environ.copy()
    if not _env_flag("PROXY_ENABLED"):
        return env

    proxy_url = _normalize_python_proxy_url(os.getenv("PROXY_ADDRESS", ""))
    if not proxy_url:
        return env

    _replace_env_value(env, "HTTP_PROXY", proxy_url)
    _replace_env_value(env, "HTTPS_PROXY", proxy_url)

    existing_no_proxy = ",".join(
        str(value or "")
        for key, value in env.items()
        if key.lower() == "no_proxy"
    )
    bypass = _merged_no_proxy(existing_no_proxy, os.getenv("PROXY_BYPASS", ""))
    if bypass:
        _replace_env_value(env, "NO_PROXY", bypass)

    return env


def _display_current_config() -> None:
    _log()
    _log("   当前配置:")
    _log(f"        APP_HOST     : {os.getenv('APP_HOST')}")
    _log(f"        APP_PORT     : {os.getenv('APP_PORT')}")
    _log(f"        BROWSER_PORT : {os.getenv('BROWSER_PORT')}")
    _log(f"        AUTO_UPDATE  : {os.getenv('AUTO_UPDATE_ENABLED')}")
    _log(f"        PYTHON_FIXED : {os.getenv('PYTHON_INSTALL_VERSION')}")
    profile_dir = _resolve_profile_dir()
    _log(f"        PROFILE_DIR  : {profile_dir}")
    profile_name = str(os.getenv("BROWSER_PROFILE_NAME", "") or "").strip()
    if profile_name:
        _log(f"        PROFILE_NAME : {profile_name}")
    _log(f"        PROFILE_CLEAN: {os.getenv('PROFILE_CLEAN_ENABLED')}")
    _log(f"        MEMORY_SAVER : {os.getenv('BROWSER_MEMORY_SAVER', 'false')}")
    if _env_flag("PROXY_ENABLED"):
        _log(f"        PROXY        : {os.getenv('PROXY_ADDRESS', '')} (browser / Python downloads)")
    else:
        _log("        PROXY        : 已禁用")
    _log()


def _venv_python() -> Path:
    if os.name == "nt":
        return VENV_DIR / "Scripts" / "python.exe"
    return VENV_DIR / "bin" / "python"


def _run(
    cmd: list[str],
    *,
    check: bool = True,
    capture: bool = False,
    env: dict | None = None,
) -> subprocess.CompletedProcess:
    kwargs = {
        "cwd": str(PROJECT_DIR),
        "check": check,
        "text": True,
        "env": env,
    }
    if capture:
        kwargs["capture_output"] = True
        kwargs["encoding"] = "utf-8"
        kwargs["errors"] = "replace"
    return subprocess.run(cmd, **kwargs)


def _run_project_python(args: list[str], *, check: bool = True, capture: bool = False) -> subprocess.CompletedProcess:
    return _run(
        [str(_venv_python()), *args],
        check=check,
        capture=capture,
        env=_build_service_env(),
    )


def _python_install_version() -> str:
    return str(os.getenv("PYTHON_INSTALL_VERSION", DEFAULT_PYTHON_INSTALL_VERSION) or DEFAULT_PYTHON_INSTALL_VERSION).strip()


def _python_install_major_minor() -> str:
    parts = _python_install_version().split(".")
    if len(parts) >= 2:
        return f"{parts[0]}.{parts[1]}"
    return "3.13"


def _python_install_short() -> str:
    return _python_install_major_minor().replace(".", "")


def _python_install_url() -> str:
    version = _python_install_version()
    return f"https://www.python.org/ftp/python/{version}/python-{version}-amd64.exe"


def _is_windows_store_python() -> bool:
    if not sys.platform.startswith("win"):
        return False
    return "WindowsApps" in str(Path(sys.executable))


def _python_version_ok(python_path: Path) -> bool:
    try:
        result = subprocess.run(
            [
                str(python_path),
                "-c",
                "import sys; raise SystemExit(0 if sys.version_info >= (3, 8) else 1)",
            ],
            check=False,
            capture_output=True,
            text=True,
            timeout=10,
        )
        return result.returncode == 0
    except Exception:
        return False


def _find_installed_fixed_python() -> Path | None:
    if not sys.platform.startswith("win"):
        return None

    short_version = _python_install_short()
    major_minor = _python_install_major_minor()
    local_app_data = os.getenv("LOCALAPPDATA", "")
    program_files = os.getenv("ProgramFiles", r"C:\Program Files")
    program_files_x86 = os.getenv("ProgramFiles(x86)", r"C:\Program Files (x86)")
    candidates = [
        Path(local_app_data) / "Programs" / "Python" / f"Python{short_version}" / "python.exe",
        Path(program_files) / f"Python{short_version}" / "python.exe",
        Path(program_files_x86) / f"Python{short_version}" / "python.exe",
    ]

    for candidate in candidates:
        if candidate.exists() and _python_version_ok(candidate):
            return candidate

    py_launcher = shutil.which("py")
    if py_launcher:
        try:
            result = subprocess.run(
                [py_launcher, f"-{major_minor}", "-c", "import sys; print(sys.executable)"],
                check=False,
                capture_output=True,
                text=True,
                timeout=10,
            )
            if result.returncode == 0:
                candidate = Path(result.stdout.strip())
                if candidate.exists() and _python_version_ok(candidate):
                    return candidate
        except Exception:
            pass

    return None


def _download_and_install_fixed_python() -> Path:
    version = _python_install_version()
    url = _python_install_url()
    installer = Path(tempfile.gettempdir()) / f"python-{version}-amd64.exe"

    _log()
    _log(f"[INFO] 正在下载 Python {version}...")
    _log(f"[INFO] 下载来源: {url}")
    try:
        urllib.request.urlretrieve(url, installer)
    except Exception as exc:
        raise RuntimeError(f"Python 安装包下载失败: {exc}") from exc

    if not installer.exists():
        raise RuntimeError(f"Python 安装包不存在: {installer}")

    _log(f"[INFO] 正在静默安装 Python {version}...")
    result = subprocess.run(
        [
            str(installer),
            "/quiet",
            "InstallAllUsers=0",
            "PrependPath=1",
            "Include_launcher=1",
            "Include_pip=1",
            "Include_test=0",
            "SimpleInstall=1",
        ],
        check=False,
    )
    try:
        installer.unlink(missing_ok=True)
    except Exception:
        pass

    if result.returncode not in (0, 3010):
        raise RuntimeError(f"Python 安装失败，安装器退出码: {result.returncode}")

    installed = _find_installed_fixed_python()
    if not installed:
        raise RuntimeError("Python 安装完成后仍未找到可用解释器，请重新打开终端后再运行 start.py")
    return installed


def _offer_python_install(reason: str) -> bool:
    if not sys.platform.startswith("win"):
        return False

    version = _python_install_version()
    _log()
    _log(f"[INFO] {reason}")
    _log()
    _log(f"   可自动下载安装固定版本 Python {version} (64-bit)")
    _log(f"   下载来源: {_python_install_url()}")
    _log("   安装范围: 当前用户")
    _log()
    choice = input(f"是否自动下载并安装 Python {version}？(Y/N): ").strip()
    if choice.lower() != "y":
        _log("[INFO] 已取消自动安装 Python")
        return False

    installed = _download_and_install_fixed_python()
    if Path(sys.executable).resolve() == installed.resolve():
        return True

    _log(f"[OK] Python 已就绪: {installed}")
    _log("[INFO] 正在使用新 Python 重新运行 start.py...")
    os.execv(str(installed), [str(installed), str(Path(__file__).resolve()), *sys.argv[1:]])
    return True


def _check_python_version() -> None:
    _section("检查 Python 环境")
    version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
    if _is_windows_store_python():
        if _offer_python_install("检测到 Windows Store Python 占位符"):
            return
        raise RuntimeError("检测到 Windows Store Python 占位符，请关闭应用执行别名或安装完整版 Python")
    if sys.version_info < (3, 8):
        if _offer_python_install(f"Python 版本过低: {version}，最低要求 Python 3.8+"):
            return
        raise RuntimeError(f"Python 版本过低: {version}，最低要求 Python 3.8+")
    _log(f"[OK] Python {version}")
    _log(f"    路径: {sys.executable}")
    _log()


def _run_auto_update() -> None:
    if not _env_flag("AUTO_UPDATE_ENABLED", True):
        _log("[INFO] 自动更新已禁用")
        _log("       本次不会自动应用更新；服务启动后仍会检查新版本并在设置图标提示")
        _log("       如需自动应用更新，请修改 .env 中的 AUTO_UPDATE_ENABLED=true")
        _log()
        return

    _section("自动更新")
    updater_script = PROJECT_DIR / "updater.py"
    if not updater_script.exists():
        _log("[WARN] 未找到 updater.py，跳过自动更新")
        _log()
        return

    _log("[INFO] 检查 GitHub 最新版本...")
    result = _run([sys.executable, "updater.py"], check=False)
    if result.returncode == 0:
        _log("[INFO] 自动更新已应用，继续启动服务...")
    else:
        _log("[WARN] 本次未应用更新，继续启动服务")
    _log()


def _ensure_project_structure() -> None:
    _section("检查项目结构")
    missing = []
    for rel_path in REQUIRED_PROJECT_FILES:
        if not (PROJECT_DIR / rel_path).exists():
            missing.append(str(rel_path))

    sites_config = PROJECT_DIR / "config" / "sites.json"
    if not sites_config.exists():
        _log("[WARN] 缺失: config/sites.json，将自动创建")
        sites_config.parent.mkdir(parents=True, exist_ok=True)
        sites_config.write_text('{"_global": {"selector_definitions": []}}\n', encoding="utf-8")
        _log("[INFO] 已创建空配置文件")
    else:
        _log("[OK] 找到: config/sites.json")

    if missing:
        for item in missing:
            _log(f"[ERROR] 缺失: {item}")
        raise RuntimeError("项目结构不完整，请检查文件是否齐全")

    _log("[OK] 项目结构检查通过")
    _log()


def _ensure_venv() -> None:
    _section("准备虚拟环境")
    python_path = _venv_python()
    if python_path.exists():
        _log("[OK] 虚拟环境已存在")
        _log()
        return

    if VENV_DIR.exists():
        raise RuntimeError("虚拟环境损坏，缺少 Python 解释器。请删除 venv 后重新运行。")

    _log("[INFO] 创建虚拟环境...")
    builder = venv.EnvBuilder(with_pip=True)
    builder.create(str(VENV_DIR))
    if not python_path.exists():
        raise RuntimeError("虚拟环境创建失败，缺少 Python 解释器")
    _log("[OK] 虚拟环境创建成功")
    _log()


def _file_md5(path: Path) -> str:
    digest = hashlib.md5()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def _dependencies_ok() -> bool:
    check_script = PROJECT_DIR / "check_deps.py"
    if not check_script.exists():
        return True
    try:
        result = _run_project_python([check_script.name], check=False, capture=True)
    except Exception:
        return False
    return result.returncode == 0


def _ensure_dependencies() -> None:
    _section("检查依赖")
    if not REQUIREMENTS_FILE.exists():
        raise FileNotFoundError("缺少 requirements.txt 文件")

    current_hash = _file_md5(REQUIREMENTS_FILE)
    old_hash = ""
    if REQ_HASH_FILE.exists():
        try:
            old_hash = REQ_HASH_FILE.read_text(encoding="utf-8").strip()
        except Exception:
            old_hash = ""

    if old_hash == current_hash and _dependencies_ok():
        _log("[OK] 依赖已是最新")
        _log()
        return

    _log("[INFO] 安装 Python 依赖包...")
    pip_cmd = [str(_venv_python()), "-m", "pip", "install", "-r", str(REQUIREMENTS_FILE)]
    install_source = "PyPI"
    result = _run(pip_cmd, check=False)
    if result.returncode != 0:
        mirror = os.getenv("PIP_MIRROR_URL", DEFAULT_PIP_MIRROR).strip() or DEFAULT_PIP_MIRROR
        install_source = mirror
        _log(f"[WARN] 默认 PyPI 安装失败，尝试镜像: {mirror}")
        result = _run(pip_cmd + ["-i", mirror], check=False)
        if result.returncode != 0:
            if REQ_HASH_FILE.exists():
                REQ_HASH_FILE.unlink()
            raise RuntimeError("依赖安装失败")

    pip_check = _run([str(_venv_python()), "-m", "pip", "check"], check=False, capture=True)
    if pip_check.returncode != 0:
        _log("[WARN] pip check 报告依赖冲突，但不影响继续启动")

    REQ_HASH_FILE.parent.mkdir(parents=True, exist_ok=True)
    REQ_HASH_FILE.write_text(current_hash, encoding="utf-8")
    _log("[OK] 依赖安装完成")
    _log(f"[INFO] 安装来源: {install_source}")
    _log()


def _maybe_apply_patch() -> None:
    _section("应用 DrissionPage 补丁")
    patch_script = PROJECT_DIR / "patch_drissionpage.py"
    if not patch_script.exists():
        _log("[WARN] 未找到 patch_drissionpage.py，跳过补丁")
        _log()
        return
    result = _run_project_python([patch_script.name], check=False)
    if result.returncode != 0:
        _log("[WARN] 补丁应用失败，网络监听模式可能触发 CF 检测")
        _log("       项目仍可正常运行（DOM 模式不受影响）")
    _log()


def _resolve_profile_dir() -> Path:
    raw = str(os.getenv("BROWSER_PROFILE_DIR", "") or "").strip()
    if raw:
        profile_dir = Path(raw).expanduser()
        if not profile_dir.is_absolute():
            profile_dir = PROJECT_DIR / profile_dir
        return profile_dir
    return PROJECT_DIR / "chrome_profile"


def _maybe_clean_profile(profile_dir: Path) -> None:
    _section("浏览器配置瘦身")
    if not _env_flag("PROFILE_CLEAN_ENABLED"):
        _log(f"[INFO] 已禁用配置瘦身（PROFILE_CLEAN_ENABLED={os.getenv('PROFILE_CLEAN_ENABLED')}）")
        _log()
        return

    clean_script = PROJECT_DIR / "clean_profile.py"
    if not clean_script.exists():
        _log("[WARN] 未找到 clean_profile.py，跳过清理")
        _log()
        return

    _log("[INFO] 执行浏览器配置瘦身...")
    _run_project_python([clean_script.name, str(profile_dir)], check=False)
    _log()


def _debug_port_ready(port: int) -> bool:
    try:
        with socket.create_connection(("127.0.0.1", int(port)), timeout=0.4):
            return True
    except Exception:
        return False


def _check_chromium_debug_endpoint(port: int, timeout: float = 0.5) -> tuple[bool, dict[str, Any]]:
    """
    检查指定端口是否为可响应且合法的 Chromium 远程调试端点。
    返回 (is_chromium, version_info)。
    """
    version_url = f"http://127.0.0.1:{int(port)}/json/version"
    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
    try:
        req = urllib.request.Request(version_url, headers={"User-Agent": "UWAPI-Diagnostics"})
        with opener.open(req, timeout=timeout) as resp:
            data = json.loads(resp.read().decode("utf-8", errors="ignore"))
            if isinstance(data, dict) and ("Browser" in data or "webSocketDebuggerUrl" in data):
                return True, data
    except Exception:
        pass
    return False, {}


def _get_port_processes(port: int) -> list[dict[str, Any]]:
    """获取占用指定 TCP 端口（处于 LISTENING 状态）的进程信息列表。"""
    port_int = int(port)
    pids: set[int] = set()

    # 1. 在 Windows 上优先使用 netstat -ano（支持 IPv4 与 IPv6 Dual-stack 监听）
    if sys.platform == "win32":
        try:
            output = subprocess.check_output(
                ["netstat", "-ano"],
                text=True,
                errors="ignore",
                stderr=subprocess.DEVNULL,
            )
            pattern = re.compile(
                rf"^\s*TCP\s+\[?[\w\.:]+\]?:({port_int})\s+\S+\s+LISTENING\s+(\d+)",
                re.IGNORECASE | re.MULTILINE,
            )
            for match in pattern.finditer(output):
                try:
                    pids.add(int(match.group(2)))
                except (ValueError, TypeError):
                    pass
        except Exception:
            pass

    # 2. 尝试使用 psutil 获取连接
    if not pids:
        try:
            import psutil
            for conn in psutil.net_connections(kind="tcp"):
                if conn.status == psutil.CONN_LISTEN and conn.laddr and conn.laddr.port == port_int:
                    if conn.pid:
                        pids.add(conn.pid)
        except Exception:
            pass

    # 3. 获取进程详情
    results = []
    for pid in sorted(pids):
        if pid <= 0:
            continue
        info = {"pid": pid, "name": f"PID:{pid}", "cmdline": ""}
        try:
            import psutil
            proc = psutil.Process(pid)
            info["name"] = proc.name()
            try:
                cmd = proc.cmdline()
                info["cmdline"] = " ".join(cmd) if cmd else ""
            except Exception:
                pass
        except Exception:
            pass
        results.append(info)
    return results


def _terminate_port_processes(port: int, procs: list[dict[str, Any]]) -> bool:
    """终止占用指定端口的进程及其子进程树，并等待端口完全释放。"""
    pids = [p["pid"] for p in procs if p.get("pid", 0) > 0]
    if not pids:
        return not _debug_port_ready(port)

    for pid in pids:
        if sys.platform == "win32":
            try:
                subprocess.run(
                    ["taskkill", "/F", "/T", "/PID", str(pid)],
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    check=False,
                )
            except Exception:
                pass
        else:
            try:
                import signal
                os.kill(pid, getattr(signal, "SIGKILL", 9))
            except Exception:
                pass
        try:
            import psutil
            proc = psutil.Process(pid)
            for child in proc.children(recursive=True):
                try:
                    child.kill()
                except Exception:
                    pass
            proc.kill()
        except Exception:
            pass

    # 轮询等待端口释放（最多 5 秒）
    for _ in range(10):
        time.sleep(0.5)
        if not _debug_port_ready(port):
            return True
    return False


def _is_interactive() -> bool:
    """安全检查当前环境是否为可交互终端（防御 sys.stdin 为 None）。"""
    try:
        return bool(sys.stdin and hasattr(sys.stdin, "isatty") and sys.stdin.isatty())
    except Exception:
        return False


def _ask_user_yes_no(prompt: str, default: bool = False) -> bool:
    """在交互式终端中询问用户 (y/n)，返回 True 表示确认终止。"""
    if not _is_interactive():
        return default
    try:
        reply = input(prompt).strip().lower()
        if not reply:
            return default
        return reply in ("y", "yes")
    except EOFError:
        _log()
        return default
    except KeyboardInterrupt:
        _log()
        raise


def _inspect_headless_or_windowless_browser(port: int, timeout: float = 0.5) -> Optional[dict[str, str]]:
    """
    探测指定端口的 Chromium 浏览器是否处于无头(Headless)或无窗口/缺少页面标签页状态。
    返回 None 表示不是无头（或者端口未开放/非 Chromium）。
    """
    import json
    import urllib.request

    port_int = int(port)
    # 显式使用空 ProxyHandler，避免系统或环境变量代理(如 HTTP_PROXY)拦截 loopback 探测
    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))

    # 1. 检查进程启动参数 (若 psutil 可用，优先确认是否明确声明了 --headless 或 --no-startup-window)
    cmd_has_headless = False
    cmd_has_no_startup_window = False
    try:
        import psutil
        target_arg = f"--remote-debugging-port={port_int}"
        for proc in psutil.process_iter(["pid", "cmdline"]):
            try:
                cmdline_list = proc.info.get("cmdline") or []
                # 过滤 Chromium 子进程（renderer/gpu/utility 等子进程必然包含 --type=，主进程不包含）
                if any(str(arg).startswith("--type=") for arg in cmdline_list):
                    continue
                cmdline = " ".join(cmdline_list)
                if target_arg in cmdline:
                    cmd_lower = cmdline.lower()
                    if "--headless" in cmd_lower:
                        cmd_has_headless = True
                    if "--no-startup-window" in cmd_lower:
                        cmd_has_no_startup_window = True
                    if cmd_has_headless and cmd_has_no_startup_window:
                        break
            except Exception:
                continue
    except Exception:
        pass

    if cmd_has_headless:
        return {"reason": "headless_arg", "detail": "浏览器启动参数显式包含 --headless"}

    # 2. 检查 /json/version
    version_url = f"http://127.0.0.1:{port_int}/json/version"
    try:
        req = urllib.request.Request(version_url, headers={"User-Agent": "UWAPI-Diagnostics"})
        with opener.open(req, timeout=timeout) as resp:
            data = json.loads(resp.read().decode("utf-8", errors="ignore"))
    except Exception:
        return None

    browser_desc = str(data.get("Browser", "") or "")
    user_agent = str(data.get("User-Agent", "") or "")
    if "headless" in browser_desc.lower() or "headless" in user_agent.lower():
        return {
            "reason": "headless",
            "detail": f"检测到无头浏览器 ({browser_desc or user_agent})",
        }

    # 3. 检查 /json 目标列表 (targets)
    targets_url = f"http://127.0.0.1:{port_int}/json"
    try:
        req = urllib.request.Request(targets_url, headers={"User-Agent": "UWAPI-Diagnostics"})
        with opener.open(req, timeout=timeout) as resp:
            targets = json.loads(resp.read().decode("utf-8", errors="ignore"))
            if isinstance(targets, list):
                pages = [
                    t for t in targets
                    if isinstance(t, dict) and str(t.get("type", "")).lower() == "page"
                ]
                if not pages:
                    if cmd_has_no_startup_window:
                        return {
                            "reason": "no_startup_window",
                            "detail": "浏览器带有 --no-startup-window 参数且当前无前台可视页面",
                        }
                    other_types = sorted(list({
                        str(t.get("type", "unknown"))
                        for t in targets if isinstance(t, dict) and t.get("type")
                    }))
                    detail_types = ", ".join(other_types) if other_types else "无任何活动目标"
                    return {
                        "reason": "no_pages",
                        "detail": f"无可视标签页（当前后台目标: {detail_types}）",
                    }
    except Exception:
        pass

    return None


class _RestartHandoffProxy(socketserver.ThreadingTCPServer):
    """Small TCP relay which buffers a request while the child service restarts."""

    allow_reuse_address = True
    daemon_threads = True

    def __init__(self, host: str, port: int, backend_port: int):
        self.backend_port = int(backend_port)
        super().__init__((host, int(port)), _RestartHandoffProxyHandler)


class _RestartHandoffProxyHandler(socketserver.BaseRequestHandler):
    _BUFFER_SIZE = 64 * 1024
    _MAX_REQUEST_BYTES = 64 * 1024 * 1024

    def handle(self) -> None:
        try:
            payload = self._read_request()
            if payload:
                self._forward_when_backend_ready(payload)
        except Exception:
            return

    def _read_request(self) -> bytes:
        self.request.settimeout(60.0)
        payload = bytearray()
        while b"\r\n\r\n" not in payload:
            chunk = self.request.recv(self._BUFFER_SIZE)
            if not chunk:
                return b""
            payload.extend(chunk)
            if len(payload) > self._MAX_REQUEST_BYTES:
                return b""

        header_end = payload.find(b"\r\n\r\n") + 4
        headers = bytes(payload[:header_end]).decode("iso-8859-1", errors="replace")
        content_length = 0
        for line in headers.split("\r\n")[1:]:
            if line.lower().startswith("content-length:"):
                content_length = max(0, int(line.split(":", 1)[1].strip()))
                break
        target_size = header_end + content_length
        if target_size > self._MAX_REQUEST_BYTES:
            return b""
        while len(payload) < target_size:
            chunk = self.request.recv(min(self._BUFFER_SIZE, target_size - len(payload)))
            if not chunk:
                return b""
            payload.extend(chunk)
        # One request per client connection makes the restart boundary explicit
        # and lets us safely replay a request that arrived while draining.
        header_lines = [
            line for line in headers.split("\r\n")
            if not line.lower().startswith("connection:")
        ]
        rewritten_headers = "\r\n".join(header_lines).rstrip("\r\n") + "\r\nConnection: close\r\n\r\n"
        return rewritten_headers.encode("iso-8859-1") + bytes(payload[header_end:])

    def _client_disconnected(self) -> bool:
        # One HTTP request per connection. Never leave an old backend running
        # merely because it has not produced its first response byte yet.
        try:
            readable, _, _ = select.select([self.request], [], [], 0)
            if not readable:
                return False
            return not self.request.recv(1, socket.MSG_PEEK)
        except OSError:
            return True

    def _forward_when_backend_ready(self, payload: bytes) -> None:
        deadline = time.monotonic() + 120.0
        while time.monotonic() < deadline:
            if self._client_disconnected():
                return
            try:
                backend = socket.create_connection(("127.0.0.1", self.server.backend_port), timeout=1.0)
            except OSError:
                # Connection not established: no request could have executed.
                time.sleep(0.15)
                continue
            with backend:
                try:
                    backend.settimeout(300.0)
                    if self._client_disconnected():
                        return
                    backend.sendall(payload)
                    if self._relay_response(backend):
                        return
                except OSError:
                    # A partial send, backend reset, or CLIENT write error is
                    # ambiguous. Never replay a possibly executed POST.
                    return
            # Only an explicit pre-admission handoff response permits replay.
            time.sleep(0.15)

    def _relay_response(self, backend: socket.socket) -> bool:
        header = bytearray()
        headers_sent = False
        deadline = time.monotonic() + 300.0
        while time.monotonic() < deadline:
            if self._client_disconnected():
                return True
            readable, _, _ = select.select([backend], [], [], 0.1)
            if not readable:
                continue
            chunk = backend.recv(self._BUFFER_SIZE)
            if not chunk:
                return True  # EOF after dispatch is NOT permission to retry.
            deadline = time.monotonic() + 300.0
            if not headers_sent:
                header.extend(chunk)
                boundary = header.find(b"\r\n\r\n")
                if boundary < 0:
                    if len(header) > self._BUFFER_SIZE:
                        return True
                    continue
                lines = bytes(header[:boundary]).split(b"\r\n")
                status = lines[0].split()
                handoff = any(line.lower().strip() == b"x-uwa-restart-handoff: pending" for line in lines[1:])
                if len(status) >= 2 and status[1] == b"503" and handoff:
                    return False
                self.request.sendall(header)
                headers_sent = True
            else:
                self.request.sendall(chunk)
        return True


def _start_restart_handoff_proxy(host: str, public_port: int, backend_port: int):
    try:
        proxy = _RestartHandoffProxy(host, public_port, backend_port)
    except OSError as exc:
        raise RuntimeError(f"无法启动重启守护代理（端口 {public_port}）：{exc}") from exc
    threading.Thread(target=proxy.serve_forever, daemon=True, name="restart-handoff-proxy").start()
    _log(f"[OK] 重启守护代理已监听 {host}:{public_port}（后端内部端口 {backend_port}）")
    return proxy


def _focus_browser_window_for_port(port: int) -> bool:
    """Best-effort Windows foreground restore for an already-running browser."""
    if not sys.platform.startswith("win"):
        return False

    try:
        import ctypes
        from ctypes import wintypes
        import psutil
    except Exception as e:
        _log(f"[DEBUG] 跳过浏览器唤起：{e}")
        return False

    target_arg = f"--remote-debugging-port={int(port)}"
    target_pid = 0
    for proc in psutil.process_iter(["pid", "cmdline"]):
        try:
            cmdline = " ".join(proc.info.get("cmdline") or [])
            if target_arg in cmdline:
                target_pid = int(proc.info["pid"])
                break
        except Exception:
            continue

    if not target_pid:
        return False

    user32 = ctypes.windll.user32
    SW_RESTORE = 9
    hwnd_box = {"value": 0}
    enum_proc = ctypes.WINFUNCTYPE(ctypes.c_bool, wintypes.HWND, wintypes.LPARAM)

    @enum_proc
    def _enum_windows(hwnd, _lparam):
        pid = wintypes.DWORD()
        user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
        if pid.value != target_pid:
            return True

        title_len = user32.GetWindowTextLengthW(hwnd)
        if title_len > 0:
            hwnd_box["value"] = int(hwnd)
            if user32.IsWindowVisible(hwnd):
                return False
        elif hwnd_box["value"] == 0:
            hwnd_box["value"] = int(hwnd)
        return True

    try:
        user32.EnumWindows(_enum_windows, 0)
        hwnd = int(hwnd_box["value"] or 0)
        if not hwnd:
            return False
        user32.ShowWindowAsync(hwnd, SW_RESTORE)
        time.sleep(0.15)
        user32.SetForegroundWindow(hwnd)
        return True
    except Exception as e:
        _log(f"[DEBUG] 唤起浏览器窗口失败: {e}")
        return False


def _windows_browser_candidates() -> list[str]:
    local_app_data = os.getenv("LOCALAPPDATA", "")
    program_files = os.getenv("ProgramFiles", r"C:\Program Files")
    program_files_x86 = os.getenv("ProgramFiles(x86)", r"C:\Program Files (x86)")
    candidates = [
        Path(program_files) / "Google" / "Chrome" / "Application" / "chrome.exe",
        Path(program_files_x86) / "Google" / "Chrome" / "Application" / "chrome.exe",
        Path(local_app_data) / "Google" / "Chrome" / "Application" / "chrome.exe",
        Path(program_files_x86) / "Microsoft" / "Edge" / "Application" / "msedge.exe",
        Path(program_files) / "Microsoft" / "Edge" / "Application" / "msedge.exe",
        Path(local_app_data) / "BraveSoftware" / "Brave-Browser" / "Application" / "brave.exe",
        Path(program_files) / "BraveSoftware" / "Brave-Browser" / "Application" / "brave.exe",
        Path(local_app_data) / "Vivaldi" / "Application" / "vivaldi.exe",
        Path(program_files) / "Vivaldi" / "Application" / "vivaldi.exe",
        Path(local_app_data) / "Programs" / "Opera" / "opera.exe",
        Path(program_files) / "Opera" / "opera.exe",
    ]
    return [str(path) for path in candidates]


def _platform_browser_candidates() -> list[str]:
    custom = str(os.getenv("BROWSER_PATH", "") or "").strip()
    if custom:
        return [custom]

    if sys.platform.startswith("win"):
        return _windows_browser_candidates()

    if sys.platform == "darwin":
        return [
            "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
            "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
            "/Applications/Brave Browser.app/Contents/MacOS/Brave Browser",
            "/Applications/Vivaldi.app/Contents/MacOS/Vivaldi",
            "/Applications/Opera.app/Contents/MacOS/Opera",
        ]

    if sys.platform.startswith("linux"):
        names = [
            "google-chrome",
            "google-chrome-stable",
            "chromium",
            "chromium-browser",
            "microsoft-edge",
            "brave-browser",
            "vivaldi",
            "opera",
        ]
        resolved = []
        for name in names:
            path = shutil.which(name)
            if path:
                resolved.append(path)
        return resolved

    return []


def _resolve_browser_path() -> str:
    custom = str(os.getenv("BROWSER_PATH", "") or "").strip()
    for candidate in _platform_browser_candidates():
        if candidate and os.path.exists(candidate):
            return candidate
    if custom:
        _log(f"[WARN] BROWSER_PATH 指定的路径不存在: {custom}")
    return ""


def _launch_browser_if_needed() -> None:
    _section("准备 Chromium 内核浏览器")
    browser_port = int(os.getenv("BROWSER_PORT", "9222") or "9222")
    profile_dir = _resolve_profile_dir()
    profile_dir.mkdir(parents=True, exist_ok=True)
    _log(f"[INFO] 浏览器配置目录: {profile_dir}")

    while _debug_port_ready(browser_port):
        port_procs = _get_port_processes(browser_port)
        proc_desc = ", ".join(f"{p['name']} (PID: {p['pid']})" for p in port_procs) if port_procs else "未知进程"

        is_chromium, _ = _check_chromium_debug_endpoint(browser_port)
        headless_probe = None
        if is_chromium:
            for attempt in range(3):
                headless_probe = _inspect_headless_or_windowless_browser(browser_port)
                if not headless_probe:
                    break
                # 明确的 --headless 或无头标识无需重试
                if headless_probe.get("reason") in {"headless", "headless_arg"}:
                    break
                # 若仅是暂时未检测到 page 目标，等待 0.4s 重试（给新打开的浏览器创建主页留出缓冲）
                if attempt < 2:
                    time.sleep(0.4)

        # 场景 1：端口被占用，但不是正常的受控前台浏览器（非 Chromium 调试服务，或处于无头/无窗口状态）
        if not is_chromium or headless_probe:
            if not is_chromium:
                reason = "该端口被其他非 Chromium 调试服务占用"
            else:
                reason = f"该浏览器处于无头或无窗口状态（{headless_probe['detail']}）"

            _log(f"[WARN] 检测到浏览器端口 {browser_port} 已被占用，但无法作为受控前台浏览器使用！")
            _log(f"[WARN] 状态原因: {reason}")
            _log(f"[WARN] 占用进程: {proc_desc}")

            if not _is_interactive():
                raise RuntimeError(
                    f"检测到端口 {browser_port} 被占用（{proc_desc}，{reason}）。"
                    "当前处于非交互环境，无法询问用户，请手动结束占用进程后重试。"
                )

            should_kill = _ask_user_yes_no(
                f"是否终止占用端口 {browser_port} 的进程？(y/N): ",
                default=False,
            )
            if should_kill:
                _log(f"[INFO] 正在终止占用端口 {browser_port} 的进程 ({proc_desc})...")
                if _terminate_port_processes(browser_port, port_procs):
                    _log(f"[OK] 进程已终止，端口 {browser_port} 已释放。准备重新启动受控浏览器...")
                    _log()
                    break
                else:
                    _log(f"[ERROR] 终止进程后，端口 {browser_port} 仍被占用，请检查权限或手动在任务管理器中结束。")
                    raise RuntimeError(f"端口 {browser_port} 无法释放，本次启动中止。")
            else:
                _log("[INFO] 已取消操作，本次启动中止。")
                raise SystemExit(1)

        # 场景 2：已有正常的前台可视 Chromium 实例
        _log(f"[WARN] Debug 端口已被占用（进程: {proc_desc}）")
        if _is_interactive():
            should_kill = _ask_user_yes_no(
                f"检测到已有运行中的浏览器，是否终止它并全新启动？(y/N, 默认复用): ",
                default=False,
            )
            if should_kill:
                _log(f"[INFO] 正在终止现有浏览器进程 ({proc_desc})...")
                if _terminate_port_processes(browser_port, port_procs):
                    _log(f"[OK] 现有浏览器已关闭，端口 {browser_port} 已释放。准备全新启动浏览器...")
                    _log()
                    break
                else:
                    _log(f"[ERROR] 关闭现有浏览器失败，端口 {browser_port} 仍被占用。")
                    raise RuntimeError(f"端口 {browser_port} 无法释放，本次启动中止。")

        _log("[WARN] 将复用现有浏览器实例")
        _log("[WARN] 浏览器启动参数只在新进程生效；如需应用内存节省模式，请先关闭现有浏览器")
        if _focus_browser_window_for_port(browser_port):
            _log("[INFO] 已唤起现有浏览器窗口")
        _log(f"[OK] Debug 端口就绪: {browser_port}")
        _log()
        return

    browser_path = _resolve_browser_path()
    if not browser_path:
        raise RuntimeError("找不到可用的 Chromium 内核浏览器。请安装 Chrome/Edge/Brave/Vivaldi/Opera，或设置 BROWSER_PATH。")

    browser_args = [
        browser_path,
        f"--remote-debugging-port={browser_port}",
        f"--user-data-dir={profile_dir}",
        "--no-first-run",
        "--no-default-browser-check",
        "about:blank",
    ]

    # The previous launcher explicitly disabled Chromium's background memory
    # controls. Keep that behavior as the default for compatibility, while
    # allowing a validated opt-in memory saver mode.
    if _env_flag("BROWSER_MEMORY_SAVER", False):
        browser_args.insert(
            -1,
            "--enable-features=CalculateNativeWinOcclusion,AutomaticTabDiscarding,TabFreeze,IntensiveWakeUpThrottling",
        )
        _log("[INFO] Chromium 内存节省模式已启用（后台标签页允许降频/冻结/回收）")
    else:
        browser_args[4:4] = [
            "--disable-backgrounding-occluded-windows",
            "--disable-background-timer-throttling",
            "--disable-renderer-backgrounding",
            "--disable-features=CalculateNativeWinOcclusion,AutomaticTabDiscarding,TabFreeze,IntensiveWakeUpThrottling",
        ]
        _log("[INFO] Chromium 常驻后台模式已启用（BROWSER_MEMORY_SAVER=false）")

    profile_name = str(os.getenv("BROWSER_PROFILE_NAME", "") or "").strip()
    if profile_name:
        browser_args.insert(-1, f"--profile-directory={profile_name}")

    if _env_flag("PROXY_ENABLED"):
        proxy_address = str(os.getenv("PROXY_ADDRESS", "") or "").strip()
        proxy_bypass = str(os.getenv("PROXY_BYPASS", "") or "").strip()
        if proxy_address:
            browser_args.insert(-1, f"--proxy-server={proxy_address}")
            if proxy_bypass:
                browser_args.insert(-1, f"--proxy-bypass-list={proxy_bypass}")
            _log(f"[INFO] 代理已启用: {proxy_address}")

    _log(f"[INFO] 启动浏览器: {browser_path}")
    subprocess.Popen(
        browser_args,
        cwd=str(PROJECT_DIR),
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )

    _log("[INFO] 等待浏览器远程调试端口就绪...")
    for _ in range(15):
        if _debug_port_ready(browser_port):
            _log(f"[OK] 浏览器启动成功 - 端口 {browser_port}")
            if _focus_browser_window_for_port(browser_port):
                _log("[INFO] 已唤起新启动的浏览器窗口")
            _log()
            return
        time.sleep(1.0)

    port_procs = _get_port_processes(browser_port)
    proc_hint = f"（当前该端口被以下进程占用: {', '.join(f'{p['name']}(PID:{p['pid']})' for p in port_procs)}）" if port_procs else ""
    raise RuntimeError(
        f"未检测到远程调试端口 {browser_port}{proc_hint}，为避免服务误连到错误浏览器，本次启动已中止。"
    )


def _display_version_info() -> None:
    version_file = PROJECT_DIR / "VERSION"
    if not version_file.exists():
        return
    _log("   版本信息:")
    _log("   ----------------------------------------")
    _log(version_file.read_text(encoding="utf-8").strip())
    _log("   ----------------------------------------")
    _log()


def _display_start_summary() -> None:
    host = os.getenv("APP_HOST", "127.0.0.1")
    port = os.getenv("APP_PORT", "8199")
    _log("========================================")
    _log("   服务启动中...")
    _log("========================================")
    _log()
    _log(f"   API 地址:     http://{host}:{port}")
    _log(f"   控制面板:     http://{host}:{port}/")
    _log(f"   API 文档:     http://{host}:{port}/docs")
    _log()
    _log("   项目结构:")
    _log(f"        配置目录:  {PROJECT_DIR / 'config'}")
    _log(f"        静态资源:  {PROJECT_DIR / 'static'}")
    _log()
    if _env_flag("AUTO_UPDATE_ENABLED", True):
        _log("   [WARN] 自动更新: 已启用")
    else:
        _log("   自动更新: 已禁用（仍会启动后检查新版本）")
    _log()
    _log("   按 Ctrl+C 停止服务")
    _log("   配置修改后会自动重启")
    _log("========================================")
    _log()


def _find_backend_port(public_port: int) -> int:
    preferred = int(public_port) + 1
    candidates = [preferred] if preferred <= 65535 else []
    for candidate in candidates:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
                probe.bind(("127.0.0.1", candidate))
            return candidate
        except OSError:
            pass
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
        probe.bind(("127.0.0.1", 0))
        return int(probe.getsockname()[1])


def _run_service_loop(*, public_port: int | None = None, backend_port: int | None = None) -> int:
    is_restart = False
    while True:
        _load_env_file(PROJECT_DIR / ".env")
        child_env = _build_service_env()
        if is_restart:
            child_env["UWAPI_IS_RESTART"] = "1"
        if backend_port is not None:
            child_env["APP_HOST"] = "127.0.0.1"
            child_env["APP_PORT"] = str(backend_port)
            # main.py loads .env during import.  Keep the launcher-selected
            # private port authoritative over the public APP_PORT value.
            child_env["UWAPI_DOTENV_OVERRIDE"] = "0"
        completed = _run(
            [str(_venv_python()), "main.py"],
            check=False,
            env=child_env,
        )
        if completed.returncode == 0:
            _log()
            _log("[INFO] 服务已停止")
            return 0
        is_restart = True
        if completed.returncode == 3:
            _log()
            _log("========================================")
            _log("   检测到配置更新，正在重启服务...")
            _log("========================================")
            time.sleep(2.0)
            continue
        _log()
        _log(f"[ERROR] 服务异常退出 (退出码: {completed.returncode})")
        _log("[INFO] 3 秒后自动重启...")
        time.sleep(3.0)


def main() -> int:
    os.environ.setdefault("PYTHONUTF8", "1")
    os.environ.setdefault("PYTHONIOENCODING", "utf-8")

    _log()
    _log("========================================")
    _log("   Universal Web-to-API 启动脚本")
    _log("========================================")
    _log()

    _section("加载配置")
    _load_env_file(PROJECT_DIR / ".env")
    _apply_env_defaults()
    _display_current_config()

    _check_python_version()
    _run_auto_update()
    _ensure_project_structure()
    _ensure_venv()
    _ensure_dependencies()
    _maybe_apply_patch()
    profile_dir = _resolve_profile_dir()
    _maybe_clean_profile(profile_dir)
    _launch_browser_if_needed()
    _display_version_info()
    _display_start_summary()
    public_port = int(os.getenv("APP_PORT", "8199") or "8199")
    if _env_flag("SCHEDULED_RESTART_ENABLED", False):
        backend_port = _find_backend_port(public_port)
        _start_restart_handoff_proxy(os.getenv("APP_HOST", "127.0.0.1"), public_port, backend_port)
        return _run_service_loop(public_port=public_port, backend_port=backend_port)
    return _run_service_loop()


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        _log()
        _log("[INFO] 已取消启动")
        raise SystemExit(130)
    except Exception as exc:
        _log()
        _log(f"[ERROR] {exc}")
        raise SystemExit(1)