Nexus/server/api/install.py

"""Nexus — Installation Wizard API

Runs BEFORE the main app is fully configured (no .env required).
Provides endpoints for the 5-step install wizard.

Security: All endpoints refuse to operate once .env exists or install is locked.
"""

import os
import secrets
import logging
import subprocess
from pathlib import Path
from datetime import datetime, timezone
from urllib.parse import quote_plus, urlparse

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

from server.infrastructure.database.engine_compat import patch_aiomysql_do_ping
from server.utils.text_io import read_utf8_text, write_utf8_lf

logger = logging.getLogger("nexus.install")

router = APIRouter(prefix="/api/install", tags=["install"])

# ── Paths ──

ROOT_DIR = Path(__file__).resolve().parent.parent.parent
ENV_FILE = ROOT_DIR / ".env"
INSTALL_LOCK = ROOT_DIR / ".install_locked"
CONFIG_DIR = ROOT_DIR / "web" / "data"
CONFIG_JSON = CONFIG_DIR / "config.json"
ONEPANEL_HOSTS_JSON = CONFIG_DIR / "1panel-hosts.json"
STATE_FILE = ROOT_DIR / ".install_state.json"


def _has_env() -> bool:
    """Python backend configured (.env written at install step 3)."""
    return ENV_FILE.exists()


def _is_locked() -> bool:
    return INSTALL_LOCK.exists()


def _is_installed() -> bool:
    """Install wizard fully complete — same as locked (backward-compatible for /status)."""
    return _is_locked()


# ── Pydantic Models ──

class InitDbRequest(BaseModel):
    db_host: str = "localhost"
    db_port: str = "3306"
    db_name: str = "Nexus"
    db_user: str = "Nexus"
    db_pass: str
    redis_host: str = "127.0.0.1"
    redis_port: str = "6379"
    redis_db: str = "0"
    redis_pass: str = ""
    api_port: str = "8600"
    timezone: str = "Asia/Shanghai"
    site_url: str = ""


class CreateAdminRequest(BaseModel):
    install_token: str
    db_host: str = "localhost"
    db_port: str = "3306"
    db_name: str = "Nexus"
    db_user: str = "Nexus"
    db_pass: str
    admin_username: str = "admin"
    admin_password: str
    admin_email: str = ""
    system_name: str = "Nexus"
    system_title: str = "Nexus — 服务器运维管理平台"


# ── Helpers ──

def _make_db_url(req: InitDbRequest | CreateAdminRequest) -> str:
    return (
        f"mysql+aiomysql://{req.db_user}:{quote_plus(req.db_pass)}"
        f"@{req.db_host}:{req.db_port}/{req.db_name}"
    )


def _build_redis_url(req: InitDbRequest) -> str:
    url = "redis://"
    if req.redis_pass:
        url += f"{quote_plus(req.redis_pass)}@"
    url += f"{req.redis_host}:{req.redis_port}/{req.redis_db}"
    return url


def _read_1panel_hosts_file() -> dict[str, str]:
    if not ONEPANEL_HOSTS_JSON.is_file():
        return {}
    import json

    try:
        raw = json.loads(read_utf8_text(ONEPANEL_HOSTS_JSON))
    except (json.JSONDecodeError, OSError):
        return {}
    if not isinstance(raw, dict):
        return {}
    return {str(k): str(v) for k, v in raw.items() if v}


def _onepanel_service_host(kind: str) -> str | None:
    """1Panel App Store container name (env → web/data/1panel-hosts.json)."""
    val = os.environ.get(f"NEXUS_1PANEL_{kind}_HOST", "").strip()
    if val:
        return val
    file_key = "db_host" if kind == "DB" else "redis_host"
    return _read_1panel_hosts_file().get(file_key) or None


def _resolve_redis_probe() -> tuple[str, int, str | None]:
    """Redis target for install wizard (1Panel: container name on 1panel-network)."""
    onepanel = _onepanel_service_host("REDIS")
    if onepanel:
        return onepanel, 6379, None

    url = os.environ.get("NEXUS_REDIS_URL", "").strip()
    if url:
        parsed = urlparse(url)
        host = parsed.hostname or "127.0.0.1"
        port = parsed.port or 6379
        return host, port, parsed.password

    host = os.environ.get("REDIS_HOST", "127.0.0.1")
    port = int(os.environ.get("REDIS_PORT", "6379"))
    return host, port, None


def _service_tcp_targets(port: int, onepanel_kind: str) -> list[tuple[str, int]]:
    """Probe order: 1Panel container name → host.docker.internal → loopback."""
    targets: list[tuple[str, int]] = []
    if os.environ.get("NEXUS_DOCKER_WRITE_ENV") == "1":
        host = _onepanel_service_host(onepanel_kind)
        if host:
            targets.append((host, port))
        targets.append(("host.docker.internal", port))
    targets.append(("127.0.0.1", port))
    seen: set[tuple[str, int]] = set()
    ordered: list[tuple[str, int]] = []
    for item in targets:
        if item in seen:
            continue
        seen.add(item)
        ordered.append(item)
    return ordered


def _probe_tcp_service(port: int, onepanel_kind: str) -> tuple[bool, str]:
    import socket

    for host, p in _service_tcp_targets(port, onepanel_kind):
        try:
            with socket.create_connection((host, p), timeout=2.0):
                return True, f"✓ 已安装（{host}:{p} 可访问）"
        except OSError:
            continue
    hint_host = _onepanel_service_host(onepanel_kind)
    if not hint_host and os.environ.get("NEXUS_DOCKER_WRITE_ENV") == "1":
        hint_host = "host.docker.internal"
    if not hint_host:
        hint_host = "127.0.0.1"
    return False, f"✗ 未检测到服务（{hint_host}:{port}）"


def _probe_redis_installed() -> tuple[bool, str]:
    """Step 2: TCP probe only — Redis 是否可访问（不验证密码/库号）。"""
    return _probe_tcp_service(6379, "REDIS")


def _probe_mysql_installed() -> tuple[bool, str]:
    """Step 2: TCP probe — MySQL 是否可访问（不验证账号密码）。"""
    return _probe_tcp_service(3306, "DB")


def _service_tcp_reachable(host: str, port: int, timeout: float = 3.0) -> bool:
    import socket

    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def _mysql_auth_error_detail(db_user: str) -> str:
    """1045 Access denied — common on 1Panel when user is only nexus@localhost."""
    if os.environ.get("NEXUS_DOCKER_WRITE_ENV") != "1":
        return (
            f"MySQL 拒绝用户 {db_user!r} 的密码或主机权限。"
            "请核对库名/用户名/密码，并确认该用户允许从应用主机连接（非仅 localhost）。"
        )
    return (
        f"MySQL 拒绝用户 {db_user!r}（错误 1045）。Nexus 在 Docker 内通过 1panel-network 连接，"
        "来源 IP 为容器网段（如 172.18.x.x），不是 localhost。\n"
        "请在 1Panel MySQL 中确保：① 库 nexus、用户 nexus 已创建 ② 密码与向导一致 "
        "③ 存在 nexus@'%' 或允许 Docker 网段（仅 nexus@localhost 会失败）。\n"
        "服务器执行: cd /opt/nexus && bash deploy/fix-1panel-mysql-grant.sh"
    )


def _mysql_connect_error_detail(host: str, port: int) -> str:
    if os.environ.get("NEXUS_DOCKER_WRITE_ENV") != "1":
        return ""
    onepanel_db = _onepanel_service_host("DB")
    if onepanel_db and host == onepanel_db:
        return (
            f"无法连接 1Panel MySQL 容器 {host}:{port}。"
            "请确认：① MySQL 容器运行中 ② Nexus 已接入 1panel-network（服务器执行 nx update）"
            "③ 已在 1Panel 创建库 nexus 与用户 nexus。"
        )
    if host in ("host.docker.internal", "172.17.0.1"):
        example = onepanel_db or "1Panel-mysql-xxxx"
        return (
            f"无法通过 {host} 连接 MySQL。"
            "1Panel 应用商店 MySQL 在 1panel-network 内，容器间应使用 MySQL 容器名"
            f"（如 {example}），而非 host.docker.internal。"
            "在服务器执行 nx update 可自动接入网络并预填容器名。"
        )
    return ""


def _redis_connect_error_detail(host: str, port: int) -> str:
    if os.environ.get("NEXUS_DOCKER_WRITE_ENV") != "1":
        return ""
    onepanel_redis = _onepanel_service_host("REDIS")
    if onepanel_redis and host == onepanel_redis:
        return (
            f"无法连接 1Panel Redis 容器 {host}:{port}。"
            "请确认：① Redis 容器运行中 ② Nexus 已接入 1panel-network（服务器执行 nx update）"
            "③ 若 1Panel Redis 设置了密码，请在步骤 3 填写 redis_pass。"
        )
    if host in ("host.docker.internal", "172.17.0.1"):
        example = onepanel_redis or "1Panel-redis-xxxx"
        return (
            f"无法通过 {host} 连接 Redis。"
            "1Panel 应用商店 Redis 在 1panel-network 内，容器间应使用 Redis 容器名"
            f"（如 {example}），而非 host.docker.internal。"
            "在服务器执行 nx update 可自动接入并预填容器名。"
        )
    return ""


def _parse_dotenv(path: Path) -> dict[str, str]:
    data: dict[str, str] = {}
    if not path.is_file():
        return data
    for line in read_utf8_text(path).splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, raw = line.partition("=")
        val = raw.strip()
        if len(val) >= 2 and val[0] == val[-1] == '"':
            val = val[1:-1].replace('\\"', '"').replace("\\\\", "\\")
        data[key.strip()] = val
    return data


async def _test_mysql_url(db_url: str) -> tuple[bool, str]:
    if not db_url:
        return False, "✗ 未配置 NEXUS_DATABASE_URL"
    try:
        patch_aiomysql_do_ping()
        engine = create_async_engine(db_url, pool_pre_ping=True, pool_recycle=300)
        async with engine.connect() as conn:
            await conn.execute(text("SELECT 1"))
        await engine.dispose()
        return True, "✓ MySQL 连接正常"
    except Exception as e:
        err = str(e)
        if "1045" in err or "Access denied" in err:
            return False, _mysql_auth_error_detail(
                urlparse(db_url).username or "nexus"
            )
        return False, f"✗ MySQL 连接失败: {e}"


def _test_redis_url(redis_url: str) -> tuple[bool, str]:
    if not redis_url:
        return False, "✗ 未配置 NEXUS_REDIS_URL"
    parsed = urlparse(redis_url)
    host = parsed.hostname or "127.0.0.1"
    port = parsed.port or 6379
    if not _service_tcp_reachable(host, port):
        detail = _redis_connect_error_detail(host, port)
        return False, detail or f"✗ Redis TCP 不通（{host}:{port}）"
    try:
        import redis as rlib

        client = rlib.Redis.from_url(redis_url, socket_timeout=3.0)
        client.ping()
        return True, "✓ Redis 连接正常"
    except Exception as e:
        err = str(e)
        if "Connection refused" in err or "Error 111" in err:
            detail = _redis_connect_error_detail(host, port)
            if detail:
                return False, detail
        return False, f"✗ Redis 连接失败: {e}"


def _docker_wizard_defaults() -> dict[str, str] | None:
    """Prefill install step 3 when running inside docker-compose.prod.yml."""
    if os.environ.get("NEXUS_DOCKER_WRITE_ENV") != "1":
        return None
    db_host = _onepanel_service_host("DB") or "host.docker.internal"
    redis_host = _onepanel_service_host("REDIS") or "host.docker.internal"
    return {
        "db_host": db_host,
        "db_port": "3306",
        "db_name": "nexus",
        "db_user": "nexus",
        "redis_host": redis_host,
        "redis_port": "6379",
        "redis_db": "0",
    }


def _install_keys_from_environment() -> tuple[str, str, str] | None:
    """Reuse keys pre-generated by Docker install (docker/.env.prod → compose env)."""
    import os

    sk = os.environ.get("NEXUS_SECRET_KEY", "").strip()
    ak = os.environ.get("NEXUS_API_KEY", "").strip()
    ek = os.environ.get("NEXUS_ENCRYPTION_KEY", "").strip()
    if sk and ak and ek:
        return sk, ak, ek
    return None


def _escape_env_value(value: str) -> str:
    """Escape a value for safe inclusion in a .env file.

    .env files are essentially shell source files — unquoted values break
    on spaces, ``#``, ``$``, newlines, backslashes, etc.  The safest
    approach is to double-quote the value and escape any internal
    double-quotes and backslashes (Docker/python-dotenv convention).
    """
    escaped = value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
    return f'"{escaped}"'


def _write_env(req: InitDbRequest, secret_key: str, api_key: str, encryption_key: str,
               pool_size: int, max_overflow: int, redis_url: str,
               site_url: str) -> None:
    install_dir = str(ROOT_DIR)
    db_url = (
        f"mysql+aiomysql://{req.db_user}:{quote_plus(req.db_pass)}"
        f"@{req.db_host}:{req.db_port}/{req.db_name}"
    )

    lines = [
        "# Nexus .env — Auto-generated by installer",
        f"# {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC",
        "",
        f"NEXUS_SYSTEM_NAME={_escape_env_value('Nexus')}",
        f"NEXUS_SYSTEM_TITLE={_escape_env_value('Nexus — 服务器运维管理平台')}",
        "",
        "NEXUS_HOST=0.0.0.0",
        f"NEXUS_PORT={req.api_port}",
        "",
        "# Database (MySQL 8.4 + aiomysql async driver)",
        f"NEXUS_DATABASE_URL={_escape_env_value(db_url)}",
        f"NEXUS_DB_POOL_SIZE={pool_size}",
        f"NEXUS_DB_MAX_OVERFLOW={max_overflow}",
        "",
        "# Security — PRODUCTION KEYS (immutable after install)",
        f"NEXUS_SECRET_KEY={_escape_env_value(secret_key)}",
        f"NEXUS_API_KEY={_escape_env_value(api_key)}",
        f"NEXUS_ENCRYPTION_KEY={_escape_env_value(encryption_key)}",
        "",
        "# Redis",
        f"NEXUS_REDIS_URL={_escape_env_value(redis_url)}",
        "",
        "# Deployment",
        f"NEXUS_DEPLOY_PATH={_escape_env_value(install_dir)}",
        f"NEXUS_CORS_ORIGINS={_escape_env_value(f'{site_url},http://localhost:{req.api_port}')}",
        "",
        "# SSH",
        "NEXUS_SSH_STRICT_HOST_CHECKING=true",
        "",
        "# Health Check",
        "NEXUS_HEALTH_CHECK_INTERVAL=60",
        f"NEXUS_API_BASE_URL={_escape_env_value(site_url)}",
        "",
        "# Alert Thresholds",
        "NEXUS_CPU_ALERT_THRESHOLD=80",
        "NEXUS_MEM_ALERT_THRESHOLD=80",
        "NEXUS_DISK_ALERT_THRESHOLD=80",
        "",
        "# Telegram Alerts (optional — configure in Settings UI)",
        "NEXUS_TELEGRAM_BOT_TOKEN=",
        "NEXUS_TELEGRAM_CHAT_ID=",
        "",
    ]

    write_utf8_lf(ENV_FILE, "\n".join(lines) + "\n")
    logger.info(f".env written to {ENV_FILE}")


def _write_config_json(req: InitDbRequest, api_key: str, site_url: str) -> None:
    """Write shared configuration as JSON (replaces legacy config.php)."""
    import json
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    install_dir = str(ROOT_DIR)

    config = {
        "_generated": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"),
        "_generator": "Nexus 6.0 Installer",
        "api_base_url": site_url,
        "api_key": api_key,
        "db_host": req.db_host,
        "db_port": req.db_port,
        "db_name": req.db_name,
        "db_user": req.db_user,
        "app_name": "Nexus",
        "app_version": "6.0.0",
        "session_lifetime": 28800,
        "deploy_root": install_dir,
        "timezone": req.timezone,
    }

    write_utf8_lf(CONFIG_JSON, json.dumps(config, indent=2, ensure_ascii=False) + "\n")
    logger.info(f"config.json written to {CONFIG_JSON}")


def _detect_bt_panel() -> bool:
    """Detect if BT Panel (宝塔面板) is installed on this system."""
    return Path("/www/server/panel/class/common.py").exists()


def _configure_guardian(install_dir: str, api_port: str) -> list[str]:
    """Best-effort: configure Supervisor, crontab, health_monitor.sh"""
    results: list[str] = []
    bt_panel = _detect_bt_panel()

    # 1. Log directory
    if bt_panel:
        log_dir = Path("/www/wwwlogs")
    else:
        log_dir = Path("/var/log/nexus")
    try:
        log_dir.mkdir(parents=True, exist_ok=True)
        results.append(f"✓ 日志目录已创建 {log_dir}")
    except OSError:
        results.append("✗ 日志目录创建失败（需 root 权限）")

    # 2. Supervisor config
    if bt_panel:
        conf_path = Path("/www/server/panel/plugin/supervisor/profile/nexus.ini")
    else:
        conf_path = Path("/etc/supervisor/conf.d/nexus.conf")

    supervisor_conf = (
        f"[program:nexus]\n"
        f"command={install_dir}/venv/bin/uvicorn server.main:app --host 127.0.0.1 --port {api_port}\n"
        f"directory={install_dir}\n"
        f"user=root\n"
        f"autostart=true\n"
        f"autorestart=true\n"
        f"startretries=10\n"
        f"startsecs=3\n"
        f"stopwaitsecs=10\n"
        f"stopsignal=INT\n"
        f"environment=PATH=\"{install_dir}/venv/bin:/usr/local/bin:%(ENV_PATH)s\",HOME=\"/root\"\n"
        f"stderr_logfile={log_dir}/nexus_error.log\n"
        f"stdout_logfile={log_dir}/nexus_access.log\n"
        f"stderr_logfile_maxbytes=50MB\n"
        f"stdout_logfile_maxbytes=50MB\n"
        f"stderr_logfile_backups=5\n"
        f"stdout_logfile_backups=5\n"
    )
    try:
        conf_path.parent.mkdir(parents=True, exist_ok=True)
        write_utf8_lf(conf_path, supervisor_conf)
        mode_label = "宝塔" if bt_panel else "标准"
        results.append(f"✓ Supervisor 配置已写入 {conf_path}（{mode_label}模式）")
    except OSError:
        results.append("✗ Supervisor 配置写入失败（需 root 权限）")

    # 3. Update health_monitor.sh INSTALL_DIR
    health_sh = Path(install_dir) / "deploy" / "health_monitor.sh"
    if health_sh.exists():
        try:
            content = read_utf8_text(health_sh)
            import re
            content = re.sub(
                r'INSTALL_DIR="[^"]*"',
                f'INSTALL_DIR="{install_dir}"',
                content,
            )
            write_utf8_lf(health_sh, content)
            health_sh.chmod(0o755)
            results.append("✓ health_monitor.sh INSTALL_DIR 已更新")
        except OSError:
            results.append("✗ health_monitor.sh 更新失败")
    else:
        results.append("⚠ health_monitor.sh 未找到，跳过")

    # 4. Crontab
    cron_entry = f"* * * * * {install_dir}/deploy/health_monitor.sh"
    try:
        current = subprocess.run(
            ["crontab", "-l"], capture_output=True, text=True, timeout=5
        ).stdout or ""
        if "health_monitor.sh" not in current:
            new_cron = current.rstrip() + "\n" + cron_entry + "\n"
            proc = subprocess.run(
                ["crontab", "-"], input=new_cron, capture_output=True, text=True, timeout=5
            )
            if proc.returncode == 0:
                results.append("✓ Crontab 已配置（每分钟健康检查）")
            else:
                results.append("✗ Crontab 配置失败 — 请手动添加")
        else:
            results.append("✓ Crontab 已存在 health_monitor 条目")
    except Exception:
        results.append("⚠ Crontab 配置跳过（权限不足或 cron 不可用）")

    # 5. Reload Supervisor
    try:
        subprocess.run(["supervisorctl", "reread"], capture_output=True, timeout=5)
        subprocess.run(["supervisorctl", "update"], capture_output=True, timeout=5)
        results.append("✓ Supervisor 已重载配置")
    except Exception:
        results.append("⚠ Supervisor 未运行 — 请手动执行 supervisorctl reread && supervisorctl update")

    return results


def _finalize_install_lock() -> None:
    """Write install lock marker and remove wizard state file."""
    write_utf8_lf(
        INSTALL_LOCK,
        f"Nexus installation locked at {datetime.now(timezone.utc).isoformat()}\n",
    )
    if STATE_FILE.exists():
        try:
            STATE_FILE.unlink()
        except OSError as e:
            logger.warning("Failed to remove install state file: %s", e)


def _verify_install_token(provided: str) -> None:
    """Require one-time install token issued at step 3 (init-db)."""
    if not provided or not provided.strip():
        raise HTTPException(status_code=403, detail="缺少安装令牌，请从步骤3重新开始")
    if not STATE_FILE.exists():
        raise HTTPException(status_code=403, detail="安装会话无效，请从步骤3重新开始")
    import json
    try:
        state = json.loads(read_utf8_text(STATE_FILE))
    except (json.JSONDecodeError, OSError) as exc:
        raise HTTPException(status_code=403, detail="安装会话无效，请从步骤3重新开始") from exc
    expected = state.get("install_token")
    if not expected or not secrets.compare_digest(provided.strip(), expected):
        raise HTTPException(status_code=403, detail="安装令牌无效或已过期")


# ── Endpoints ──

@router.get("/status")
async def install_status():
    """Check whether Nexus is already installed."""
    if _is_locked():
        raise HTTPException(status_code=404, detail="Not found")
    return {
        "installed": _is_installed(),
        "locked": _is_locked(),
        "configured": _has_env(),
    }


def _reject_if_locked():
    """Block install wizard mutations after step 5 lock."""
    if _is_locked():
        raise HTTPException(
            status_code=403,
            detail="安装向导已锁定。仅可查询 /api/install/status",
        )


def _reject_if_env_exists():
    """Block repeat database init (step 3) once .env exists."""
    if _has_env():
        raise HTTPException(
            status_code=403,
            detail="系统已初始化，无法重复执行数据库安装。请继续创建管理员或前往登录。",
        )


def _reject_if_installed():
    """Backward-compatible name — means wizard locked, not merely .env present."""
    _reject_if_locked()


# Steps 2–5 (until lock): allow while .env exists
_reject_post_install_wizard = _reject_if_locked


@router.get("/env-check")
async def env_check():
    """Detect environment readiness — Python, MySQL client, Redis, write permissions."""
    _reject_post_install_wizard()
    import sys

    checks = []

    # Python version
    checks.append({
        "name": "Python 版本",
        "required": "3.12+",
        "current": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
        "pass": sys.version_info >= (3, 12),
    })

    # MySQL client (aiomysql)
    try:
        import aiomysql
        checks.append({
            "name": "aiomysql 驱动",
            "required": "已安装",
            "current": f"✓ {aiomysql.__version__}",
            "pass": True,
        })
    except ImportError:
        checks.append({
            "name": "aiomysql 驱动",
            "required": "已安装",
            "current": "✗ 未安装",
            "pass": False,
        })

    # Redis — 步骤2仅检测宿主机是否已安装（TCP）；步骤4再验证连接
    redis_installed, redis_msg = _probe_redis_installed()
    checks.append({
        "name": "Redis（宿主机）",
        "required": "建议已安装",
        "current": redis_msg,
        "pass": redis_installed,
        "blocks_wizard": False,
    })

    # MySQL — 步骤2 TCP 检测；步骤4再验证账号密码
    mysql_installed, mysql_msg = _probe_mysql_installed()
    checks.append({
        "name": "MySQL（宿主机）",
        "required": "建议已安装",
        "current": mysql_msg,
        "pass": mysql_installed,
        "blocks_wizard": False,
    })

    # Write permissions
    data_writable = CONFIG_DIR.exists() and os.access(CONFIG_DIR, os.W_OK) or \
                    os.access(ROOT_DIR / "web", os.W_OK)
    root_writable = os.access(ROOT_DIR, os.W_OK)
    checks.append({
        "name": "web/data 写入权限",
        "required": "可写",
        "current": "✓ 可写" if data_writable else "✗ 不可写",
        "pass": data_writable,
    })
    checks.append({
        "name": "根目录写入权限 (.env)",
        "required": "可写",
        "current": "✓ 可写" if root_writable else "✗ 不可写",
        "pass": root_writable,
    })

    all_pass = all(c["pass"] for c in checks if c.get("blocks_wizard", True))
    payload: dict = {"checks": checks, "all_pass": all_pass}
    defaults = _docker_wizard_defaults()
    if defaults:
        payload["docker_defaults"] = defaults
    return payload


@router.get("/connection-check")
async def connection_check():
    """Step 4: Verify MySQL + Redis using .env written in step 3."""
    _reject_if_locked()
    if not _has_env():
        raise HTTPException(400, "请先完成步骤3：数据库初始化")

    env = _parse_dotenv(ENV_FILE)
    db_url = env.get("NEXUS_DATABASE_URL", "")
    redis_url = env.get("NEXUS_REDIS_URL", "")

    mysql_ok, mysql_msg = await _test_mysql_url(db_url)
    redis_ok, redis_msg = _test_redis_url(redis_url)
    checks = [
        {"name": "MySQL", "required": "可连接", "current": mysql_msg, "pass": mysql_ok},
        {"name": "Redis", "required": "可连接", "current": redis_msg, "pass": redis_ok},
    ]
    return {"checks": checks, "all_pass": mysql_ok and redis_ok}


@router.post("/init-db")
async def init_db(req: InitDbRequest):
    """Step 3: Connect to MySQL, create tables, write configs."""
    _reject_if_locked()
    _reject_if_env_exists()

    db_url = _make_db_url(req)
    redis_url = _build_redis_url(req)
    site_url = req.site_url.rstrip("/") if req.site_url else f"http://127.0.0.1:{req.api_port}"
    db_port = int(req.db_port)
    redis_port = int(req.redis_port)

    if not _service_tcp_reachable(req.db_host, db_port):
        detail = _mysql_connect_error_detail(req.db_host, db_port)
        raise HTTPException(
            400,
            detail or f"无法连接 MySQL 服务器 {req.db_host}:{db_port}（TCP 不通，请确认服务已启动）",
        )

    if not _service_tcp_reachable(req.redis_host, redis_port):
        detail = _redis_connect_error_detail(req.redis_host, redis_port)
        raise HTTPException(
            400,
            detail or f"无法连接 Redis 服务器 {req.redis_host}:{redis_port}（TCP 不通，请确认服务已启动）",
        )

    try:
        patch_aiomysql_do_ping()
        engine = create_async_engine(db_url, pool_pre_ping=True, pool_recycle=300)
    except Exception as e:
        raise HTTPException(400, f"数据库引擎创建失败: {e}") from e

    try:
        async with engine.begin() as conn:
            # Create tables using SQLAlchemy metadata
            from server.domain.models import Base
            await conn.run_sync(Base.metadata.create_all)

            # Create additional indexes
            indexes = [
                "ALTER TABLE servers ADD INDEX idx_servers_is_online (is_online)",
                "ALTER TABLE servers ADD INDEX idx_servers_category (category)",
                "ALTER TABLE servers ADD INDEX idx_servers_platform_id (platform_id)",
                "ALTER TABLE servers ADD INDEX idx_servers_node_id (node_id)",
                "ALTER TABLE sync_logs ADD INDEX idx_sync_logs_srv_start (server_id, started_at)",
                "ALTER TABLE login_attempts ADD INDEX idx_login_attempts_username_ip (username, ip_address)",
                "ALTER TABLE ssh_sessions ADD INDEX idx_ssh_sessions_server_id (server_id)",
                "ALTER TABLE ssh_sessions ADD INDEX idx_ssh_sessions_admin_id (admin_id)",
                "ALTER TABLE command_logs ADD INDEX idx_cmdlog_srv_time (server_id, created_at)",
                "ALTER TABLE command_logs ADD INDEX idx_cmdlog_session_id (session_id)",
            ]
            for idx_sql in indexes:
                try:
                    await conn.execute(text(idx_sql))
                except Exception:  # noqa: S110 — DDL idempotent, may already exist
                    pass  # Index may already exist

            # Schema migrations for existing databases (safe — no-op if column exists)
            migrations = [
                "ALTER TABLE push_schedules ADD COLUMN sync_mode VARCHAR(20) DEFAULT 'incremental' COMMENT '同步模式'",
            ]
            for mig_sql in migrations:
                try:
                    await conn.execute(text(mig_sql))
                except Exception:  # noqa: S110 — DDL idempotent, may already exist
                    pass  # Column may already exist

            # Calculate pool_size from max_connections
            try:
                result = await conn.execute(text("SHOW VARIABLES LIKE 'max_connections'"))
                row = result.fetchone()
                max_conn = max(100, int(row[1]) if row else 100)
                pool_size = max(20, int(max_conn * 0.4))
                max_overflow = max(20, int(max_conn * 0.3))
            except Exception:
                pool_size, max_overflow = 160, 120

            # Keys: reuse Docker install pre-seed, else generate (must match install wizard)
            preseed = _install_keys_from_environment()
            if preseed:
                secret_key, api_key, encryption_key = preseed
                logger.info("Using pre-generated NEXUS keys from install environment")
            else:
                secret_key = secrets.token_hex(32)
                api_key = secrets.token_hex(16)
                import base64

                encryption_key = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode()

            # Insert settings
            settings_kv = {
                "api_key": api_key,
                "secret_key": secret_key,
                "api_base_url": site_url,
                "system_name": "Nexus",
                "system_title": "Nexus — 服务器运维管理平台",
                "db_pool_size": str(pool_size),
                "db_max_overflow": str(max_overflow),
                "redis_url": redis_url,
                "heartbeat_timeout": "60",
                "cpu_alert_threshold": "80",
                "mem_alert_threshold": "80",
                "disk_alert_threshold": "80",
                "telegram_bot_token": "",
                "telegram_chat_id": "",
            }
            for k, v in settings_kv.items():
                await conn.execute(
                    text("INSERT INTO `settings` (`key`, `value`) VALUES (:k, :v) "
                         "ON DUPLICATE KEY UPDATE `value` = :v2"),
                    {"k": k, "v": v, "v2": v},
                )

    except Exception as e:
        await engine.dispose()
        err = str(e)
        if "1045" in err or "Access denied" in err:
            raise HTTPException(400, _mysql_auth_error_detail(req.db_user)) from e
        if "2003" in err or "Can't connect to MySQL" in err:
            extra = _mysql_connect_error_detail(req.db_host, db_port)
            if extra:
                raise HTTPException(400, extra) from e
        raise HTTPException(400, f"数据库初始化失败: {e}") from e

    await engine.dispose()

    # Write .env
    _write_env(req, secret_key, api_key, encryption_key, pool_size, max_overflow, redis_url, site_url)

    # Write config.json
    _write_config_json(req, api_key, site_url)

    # Configure process guardian (best-effort)
    install_dir = str(ROOT_DIR)
    guardian_results = _configure_guardian(install_dir, req.api_port)

    # Save install state for step 4
    import json
    install_token = secrets.token_urlsafe(32)
    state = {
        "db_host": req.db_host,
        "db_port": req.db_port,
        "db_name": req.db_name,
        "db_user": req.db_user,
        # P2-19: Don't store db_pass in state file — only needed for _make_db_url()
        # which is called with the full CreateAdminRequest in step 4
        "install_token": install_token,
        "pool_size": pool_size,
        "max_overflow": max_overflow,
        "install_dir": install_dir,
        "site_url": site_url,
        "api_port": req.api_port,
        "guardian_results": guardian_results,
        "step": 4,
    }
    write_utf8_lf(STATE_FILE, json.dumps(state, ensure_ascii=False) + "\n")

    return {
        "success": True,
        "install_token": install_token,
        "pool_size": pool_size,
        "max_overflow": max_overflow,
        "tables_created": 14,
        "guardian_results": guardian_results,
        "bt_panel": _detect_bt_panel(),
    }


@router.post("/create-admin")
async def create_admin(req: CreateAdminRequest):
    """Step 4: Create admin account and set brand name."""
    _reject_if_locked()
    if not _has_env():
        raise HTTPException(400, "请先完成步骤3：数据库初始化")

    _verify_install_token(req.install_token)

    if len(req.admin_password) < 6:
        raise HTTPException(400, "密码长度至少6位")

    db_url = _make_db_url(req)

    try:
        patch_aiomysql_do_ping()
        engine = create_async_engine(db_url, pool_pre_ping=True, pool_recycle=300)
        async with engine.begin() as conn:
            # Hash password with bcrypt
            import bcrypt
            password_hash = bcrypt.hashpw(
                req.admin_password.encode("utf-8"),
                bcrypt.gensalt(),
            ).decode("utf-8")

            # Insert admin
            await conn.execute(
                text(
                    "INSERT INTO admins (username, password_hash, email, is_active) "
                    "VALUES (:u, :h, :e, 1) "
                    "ON DUPLICATE KEY UPDATE password_hash = :h2, email = :e2, is_active = 1"
                ),
                {
                    "u": req.admin_username,
                    "h": password_hash,
                    "e": req.admin_email or None,
                    "h2": password_hash,
                    "e2": req.admin_email or None,
                },
            )

            # Update brand in settings
            await conn.execute(
                text("INSERT INTO `settings` (`key`, `value`) VALUES (:k, :v) "
                     "ON DUPLICATE KEY UPDATE `value` = :v2"),
                {"k": "system_name", "v": req.system_name, "v2": req.system_name},
            )
            await conn.execute(
                text("INSERT INTO `settings` (`key`, `value`) VALUES (:k, :v) "
                     "ON DUPLICATE KEY UPDATE `value` = :v2"),
                {"k": "system_title", "v": req.system_title, "v2": req.system_title},
            )

        await engine.dispose()
    except Exception as e:
        raise HTTPException(400, f"创建管理员失败: {e}") from e

    # Update config.json with brand
    if CONFIG_JSON.exists():
        import json
        try:
            config = json.loads(read_utf8_text(CONFIG_JSON))
            config["app_name"] = req.system_name
            write_utf8_lf(CONFIG_JSON, json.dumps(config, indent=2, ensure_ascii=False) + "\n")
        except Exception:  # noqa: S110 — best-effort cleanup
            pass

    # Update .env with brand
    if ENV_FILE.exists():
        import re
        content = read_utf8_text(ENV_FILE)
        content = re.sub(
            r"NEXUS_SYSTEM_NAME=.*",
            f"NEXUS_SYSTEM_NAME={_escape_env_value(req.system_name)}",
            content,
        )
        content = re.sub(
            r"NEXUS_SYSTEM_TITLE=.*",
            f"NEXUS_SYSTEM_TITLE={_escape_env_value(req.system_title)}",
            content,
        )
        write_utf8_lf(ENV_FILE, content)

    # Update state
    if STATE_FILE.exists():
        import json
        state = json.loads(read_utf8_text(STATE_FILE))
        state["step"] = 4
        state["system_name"] = req.system_name
        write_utf8_lf(STATE_FILE, json.dumps(state, ensure_ascii=False) + "\n")

    # Atomically lock after admin creation (closes unauthenticated install API window)
    _finalize_install_lock()
    return {"success": True, "locked": True}


@router.post("/lock")
async def lock_install():
    """Step 5: Lock the installer — prevent re-installation.

    Requires that an admin account exists (step 4 completed) before locking.
    This prevents unauthenticated callers from locking the installer
    before the legitimate admin has finished setup.
    """
    if _is_locked():
        return {"success": True, "already_locked": True}
    if not _has_env():
        raise HTTPException(400, "请先完成步骤3：数据库初始化")

    # Verify at least one active admin account exists before allowing lock
    try:
        from server.infrastructure.database.session import AsyncSessionLocal
        from sqlalchemy import text as _text
        async with AsyncSessionLocal() as session:
            result = await session.execute(
                _text("SELECT COUNT(*) FROM admins WHERE is_active = 1")
            )
            count = result.scalar() or 0
            if count == 0:
                raise HTTPException(
                    status_code=400,
                    detail="无法锁定：尚无活跃管理员账户。请先完成安装步骤4。",
                )
    except HTTPException:
        raise
    except Exception as e:
        logger.warning("Failed to verify admin account before locking: %s", e)
        # If DB is temporarily unreachable, conservatively reject lock
        raise HTTPException(
            status_code=503,
            detail="数据库暂时不可用，无法验证管理员账户，请稍后重试。",
        ) from e

    _finalize_install_lock()
    return {"success": True}


@router.get("/state")
async def get_install_state():
    """Get current install state (for step navigation)."""
    _reject_post_install_wizard()
    import json

    if not STATE_FILE.exists():
        return {"step": 1, "installed": _is_installed(), "locked": _is_locked()}

    try:
        state = json.loads(read_utf8_text(STATE_FILE))
        state["installed"] = _is_installed()
        state["locked"] = _is_locked()
        return state
    except Exception:
        return {"step": 1, "installed": _is_installed(), "locked": _is_locked()}