install.py

#!/usr/bin/env python3
"""Webnovel Writer - 安装/更新/卸载工具

交互菜单（无参数）：
    python install.py

命令行模式：
    python install.py --update       # 更新到最新版
    python install.py --incremental  # 增量更新（仅变更文件）
    python install.py --clean        # 清洁安装
    python install.py --uninstall    # 卸载
    python install.py --venv         # 虚拟环境安装
    python install.py --apply        # 应用暂存更新（OpenCode 重启后）
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import platform
import shutil
import subprocess
import sys
import urllib.request
import urllib.error
from pathlib import Path
from typing import Optional

# ── 常量 ────────────────────────────────────────────────────────────────────

PROJECT_ROOT = Path(__file__).resolve().parent
OPENCODE_DIR = PROJECT_ROOT / ".opencode"
VENV_DIR = PROJECT_ROOT / ".venv"
STAGING_DIR = PROJECT_ROOT / ".opencode_staging"
REQUIREMENTS_FILE = OPENCODE_DIR / "installer" / "requirements.txt"
SCRIPTS_DIR = OPENCODE_DIR / "scripts"
SYNC_SCRIPT = PROJECT_ROOT / "scripts" / "sync-upstream.ps1"

UPSTREAM_REPO = "https://github.com/lingfengQAQ/webnovel-writer.git"
OWN_REPO = "https://github.com/Saemer2023/webnovel-writer-opencode.git"

# GitHub 镜像（中国大陆用户自动切换）
GITHUB_MIRROR = "https://mirror.ghproxy.com/https://github.com"

MIN_PYTHON = (3, 10)

# ── 工具函数 ────────────────────────────────────────────────────────────────


def color(text: str, code: str) -> str:
    """终端颜色，Windows 自动检测."""
    if platform.system() == "Windows" and not os.environ.get("TERM"):
        return text  # Windows CMD 不支持 ANSI
    codes = {"green": "32", "yellow": "33", "red": "31", "cyan": "36", "gray": "90", "bold": "1"}
    c = codes.get(code, "0")
    return f"\033[{c}m{text}\033[0m"


def info(msg: str) -> None:
    print(f"  {color('[INFO]', 'cyan')} {msg}")


def ok(msg: str) -> None:
    print(f"  {color('[OK]', 'green')} {msg}")


def warn(msg: str) -> None:
    print(f"  {color('[WARN]', 'yellow')} {msg}")


def fail(msg: str) -> None:
    print(f"  {color('[FAIL]', 'red')} {msg}")


def step(msg: str) -> None:
    print(f"\n{color('==>', 'bold')} {color(msg, 'cyan')}")


def run_cmd(cmd: list[str], cwd: Optional[Path] = None, check: bool = True, capture: bool = False) -> subprocess.CompletedProcess:
    """运行 shell 命令."""
    try:
        if capture:
            return subprocess.run(cmd, cwd=cwd, check=check, capture_output=True, text=True)
        return subprocess.run(cmd, cwd=cwd, check=check)
    except subprocess.CalledProcessError as e:
        if not check:
            return e
        raise


def is_china_network() -> bool:
    """检测是否中国大陆网络."""
    try:
        req = urllib.request.Request("https://www.baidu.com", method="HEAD")
        urllib.request.urlopen(req, timeout=3)
        return True
    except Exception:
        return False


def mirror_url(original: str) -> str:
    """如果在中国，返回镜像 URL."""
    if is_china_network() and original.startswith("https://github.com/"):
        return original.replace("https://github.com", GITHUB_MIRROR, 1)
    return original


def is_opencode_running() -> bool:
    """检测 OpenCode 是否正在使用此项目目录."""
    lock_file = OPENCODE_DIR / ".opencode.lock"
    return lock_file.exists()


# ── 系统检查 ────────────────────────────────────────────────────────────────


def check_python() -> Optional[str]:
    """检查 Python 版本. 返回 python 可执行路径."""
    this_python = sys.executable
    v = sys.version_info
    if (v.major, v.minor) >= MIN_PYTHON:
        return this_python

    # 尝试找系统中其他 Python
    candidates = ["python3", "python"]
    for name in candidates:
        try:
            r = subprocess.run([name, "--version"], capture_output=True, text=True, check=False)
            if r.returncode != 0:
                continue
            parts = r.stdout.strip().split()
            if len(parts) >= 2:
                ver = parts[1].split(".")
                if int(ver[0]) >= 3 and int(ver[1]) >= 10:
                    return shutil.which(name)
        except Exception:
            continue

    return None


def check_git() -> bool:
    """检查 git 是否可用."""
    try:
        subprocess.run(["git", "--version"], capture_output=True, check=True)
        return True
    except Exception:
        return False


def check_venv() -> Optional[Path]:
    """检查 .venv 是否存在，返回 python 路径."""
    if platform.system() == "Windows":
        python_path = VENV_DIR / "Scripts" / "python.exe"
    else:
        python_path = VENV_DIR / "bin" / "python"

    if python_path.exists():
        return python_path
    return None


# ── 核心操作 ────────────────────────────────────────────────────────────────


def create_venv(python: str) -> Optional[Path]:
    """创建虚拟环境."""
    step("Creating virtual environment...")
    if VENV_DIR.exists():
        warn(f"Virtual env already exists: {VENV_DIR}")
        if platform.system() == "Windows":
            return VENV_DIR / "Scripts" / "python.exe"
        return VENV_DIR / "bin" / "python"

    try:
        subprocess.run([python, "-m", "venv", str(VENV_DIR)], check=True)
        ok("Virtual environment created")
        if platform.system() == "Windows":
            return VENV_DIR / "Scripts" / "python.exe"
        return VENV_DIR / "bin" / "python"
    except subprocess.CalledProcessError as e:
        fail(f"Failed to create virtual env: {e}")
        return None


def install_deps(python_path: Path) -> bool:
    """安装 Python 依赖."""
    step("Installing dependencies...")

    if not REQUIREMENTS_FILE.exists():
        fail(f"requirements.txt not found: {REQUIREMENTS_FILE}")
        return False

    # pip 镜像（中国用户）
    pip_args = [str(python_path), "-m", "pip", "install", "-r", str(REQUIREMENTS_FILE)]
    if is_china_network():
        pip_args.extend(["-i", "https://pypi.tuna.tsinghua.edu.cn/simple"])
        info("Using Tsinghua PyPI mirror")

    try:
        subprocess.run(pip_args, check=True)
        ok("Dependencies installed")
        return True
    except subprocess.CalledProcessError as e:
        fail(f"Dependency installation failed: {e}")
        return False


def verify_install(python_path: Path) -> bool:
    """验证环境是否就绪."""
    step("Verifying installation...")
    try:
        code = "import yaml, aiohttp, pydantic, filelock; print('OK')"
        r = subprocess.run([str(python_path), "-c", code], capture_output=True, text=True, check=True)
        if "OK" in r.stdout:
            ok("All dependencies verified")
            return True
        fail("Dependency verification failed")
        return False
    except subprocess.CalledProcessError as e:
        fail(f"Verification error: {e.stderr.strip()}")
        return False


def sync_upstream_files(incremental: bool = False) -> bool:
    """从上游仓库同步文件到 .opencode/."""
    step("Syncing upstream files...")

    import tempfile
    import shutil

    temp_dir = Path(tempfile.mkdtemp(prefix="webnovel-upstream-"))

    try:
        url = mirror_url(UPSTREAM_REPO)
        info(f"Cloning upstream from {url}")
        subprocess.run(
            ["git", "clone", "--depth", "1", url, str(temp_dir)],
            check=True, capture_output=True
        )

        # 上游项目内容在 webnovel-writer/ 子目录下
        upstream_src = temp_dir / "webnovel-writer"
        if not upstream_src.exists():
            upstream_src = temp_dir

        sync_dirs = ["scripts", "references", "genres", "templates", "dashboard", "skills", "agents"]
        protected = [
            "scripts/publisher",
            "scripts/export_manager",
            "scripts/gen_manifest.py",
            "installer",
            "skills/webnovel-export",
            "skills/webnovel-publish",
            "skills/webnovel-write-batch",
            "agents/deconstruction-agent.md",
            "agents/chapter-writer-agent.md",
        ]

        changed = 0
        added = 0
        skipped = 0

        for dir_name in sync_dirs:
            src = upstream_src / dir_name
            dst = OPENCODE_DIR / dir_name

            if not src.exists():
                continue

            if not dst.exists():
                dst.mkdir(parents=True)
                if incremental:
                    info(f"New directory: {dir_name}/")
                added += 1

            for src_file in src.rglob("*"):
                if src_file.is_dir():
                    continue

                rel = src_file.relative_to(src)
                check_rel = str(Path(dir_name) / rel).replace("\\", "/")

                # 检查是否受保护
                is_protected = False
                for p in protected:
                    if check_rel.startswith(p) or check_rel == p:
                        is_protected = True
                        break
                if is_protected:
                    skipped += 1
                    continue

                dst_file = dst / rel
                dst_file.parent.mkdir(parents=True, exist_ok=True)

                if dst_file.exists():
                    # 比较哈希
                    with open(src_file, "rb") as f:
                        src_hash = hashlib.sha256(f.read()).hexdigest()
                    with open(dst_file, "rb") as f:
                        dst_hash = hashlib.sha256(f.read()).hexdigest()
                    if src_hash == dst_hash:
                        continue

                if incremental and not dst_file.exists():
                    continue  # 增量模式只更新已有文件

                shutil.copy2(src_file, dst_file)
                if dst_file.exists():
                    changed += 1

        ok(f"Synced: {changed} changed, {added} added, {skipped} protected skipped")
        return True

    except subprocess.CalledProcessError as e:
        fail(f"Failed to sync upstream: {e.stderr.decode() if e.stderr else str(e)}")
        return False
    except Exception as e:
        fail(f"Sync error: {e}")
        return False
    finally:
        if temp_dir.exists():
            shutil.rmtree(temp_dir, ignore_errors=True)


def self_update(incremental: bool) -> bool:
    """从 GitHub 更新自身（项目代码）. 如果 OpenCode 正在运行则暂存."""
    step("Checking for updates...")

    is_running = is_opencode_running()
    target_dir = STAGING_DIR if is_running else PROJECT_ROOT

    url = mirror_url(OWN_REPO)
    info(f"Fetching from {url}")

    try:
        # 克隆到临时目录
        import tempfile
        import shutil

        temp_dir = Path(tempfile.mkdtemp(prefix="webnovel-self-update-"))

        subprocess.run(
            ["git", "clone", "--depth", "1", url, str(temp_dir)],
            check=True, capture_output=True
        )

        # 如果是暂存模式，复制到 staging 目录
        if is_running:
            info("OpenCode is running. Saving to staging directory...")
            if target_dir.exists():
                shutil.rmtree(target_dir)
            shutil.copytree(temp_dir, target_dir)
            ok(f"Update staged to {target_dir}")
            warn("Restart OpenCode, then run: python install.py --apply")
            shutil.rmtree(temp_dir, ignore_errors=True)
            return True

        # 直接更新（非暂存模式）
        # 排除 .venv, .git, .opencode_staging, __pycache__
        excludes = {".venv", ".git", ".opencode_staging", "__pycache__", ".pytest_cache", ".tmp"}
        changes = 0

        for item in temp_dir.iterdir():
            if item.name in excludes:
                continue
            dst = target_dir / item.name
            if incremental and not dst.exists():
                continue  # 增量模式只更新已有的

            if dst.exists():
                if dst.is_dir():
                    shutil.rmtree(dst)
                else:
                    dst.unlink()

            if item.is_dir():
                shutil.copytree(item, dst)
            else:
                shutil.copy2(item, dst)
            changes += 1

        shutil.rmtree(temp_dir, ignore_errors=True)

        if changes == 0:
            ok("Already up to date")
        else:
            ok(f"Updated {changes} items")

        return True

    except subprocess.CalledProcessError as e:
        stderr = e.stderr.decode() if e.stderr else str(e)
        fail(f"Update failed: {stderr}")
        return False
    except Exception as e:
        fail(f"Update error: {e}")
        return False


def apply_staging() -> bool:
    """应用暂存的更新."""
    step("Applying staged update...")

    if not STAGING_DIR.exists():
        fail("No staged update found")
        return False

    excludes = {".venv", ".git", ".opencode_staging", "__pycache__", ".pytest_cache", ".tmp"}
    changes = 0

    for item in STAGING_DIR.iterdir():
        if item.name in excludes:
            continue
        dst = PROJECT_ROOT / item.name
        if dst.exists():
            if dst.is_dir():
                shutil.rmtree(dst)
            else:
                dst.unlink()
        if item.is_dir():
            shutil.copytree(item, dst)
        else:
            shutil.copy2(item, dst)
        changes += 1

    shutil.rmtree(STAGING_DIR, ignore_errors=True)
    ok(f"Applied {changes} items")
    return True


def uninstall() -> bool:
    """卸载."""
    step("Uninstalling...")

    removed = []

    if VENV_DIR.exists():
        shutil.rmtree(VENV_DIR, ignore_errors=True)
        removed.append(str(VENV_DIR))

    if STAGING_DIR.exists():
        shutil.rmtree(STAGING_DIR, ignore_errors=True)
        removed.append(str(STAGING_DIR))

    # 清除 __pycache__ 和 .pytest_cache
    for pattern in ("__pycache__", ".pytest_cache", ".tmp"):
        for p in PROJECT_ROOT.rglob(pattern):
            if p.is_dir():
                shutil.rmtree(p, ignore_errors=True)
                removed.append(str(p))

    # 清除 .venv 相关 pip 缓存不处理（全局）
    for item in removed:
        ok(f"Removed: {item}")

    if not removed:
        warn("Nothing to uninstall")
    else:
        ok(f"Uninstalled {len(removed)} items")

    info("To fully remove, delete this directory manually.")
    return True


def clean_install() -> bool:
    """清洁安装：卸载 + 重新安装."""
    step("Clean install...")
    warn("This will remove .venv and reinstall everything.")

    confirm = input(f"  {color('Continue? [y/N] ', 'yellow')}")
    if confirm.lower() != "y":
        info("Cancelled")
        return False

    uninstall()

    python = check_python()
    if not python:
        fail("Python 3.10+ required")
        return False

    py = create_venv(python)
    if not py:
        return False

    if not install_deps(py):
        return False

    if not verify_install(py):
        return False

    # 同步上游文件
    sync_upstream_files()

    ok("Clean install complete")
    return True


# ── 交互式菜单 ──────────────────────────────────────────────────────────────


def interactive_menu() -> None:
    """交互式主菜单."""
    print(f"""
{color('=' * 50, 'cyan')}
{color('  Webnovel Writer Installer', 'bold')}
{color('=' * 50, 'cyan')}
""")

    # 系统预检
    python_path = check_python()
    git_ok = check_git()
    venv_path = check_venv()

    # 显示状态
    status = []
    if python_path:
        py_ver = subprocess.run([python_path, "--version"], capture_output=True, text=True).stdout.strip()
        status.append(f"  {color('[OK]', 'green')} Python: {py_ver}")
    else:
        status.append(f"  {color('[FAIL]', 'red')} Python 3.10+ not found")

    if git_ok:
        git_ver = subprocess.run(["git", "--version"], capture_output=True, text=True).stdout.strip()
        status.append(f"  {color('[OK]', 'green')} Git: {git_ver}")
    else:
        status.append(f"  {color('[FAIL]', 'red')} Git not found")

    if venv_path:
        status.append(f"  {color('[OK]', 'green')} Virtual env: {venv_path}")
    else:
        status.append(f"  {color('[--]', 'gray')} Virtual env: not created")

    if is_china_network():
        status.append(f"  {color('[INFO]', 'cyan')} China network detected, using mirrors")

    print("System Status:")
    for s in status:
        print(s)

    # 菜单选项
    print(f"""
{color('Options:', 'bold')}
  {color('1)', 'cyan')} Install / Update    - 安装或更新到最新版
  {color('2)', 'cyan')} Incremental Update  - 仅更新已存在的文件
  {color('3)', 'cyan')} Clean Install       - 重新安装
  {color('4)', 'cyan')} Uninstall           - 卸载
  {color('5)', 'cyan')} Verify              - 验证安装
  {color('6)', 'cyan')} Sync Upstream       - 同步上游素材
  {color('0)', 'cyan')} Exit
""")

    choice = input(f"  {color('Select [0-6]: ', 'yellow')}").strip()

    if choice == "1":
        if not python_path:
            fail("Python 3.10+ required")
            input("\n  Press Enter to return...")
            return
        py = venv_path or create_venv(python_path)
        if py:
            install_deps(py)
            verify_install(py)
        self_update(incremental=False)

    elif choice == "2":
        py = venv_path
        if not py:
            warn("No virtual env found. Run option 1 first.")
        else:
            verify_install(py)
        self_update(incremental=True)

    elif choice == "3":
        clean_install()

    elif choice == "4":
        confirm = input(f"  {color('Remove .venv and caches? [y/N] ', 'yellow')}")
        if confirm.lower() == "y":
            uninstall()
        else:
            info("Cancelled")

    elif choice == "5":
        py = check_venv()
        if not py:
            fail("Not installed. Run option 1 first.")
        else:
            verify_install(py)

    elif choice == "6":
        sync_upstream_files()

    elif choice == "0":
        print("  Goodbye!")

    input(f"\n  {color('Press Enter to return...', 'gray')}")


# ── 命令行入口 ──────────────────────────────────────────────────────────────


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Webnovel Writer - Installation and update tool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python install.py              # Interactive menu
  python install.py --update     # Update to latest
  python install.py --incremental # Incremental update
  python install.py --clean      # Clean install
  python install.py --uninstall  # Uninstall
  python install.py --venv       # Virtual env install (interactive fallback)
  python install.py --apply      # Apply staged update
        """,
    )
    parser.add_argument("--update", action="store_true", help="Update to latest version")
    parser.add_argument("--incremental", action="store_true", help="Incremental update (changed files only)")
    parser.add_argument("--clean", action="store_true", help="Clean install")
    parser.add_argument("--uninstall", action="store_true", help="Uninstall")
    parser.add_argument("--venv", action="store_true", help="Create virtual environment and install")
    parser.add_argument("--apply", action="store_true", help="Apply staged update")
    args = parser.parse_args()

    try:
        if args.uninstall:
            uninstall()
        elif args.clean:
            clean_install()
        elif args.update:
            python_path = check_python()
            if python_path:
                py = check_venv() or create_venv(python_path)
                if py:
                    install_deps(py)
            self_update(incremental=False)
        elif args.incremental:
            self_update(incremental=True)
        elif args.apply:
            apply_staging()
        elif args.venv:
            python_path = check_python()
            if not python_path:
                fail("Python 3.10+ required")
                sys.exit(1)
            py = check_venv() or create_venv(python_path)
            if py:
                install_deps(py)
                verify_install(py)
        else:
            # 无参数 → 交互菜单
            while True:
                interactive_menu()
    except KeyboardInterrupt:
        print(f"\n  {color('Interrupted.', 'yellow')}")
        sys.exit(1)


if __name__ == "__main__":
    main()