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, unquote_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"
INSTALL_HTML = ROOT_DIR / "web" / "app" / "install.html"
INSTALL_HTML_BAK = ROOT_DIR / "web" / "app" / "install.html.bak"
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_user: str = ""
    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:
    """Build redis:// URL. Password-only auth must use redis://:pass@host (leading colon)."""
    user = (req.redis_user or "").strip()
    password = (req.redis_pass or "").strip()
    auth = ""
    if user and password:
        auth = f"{quote_plus(user)}:{quote_plus(password)}@"
    elif password:
        auth = f":{quote_plus(password)}@"
    elif user:
        auth = f"{quote_plus(user)}@"
    return f"redis://{auth}{req.redis_host}:{req.redis_port}/{req.redis_db}"


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_auth_error_detail() -> str:
    if os.environ.get("NEXUS_DOCKER_WRITE_ENV") == "1":
        return (
            "Redis 认证失败。1Panel：主机 1Panel-redis-xxxx；密码为应用参数；"
            "用户名通常留空（自动尝试 :pass@）或填 default。勿填 root（多数实例不支持）。"
            "诊断: bash deploy/test-1panel-redis.sh"
        )
    return "Redis 密码错误或需要 AUTH；请核对用户名/密码（无密码则留空）。"


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
        lowered = err.lower()
        if (
            "invalid username-password" in lowered
            or "wrongpass" in lowered
            or "noauth" in lowered
            or "authentication required" in lowered
        ):
            return False, _redis_auth_error_detail()
        return False, f"✗ Redis 连接失败: {e}"


def _is_docker_install_mode() -> bool:
    return os.environ.get("NEXUS_DOCKER_WRITE_ENV") == "1"


def _sanitize_redis_request(req: InitDbRequest) -> InitDbRequest:
    """1Panel/Docker: Redis has no account — ignore username (incl. mistaken root)."""
    user = (req.redis_user or "").strip()
    if _is_docker_install_mode():
        if user:
            return req.model_copy(update={"redis_user": ""})
        return req
    if user.lower() == "root":
        return req.model_copy(update={"redis_user": ""})
    return req


def _redis_url_candidates(req: InitDbRequest) -> list[tuple[str, str]]:
    """Auth variants: 1Panel tries :pass@ then default only (no root)."""
    password = (req.redis_pass or "").strip()
    if not password:
        return [("无密码", _build_redis_url(req))]

    seen: set[str] = set()
    out: list[tuple[str, str]] = []

    def add(label: str, candidate: InitDbRequest) -> None:
        url = _build_redis_url(candidate)
        if url in seen:
            return
        seen.add(url)
        out.append((label, url))

    if _is_docker_install_mode():
        add("仅密码 (:pass@)", req.model_copy(update={"redis_user": ""}))
        add("用户 default", req.model_copy(update={"redis_user": "default"}))
        return out

    user = (req.redis_user or "").strip()
    if user and user.lower() != "root":
        add(f"用户 {user}", req)
    add("仅密码 (:pass@)", req.model_copy(update={"redis_user": ""}))
    if user.lower() != "default":
        add("用户 default", req.model_copy(update={"redis_user": "default"}))
    return out


def _init_db_request_from_redis_url(redis_url: str) -> InitDbRequest | None:
    """Parse NEXUS_REDIS_URL into InitDbRequest fields for re-probe / repair."""
    if not redis_url.strip():
        return None
    parsed = urlparse(redis_url)
    if not parsed.hostname:
        return None
    user = unquote_plus(parsed.username or "")
    password = unquote_plus(parsed.password or "")
    if not password and user:
        # Legacy bug: redis://pass@host (password stored as username)
        password = user
        user = ""
    db = (parsed.path or "/0").lstrip("/") or "0"
    return InitDbRequest(
        db_pass=".",
        redis_host=parsed.hostname,
        redis_port=str(parsed.port or 6379),
        redis_db=db,
        redis_user=user,
        redis_pass=password,
    )


def _redis_repair_request(redis_url: str) -> InitDbRequest | None:
    """Build InitDbRequest for connection-check repair (URL + install state password)."""
    req = _init_db_request_from_redis_url(redis_url)
    if not req:
        return None
    if (req.redis_pass or "").strip():
        return req
    if not STATE_FILE.is_file():
        return req
    import json

    try:
        state = json.loads(read_utf8_text(STATE_FILE))
    except (json.JSONDecodeError, OSError):
        return req
    saved_pass = (state.get("redis_pass") or "").strip()
    if saved_pass:
        return req.model_copy(update={"redis_pass": saved_pass})
    return req


def _mask_redis_url(redis_url: str) -> str:
    parsed = urlparse(redis_url)
    if not parsed.password:
        return redis_url
    host = parsed.hostname or ""
    port = parsed.port
    user = parsed.username or ""
    if user:
        netloc = f"{user}:***@{host}"
    else:
        netloc = f":***@{host}"
    if port:
        netloc = f"{netloc}:{port}"
    path = parsed.path or "/0"
    return f"redis://{netloc}{path}"


def _patch_env_redis_url(new_url: str) -> None:
    """Update only NEXUS_REDIS_URL in existing .env (install step 4 repair)."""
    if not ENV_FILE.is_file():
        return
    lines = read_utf8_text(ENV_FILE).splitlines()
    new_line = f"NEXUS_REDIS_URL={_escape_env_value(new_url)}"
    replaced = False
    out: list[str] = []
    for line in lines:
        if line.strip().startswith("NEXUS_REDIS_URL="):
            out.append(new_line)
            replaced = True
        else:
            out.append(line)
    if not replaced:
        out.append(new_line)
    write_utf8_lf(ENV_FILE, "\n".join(out) + "\n")
    logger.info("Patched NEXUS_REDIS_URL in %s", ENV_FILE)
    _sync_env_to_persist_volume()


def _resolve_redis_url(req: InitDbRequest) -> tuple[bool, str, str | None]:
    """Pick first working Redis URL among 1Panel-compatible auth variants."""
    last_msg = "✗ 未配置 Redis"
    for label, url in _redis_url_candidates(req):
        ok, msg = _test_redis_url(url)
        last_msg = msg
        if ok:
            return True, f"✓ Redis 连接正常（{label}）", url
    return False, last_msg, None


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}")
    _sync_env_to_persist_volume()


def _persist_dir() -> Path:
    raw = os.environ.get("NEXUS_PERSIST_DIR", "/var/lib/nexus").strip() or "/var/lib/nexus"
    return Path(raw)


def _persist_env_path() -> Path:
    return _persist_dir() / ".env"


def _persist_lock_path() -> Path:
    return _persist_dir() / ".install_locked"


def _sync_env_to_persist_volume() -> None:
    """Copy /app/.env to nexus-state volume immediately (Docker install wizard)."""
    if not _is_docker_install_mode() or not ENV_FILE.is_file():
        return
    dest = _persist_env_path()
    try:
        dest.parent.mkdir(parents=True, exist_ok=True)
        import shutil

        shutil.copy2(ENV_FILE, dest)
        dest.chmod(0o600)
        logger.info("Synced .env to persist volume %s", dest)
    except OSError as e:
        logger.warning("Failed to sync .env to persist volume: %s", e)


def _sync_lock_to_persist_volume() -> None:
    """Copy /app/.install_locked to nexus-state volume (survives container recreate)."""
    if not INSTALL_LOCK.is_file():
        return
    dest = _persist_lock_path()
    try:
        dest.parent.mkdir(parents=True, exist_ok=True)
        import shutil

        shutil.copy2(INSTALL_LOCK, dest)
        dest.chmod(0o600)
        logger.info("Synced install lock to persist volume %s", dest)
    except OSError as e:
        logger.warning("Failed to sync install lock to persist volume: %s", e)


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 _host_deploy_root(install_dir: str) -> str:
    """宿主机仓库路径（容器内 /app 与宿主机 /opt/nexus 不同）。"""
    host = os.environ.get("NEXUS_HOST_ROOT", "").strip()
    if host:
        return host
    if install_dir != "/app" and Path(install_dir).is_dir():
        return install_dir
    return "/opt/nexus"


def _append_ops_cron_results(results: list[str], install_dir: str) -> None:
    """Crontab 条目与 deploy/install_ops_cron.sh 保持一致。"""
    health_marker = "nexus-health-monitor"
    backup_marker = "nexus-mysql-backup"
    health_line = (
        f"* * * * * NEXUS_DEPLOY_DIR={install_dir} "
        f"{install_dir}/deploy/health_monitor.sh "
        f">> /var/log/nexus_health.log 2>&1 # {health_marker}"
    )
    backup_line = (
        f"0 3 * * * NEXUS_ROOT={install_dir} "
        f"{install_dir}/deploy/db_backup.sh "
        f">> /var/log/nexus_backup.log 2>&1 # {backup_marker}"
    )
    try:
        current = subprocess.run(
            ["crontab", "-l"], capture_output=True, text=True, timeout=5
        ).stdout or ""
        added: list[str] = []
        if health_marker not in current and "health_monitor.sh" not in current:
            added.append(health_line)
        if backup_marker not in current and "db_backup.sh" not in current:
            added.append(backup_line)
        if added:
            new_cron = current.rstrip() + "\n" + "\n".join(added) + "\n"
            proc = subprocess.run(
                ["crontab", "-"], input=new_cron, capture_output=True, text=True, timeout=5
            )
            if proc.returncode == 0:
                results.append("✓ Crontab 已配置（health_monitor + db_backup）")
            else:
                results.append("✗ Crontab 配置失败 — 请手动: bash deploy/install_ops_cron.sh")
        else:
            results.append("✓ Crontab 已存在 health_monitor / db_backup 条目")
    except Exception:
        results.append(
            f"⚠ 宿主机 cron 需在安装完成后执行: "
            f"sudo bash {install_dir}/deploy/install_ops_cron.sh"
        )


def _configure_docker_guardian(install_dir: str, api_port: str) -> list[str]:
    """1Panel + Docker: Compose restart/healthcheck + in-app self_monitor (no host Supervisor)."""
    results: list[str] = []
    host_root = _host_deploy_root(install_dir)

    if Path("/.dockerenv").is_file():
        results.append("✓ Layer 1: Docker 容器运行中（Compose restart: unless-stopped）")
    else:
        results.append("⚠ Layer 1: 未检测到 /.dockerenv — 请确认使用 docker-compose.prod.yml 部署")

    results.append("✓ Layer 2: Python self_monitor 随正常模式启动（每 30s 自检 Redis/MySQL/WebSocket）")

    try:
        import urllib.request

        with urllib.request.urlopen(
            f"http://127.0.0.1:{api_port}/health", timeout=3.0
        ) as resp:
            body = resp.read().decode().strip()
        if body == "ok":
            results.append(f"✓ 容器内 /health 正常（127.0.0.1:{api_port}）")
        else:
            results.append(f"⚠ /health 响应异常: {body[:80]}")
    except Exception as e:
        results.append(f"⚠ 容器内 /health 暂不可达: {e}")

    db_host = _onepanel_service_host("DB")
    redis_host = _onepanel_service_host("REDIS")
    if db_host and redis_host:
        results.append(f"✓ 1panel-network 容器名: MySQL={db_host}, Redis={redis_host}")
    elif db_host or redis_host:
        results.append(
            f"⚠ 1Panel 主机名不完整（DB={db_host or '未设置'}, Redis={redis_host or '未设置'}）"
        )
    else:
        results.append("ℹ 未配置 NEXUS_1PANEL_*_HOST — 宿主机执行 nx update 可自动探测")

    site_url = os.environ.get("NEXUS_API_BASE_URL", "").strip()
    publish_port = os.environ.get("NEXUS_PUBLISH_PORT", api_port).strip() or api_port
    if site_url:
        results.append(
            f"ℹ Layer 3（宿主机）: 1Panel OpenResty 反代 {site_url} → 127.0.0.1:{publish_port}"
        )
    else:
        results.append(
            f"ℹ Layer 3（宿主机）: 配置 OpenResty 反代 → 127.0.0.1:{publish_port}（见 deploy/1panel/openresty-nexus.conf.example）"
        )

    results.append(
        f"ℹ Layer 3（宿主机 cron）: sudo bash {host_root}/deploy/install_ops_cron.sh"
    )
    results.append("ℹ 或: sudo nx cron（health_monitor 每分钟 + MySQL 备份每日 03:00）")
    results.append(f"ℹ 日常升级: 宿主机 cd {host_root} && nx update")
    return results


def _configure_guardian(install_dir: str, api_port: str) -> list[str]:
    """Best-effort process guardian: Docker/1Panel vs bare-metal Supervisor stack."""
    if _is_docker_install_mode():
        return _configure_docker_guardian(install_dir, api_port)

    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. health_monitor 默认路径（cron 行内 NEXUS_DEPLOY_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'DEPLOY_DIR="\$\{NEXUS_DEPLOY_DIR:-[^"]*\}"',
                f'DEPLOY_DIR="${{NEXUS_DEPLOY_DIR:-{install_dir}}}"',
                content,
            )
            write_utf8_lf(health_sh, content)
            health_sh.chmod(0o755)
            results.append("✓ health_monitor.sh DEPLOY_DIR 默认值已更新")
        except OSError:
            results.append("✗ health_monitor.sh 更新失败")
    else:
        results.append("⚠ health_monitor.sh 未找到，跳过")

    # 4. Crontab（与 install_ops_cron.sh 同格式）
    _append_ops_cron_results(results, install_dir)

    # 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


_INSTALL_STATE_PUBLIC_KEYS = frozenset({
    "step",
    "system_name",
    "db_host",
    "db_port",
    "db_name",
    "db_user",
    "redis_host",
    "redis_port",
    "redis_db",
    "site_url",
    "api_port",
    "pool_size",
    "max_overflow",
    "install_dir",
    "host_deploy_root",
    "docker_mode",
    "guardian_results",
    "installed",
    "locked",
    "has_token",
})


def _public_install_state(state: dict) -> dict:
    """Return install state safe for GET /state (no install_token or redis_pass)."""
    out = {k: state[k] for k in _INSTALL_STATE_PUBLIC_KEYS if k in state}
    if state.get("install_token"):
        out["has_token"] = True
    return out


def _clear_state_redis_pass() -> None:
    """Remove redis_pass from install state after connection-check repair wrote .env."""
    if not STATE_FILE.is_file():
        return
    import json

    try:
        state = json.loads(read_utf8_text(STATE_FILE))
    except (json.JSONDecodeError, OSError):
        return
    if not state.get("redis_pass"):
        return
    state["redis_pass"] = ""
    write_utf8_lf(STATE_FILE, json.dumps(state, ensure_ascii=False) + "\n")


def _archive_install_wizard_html() -> None:
    """Rename install.html → install.html.bak so the wizard is not served after lock."""
    if INSTALL_HTML_BAK.is_file():
        if INSTALL_HTML.is_file():
            try:
                INSTALL_HTML.unlink()
            except OSError as e:
                logger.warning("Failed to remove install.html after archive: %s", e)
        return
    if not INSTALL_HTML.is_file():
        return
    try:
        INSTALL_HTML.rename(INSTALL_HTML_BAK)
        logger.info("Archived install wizard: %s → %s", INSTALL_HTML, INSTALL_HTML_BAK)
    except OSError as e:
        logger.warning("Failed to archive install.html: %s", e)


def _finalize_install_lock() -> None:
    """Write install lock marker and remove wizard state file."""
    _sync_env_to_persist_volume()
    write_utf8_lf(
        INSTALL_LOCK,
        f"Nexus installation locked at {datetime.now(timezone.utc).isoformat()}\n",
    )
    _sync_lock_to_persist_volume()
    _archive_install_wizard_html()
    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


async def _credential_checks(req: InitDbRequest) -> tuple[list[dict], bool, str | None]:
    """Verify MySQL + Redis with credentials from install step 3 (TCP + AUTH)."""
    req = _sanitize_redis_request(req)
    db_url = _make_db_url(req)
    mysql_ok, mysql_msg = await _test_mysql_url(db_url)
    redis_ok, redis_msg, redis_url = _resolve_redis_url(req)
    checks = [
        {"name": "MySQL", "required": "可连接", "current": mysql_msg, "pass": mysql_ok},
        {"name": "Redis", "required": "可连接", "current": redis_msg, "pass": redis_ok},
    ]
    return checks, mysql_ok and redis_ok, redis_url if redis_ok else None


@router.post("/test-credentials")
async def test_credentials(req: InitDbRequest):
    """Step 3: Test MySQL/Redis account password before init-db writes .env."""
    _reject_if_locked()
    if _has_env():
        raise HTTPException(
            400,
            "系统已初始化，无法重复检测。请继续步骤 4 创建管理员，或修正 .env 后使用连接检测。",
        )
    checks, all_pass, redis_url = await _credential_checks(req)
    payload: dict = {"checks": checks, "all_pass": all_pass}
    if redis_url:
        payload["resolved_redis_url"] = _mask_redis_url(redis_url)
    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)
    redis_repaired = False
    resolved_display: str | None = None

    if not redis_ok and redis_url:
        req_from_env = _redis_repair_request(redis_url)
        if req_from_env:
            ok, msg, resolved = _resolve_redis_url(_sanitize_redis_request(req_from_env))
            if ok and resolved:
                if resolved != redis_url and not _is_locked():
                    _patch_env_redis_url(resolved)
                    redis_repaired = True
                    _clear_state_redis_pass()
                redis_ok, redis_msg = True, msg
                resolved_display = _mask_redis_url(resolved)

    checks = [
        {"name": "MySQL", "required": "可连接", "current": mysql_msg, "pass": mysql_ok},
        {"name": "Redis", "required": "可连接", "current": redis_msg, "pass": redis_ok},
    ]
    payload: dict = {"checks": checks, "all_pass": mysql_ok and redis_ok}
    if redis_repaired:
        payload["redis_repaired"] = True
    if resolved_display:
        payload["resolved_redis_url"] = resolved_display
    return payload


@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)
    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)

    checks, cred_ok, redis_url = await _credential_checks(req)
    if not cred_ok:
        failed = next((c for c in checks if not c["pass"]), checks[0])
        raise HTTPException(
            400,
            failed.get("current") or "MySQL 或 Redis 连接未通过，请先检测账号密码",
        )
    if not redis_url:
        redis_url = _build_redis_url(req)

    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)
    host_deploy_root = _host_deploy_root(install_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,
        "host_deploy_root": host_deploy_root,
        "site_url": site_url,
        "api_port": req.api_port,
        "redis_host": req.redis_host,
        "redis_port": req.redis_port,
        "redis_db": req.redis_db,
        # Step 4 connection-check repair only; removed when install locks
        "redis_pass": req.redis_pass,
        "guardian_results": guardian_results,
        "docker_mode": _is_docker_install_mode(),
        "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,
        "host_deploy_root": host_deploy_root,
        "bt_panel": _detect_bt_panel(),
        "docker_mode": _is_docker_install_mode(),
    }


@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 _public_install_state(state)
    except Exception:
        return {"step": 1, "installed": _is_installed(), "locked": _is_locked()}