Nexus/server/api/servers.py

"""Nexus — Servers API Routes (CRUD + heartbeat + health check + push)
Presentation layer — receives HTTP requests, delegates to ServerService/SyncService.

Live server status comes from Redis (real-time), base info from MySQL.
All write operations require JWT authentication and produce audit logs.
"""

import asyncio
import json
import logging
from typing import Optional

import bcrypt
from fastapi import APIRouter, Depends, HTTPException, Query, Request

from server.api.dependencies import get_server_service, get_sync_service, get_db
from server.api.auth_jwt import get_current_admin
from server.api.schemas import (
    ServerCreate, ServerUpdate, ServerCheck, ApiKeyRevealRequest,
    ServerImportResult, BatchAgentAction, AddByIpRequest, AddByIpsBatchRequest, AddByIpsBatchResult, AddByIpsBatchItemResult,
    BatchCategoryUpdate, BatchJobStarted, BatchJobStatus,
)
from server.application.server_batch_common import (
    install_error_msg as _install_error_msg,
    sudo_wrap as _sudo_wrap,
)
from server.application.services.server_batch_service import start_batch_job, get_batch_job, list_batch_jobs
from server.application.services.server_service import ServerService
from server.application.services.sync_service import SyncService
from server.domain.models import Server, Admin, AuditLog, PendingServer
from server.config import settings
from server.infrastructure.redis.client import get_redis
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
import shlex



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

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

REDIS_KEY_PREFIX = "heartbeat:"


def _server_has_agent_monitoring(server: Server) -> bool:
    """True when this server is expected to run Nexus Agent (key or version on file)."""
    return bool((server.agent_api_key or "").strip() or (server.agent_version or "").strip())

# ── CRUD ──

@router.get("/", response_model=dict)
async def list_servers(
    category: Optional[str] = Query(None, description="Filter by category"),
    platform_id: Optional[int] = Query(None, description="Filter by platform ID"),
    node_id: Optional[int] = Query(None, description="Filter by node ID"),
    search: Optional[str] = Query(None, description="Search by name or domain (substring)"),
    sort_by: Optional[str] = Query(None, description="Sort field: name/domain/status/heartbeat/agent_version/category/id"),
    sort_order: Optional[str] = Query("asc", description="Sort direction: asc or desc"),
    is_online: Optional[bool] = Query(None, description="Filter by online status"),
    target_path_unset: Optional[bool] = Query(
        None, description="Filter unset target paths: true=only unset, false=exclude unset",
    ),
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(50, ge=1, le=200, description="Items per page"),
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """List servers with live status from Redis.

    Static columns sort in MySQL; status/heartbeat/agent_version sort after
    Redis overlay so order matches displayed values.
    """
    from server.application.services.server_list_sort import (
        LIVE_SORT_FIELDS,
        MAX_LIVE_SORT_ROWS,
        sort_server_items,
    )
    from server.infrastructure.database.server_repo import ServerRepositoryImpl

    offset = (page - 1) * per_page
    repo = ServerRepositoryImpl(db)
    redis = get_redis()
    order = sort_order if sort_order in ("asc", "desc") else "asc"

    if sort_by in LIVE_SORT_FIELDS:
        db_online_filter = None if is_online is not None else is_online
        all_servers, db_total = await repo.get_filtered(
            category,
            platform_id,
            node_id,
            search=search,
            is_online=db_online_filter,
            target_path_unset=target_path_unset,
            max_rows=MAX_LIVE_SORT_ROWS,
        )
        if db_total > MAX_LIVE_SORT_ROWS:
            logger.warning(
                "Live sort capped at %s servers (total matching=%s)",
                MAX_LIVE_SORT_ROWS,
                db_total,
            )
        result = await _build_server_list_with_overlay(redis, all_servers)
        if is_online is not None:
            result = [s for s in result if s.get("is_online") == is_online]
        total = len(result)
        result = sort_server_items(result, sort_by, order)
        result = result[offset : offset + per_page]
    else:
        page_servers, total = await repo.get_paginated(
            category,
            platform_id,
            node_id,
            offset,
            per_page,
            search=search,
            sort_by=sort_by,
            sort_order=order,
            is_online=is_online,
            target_path_unset=target_path_unset,
        )
        result = await _build_server_list_with_overlay(redis, page_servers)
        if is_online is not None:
            result = [s for s in result if s.get("is_online") == is_online]
            total = len(result)
            result = result[:per_page]

    return {
        "items": result,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": (total + per_page - 1) // per_page if total > 0 else 1,
    }


# ── Dashboard Stats (must be before /{id} to avoid path collision) ──

@router.get("/stats", response_model=dict)
async def server_stats(
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Aggregated server stats for dashboard — SQL aggregation + Redis alert count"""
    from sqlalchemy import func

    # Total + category breakdown in a single query
    result = await db.execute(
        select(Server.category, func.count(Server.id))
        .group_by(Server.category)
    )
    categories = {}
    total = 0
    for cat, cnt in result.all():
        key = cat or "uncategorized"
        categories[key] = cnt
        total += cnt

    # Online count from Redis heartbeats (real-time); MySQL is stale between flushes
    redis = get_redis()
    online = 0
    try:
        cursor = 0
        while True:
            cursor, keys = await redis.scan(cursor, match="heartbeat:*", count=200)
            for key in keys:
                data = await redis.hgetall(key)
                if data.get("is_online") == "True":
                    online += 1
            if cursor == 0:
                break
    except Exception as e:
        logger.warning(f"Failed to count Redis heartbeats in server_stats: {e}")
        result = await db.execute(
            select(func.count(Server.id)).where(Server.is_online == True)
        )
        online = int(result.scalar() or 0)

    # Stale heartbeat:* keys (deleted servers) must not inflate online above total.
    if total > 0 and online > total:
        logger.debug(
            "server_stats: capping online %s -> %s (heartbeat keys > DB total)",
            online,
            total,
        )
        online = total

    # Offline = Agent 应监控但未在 Redis 上报；未配置 Agent 的计为 unknown（非离线）
    from sqlalchemy import or_, and_

    monitored_result = await db.execute(
        select(func.count(Server.id)).where(
            or_(
                and_(Server.agent_api_key.isnot(None), Server.agent_api_key != ""),
                and_(Server.agent_version.isnot(None), Server.agent_version != ""),
            )
        )
    )
    monitored = int(monitored_result.scalar() or 0)
    unknown = max(0, total - monitored)
    offline = max(0, monitored - online)

    alerts = 0
    try:
        cursor = 0
        while True:
            cursor, keys = await redis.scan(cursor, match="alerts:*", count=200)
            alerts += len(keys)
            if cursor == 0:
                break
    except Exception as e:
        logger.warning(f"Failed to count Redis alerts in server_stats: {e}")

    return {
        "total": total,
        "online": online,
        "offline": offline,
        "unknown": unknown,
        "alerts": alerts,
        "categories": categories,
    }


# ── Sync Logs ──

@router.get("/categories", response_model=dict)
async def server_categories(
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Return distinct categories with counts and platform list for server filtering."""
    from sqlalchemy import func as sqlfunc
    from server.domain.models import Platform

    # Categories with server counts
    cat_result = await db.execute(
        select(Server.category, sqlfunc.count(Server.id))
        .group_by(Server.category)
    )
    categories = []
    for cat, cnt in cat_result.all():
        categories.append({"name": cat or "uncategorized", "count": cnt})

    # Platforms
    plat_result = await db.execute(select(Platform).order_by(Platform.id))
    platforms = []
    for p in plat_result.scalars().all():
        platforms.append({"id": p.id, "name": p.name, "category": p.category or ""})

    return {"categories": categories, "platforms": platforms}


@router.get("/logs", response_model=dict)
async def get_all_sync_logs(
    server_id: Optional[int] = Query(None, description="Filter by server ID"),
    status: Optional[str] = Query(None, description="success|failed|running|cancelled"),
    sync_mode: Optional[str] = Query(None, description="incremental|full|overwrite|checksum|command"),
    trigger_type: Optional[str] = Query(None, description="manual|schedule|retry|batch"),
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(20, ge=1, le=100, description="Items per page"),
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Get paginated sync logs across all servers with optional filters."""
    import math
    from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
    repo = SyncLogRepositoryImpl(db)
    offset = (page - 1) * per_page
    rows, total = await repo.get_all_paginated(
        server_id=server_id, status=status, sync_mode=sync_mode,
        trigger_type=trigger_type, offset=offset, limit=per_page,
    )
    items = [_sync_log_to_dict(log, server_name=name) for log, name in rows]
    pages = math.ceil(total / per_page) if total > 0 else 1
    return {"items": items, "total": total, "page": page, "per_page": per_page, "pages": pages}


# ── CSV Batch Import ──

MAX_IMPORT_FILE_SIZE = 1_048_576  # 1 MB
MAX_IMPORT_ROWS = 500
MAX_BATCH_IP_LINES = 500

# CSV column mapping: CSV header (Chinese) → Server model field
CSV_COLUMNS = [
    ("名称", "name", True),
    ("地址", "domain", True),
    ("SSH端口", "port", False),
    ("用户名", "username", False),
    ("认证方式", "auth_method", False),
    ("密码", "password", False),
    ("密钥路径", "ssh_key_path", False),
    ("私钥内容", "ssh_key_private", False),
    ("分类", "category", False),
    ("目标路径", "target_path", False),
    ("备注", "description", False),
]

# Columns excluded from the template but still accepted on import (backward compat)
_CSV_IMPORT_ONLY_COLUMNS = [
    ("Agent端口", "agent_port", False),
    ("平台", "platform_name", False),
    ("节点", "node_name", False),
]


def _sanitize_csv_cell(value: str) -> str:
    """Sanitize a CSV cell to prevent CSV injection (Excel formula injection).

    Strips leading dangerous characters: = + - @ \t \r
    that could be interpreted as formulas by spreadsheet software.
    """
    if not value:
        return value
    # Remove leading whitespace and dangerous formula prefix characters
    stripped = value.lstrip()
    if stripped and stripped[0] in ("=", "+", "-", "@", "\t", "\r"):
        # Prepend a single quote to defang the formula (standard CSV injection defense)
        return "'" + value
    return value


@router.get("/meta/api_base_url")
async def get_api_base_url(admin: Admin = Depends(get_current_admin)):
    """Return the configured API_BASE_URL (from .env / settings).

    Used by the frontend to determine if Agent install commands can be generated.
    """
    return {"api_base_url": settings.API_BASE_URL or ""}


@router.get("/import/template")
async def download_import_template(
    admin: Admin = Depends(get_current_admin),
):
    """Download CSV import template with headers and one example row."""
    from fastapi.responses import StreamingResponse
    import io

    headers = ",".join(col[0] for col in CSV_COLUMNS)
    example = "web-01,192.168.1.10,22,root,password,your-password,,,商城,/www/wwwroot,Web服务器"
    csv_content = f"{headers}\n{example}\n"  # BOM for Excel Chinese compatibility

    return StreamingResponse(
        io.BytesIO(csv_content.encode("utf-8")),
        media_type="text/csv; charset=utf-8",
        headers={"Content-Disposition": "attachment; filename=servers_import_template.csv"},
    )


@router.post("/import", response_model=ServerImportResult)
async def import_servers(
    request: Request,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Import servers from a CSV file (multipart/form-data upload).

    Each row is processed independently — a failure in one row does not affect others.
    Duplicate domains (already in DB) are skipped, not counted as errors.
    """
    from server.infrastructure.database.crypto import encrypt_value
    from server.infrastructure.database.server_repo import ServerRepositoryImpl
    import csv
    import io
    import secrets

    # Read the uploaded file
    form = await request.form()
    file = form.get("file")
    if not file or not hasattr(file, "read"):
        raise HTTPException(status_code=400, detail="请上传 CSV 文件")

    content = await file.read()
    if len(content) > MAX_IMPORT_FILE_SIZE:
        raise HTTPException(status_code=400, detail=f"文件大小超过限制 ({MAX_IMPORT_FILE_SIZE // 1024}KB)")

    # Decode CSV
    try:
        text = content.decode("utf-8-sig")  # Handle BOM
    except UnicodeDecodeError:
        try:
            text = content.decode("gbk")  # Fallback for Chinese Excel
        except UnicodeDecodeError as enc_err:
            raise HTTPException(status_code=400, detail="CSV 编码不支持，请使用 UTF-8 或 GBK") from enc_err

    reader = csv.DictReader(io.StringIO(text))
    if not reader.fieldnames:
        raise HTTPException(status_code=400, detail="CSV 文件为空或缺少表头")

    # Validate headers — at least name and domain must exist
    header_set = set(reader.fieldnames)
    required_headers = {col[0] for col in CSV_COLUMNS if col[2]}
    missing = required_headers - header_set
    if missing:
        raise HTTPException(status_code=400, detail=f"CSV 缺少必填列: {', '.join(missing)}")

    # Build column mapping: CSV header → field name (template columns + backward-compat columns)
    col_map = {}
    all_columns = CSV_COLUMNS + _CSV_IMPORT_ONLY_COLUMNS
    for cn_name, field_name, _ in all_columns:
        if cn_name in header_set:
            col_map[cn_name] = field_name

    # Load existing domains for duplicate check
    server_repo = ServerRepositoryImpl(db)
    existing_servers = await server_repo.get_all()
    existing_domains = {s.domain for s in existing_servers}

    # Load platform/node lookup by name for mapping
    from server.domain.models import Platform, Node
    platform_map = {}  # name → id
    node_map = {}  # name → id
    try:
        result = await db.execute(select(Platform))
        for p in result.scalars().all():
            platform_map[p.name] = p.id
        result = await db.execute(select(Node))
        for n in result.scalars().all():
            node_map[n.name] = n.id
    except Exception as e:
        logger.warning(f"Platform/node lookup failed during CSV import: {e}")

    result = ServerImportResult()
    row_num = 1  # Header is row 0

    for row in reader:
        row_num += 1
        if row_num - 1 > MAX_IMPORT_ROWS:
            result.errors.append({"row": row_num, "name": "", "domain": "", "error": f"超过最大行数限制 ({MAX_IMPORT_ROWS})"})
            break

        # Map CSV columns to server data
        data = {}
        for cn_name, field_name, _ in all_columns:
            if cn_name in col_map:
                value = row.get(cn_name, "").strip()
                if value:
                    value = _sanitize_csv_cell(value)
                    data[field_name] = value

        # Validate required fields
        name = data.get("name", "").strip()
        domain = data.get("domain", "").strip()
        if not name or not domain:
            result.failed += 1
            result.errors.append({"row": row_num, "name": name, "domain": domain, "error": "名称和地址为必填项"})
            continue

        # Duplicate check
        if domain in existing_domains:
            result.skipped += 1
            continue

        # Parse numeric fields
        try:
            if "port" in data:
                data["port"] = int(data["port"])
            if "agent_port" in data:
                data["agent_port"] = int(data["agent_port"])
        except ValueError:
            result.failed += 1
            result.errors.append({"row": row_num, "name": name, "domain": domain, "error": "端口号必须为数字"})
            continue

        # Set defaults
        data.setdefault("port", 22)
        data.setdefault("username", "root")
        data.setdefault("auth_method", "password")
        data.setdefault("agent_port", 8601)

        # Validate auth_method
        if data["auth_method"] not in ("password", "key"):
            result.failed += 1
            result.errors.append({"row": row_num, "name": name, "domain": domain, "error": f"认证方式无效: {data['auth_method']}"})
            continue

        # Resolve platform/node by name
        if data.get("platform_name") and data["platform_name"] in platform_map:
            data["platform_id"] = platform_map[data["platform_name"]]
        if data.get("node_name") and data["node_name"] in node_map:
            data["node_id"] = node_map[data["node_name"]]
        # Remove helper fields that are not Server model columns
        data.pop("platform_name", None)
        data.pop("node_name", None)

        # Encrypt sensitive fields
        try:
            if data.get("password"):
                data["password"] = encrypt_value(data["password"])
            if data.get("ssh_key_private"):
                data["ssh_key_private"] = encrypt_value(data["ssh_key_private"])
                data["ssh_key_configured"] = True

            # Auto-generate Agent API Key
            data["agent_api_key"] = f"nxs-{secrets.token_urlsafe(32)}"

            server = Server(**data)
            created = await service.create_server(server)
            existing_domains.add(domain)

            result.created += 1
            result.created_ids.append(created.id)
        except Exception as e:
            result.failed += 1
            result.errors.append({"row": row_num, "name": name, "domain": domain, "error": str(e)[:200]})
            logger.warning(f"CSV import row {row_num} failed: {e}")

    result.total = result.created + result.skipped + result.failed

    # Audit log
    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="import_servers",
        target_type="server",
        target_id=0,
        detail=f"CSV导入: 成功{result.created} 跳过{result.skipped} 失败{result.failed}",
        ip_address=ip_address,
    ))

    return result


# ── Batch jobs (background, same UX as script library) ──

async def _start_server_batch(
    op: str,
    server_ids: list[int],
    admin: Admin,
    params: dict | None = None,
) -> BatchJobStarted:
    if op in ("install-agent", "upgrade-agent", "uninstall-agent"):
        base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
        if not base_url:
            raise HTTPException(status_code=400, detail="NEXUS_API_BASE_URL 未配置")
    try:
        started = await start_batch_job(
            op=op,
            server_ids=server_ids,
            operator=admin.username,
            params=params,
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    return BatchJobStarted(**started)


@router.get("/batch/jobs/{job_id}", response_model=BatchJobStatus)
async def get_server_batch_job(
    job_id: int,
    admin: Admin = Depends(get_current_admin),
):
    """Poll server batch job status and per-server results."""
    detail = await get_batch_job(job_id)
    if not detail:
        raise HTTPException(status_code=404, detail="任务不存在或已过期")
    return BatchJobStatus(**detail)


@router.get("/batch/jobs", response_model=dict)
async def list_server_batch_jobs(
    limit: int = Query(50, ge=1, le=200),
    offset: int = Query(0, ge=0),
    op: str | None = Query(None, max_length=50),
    admin: Admin = Depends(get_current_admin),
):
    """List persisted server batch job history (MySQL)."""
    return await list_batch_jobs(limit=limit, offset=offset, op=op or None)


@router.post("/batch/category", response_model=BatchJobStarted)
async def batch_update_category(
    payload: BatchCategoryUpdate,
    admin: Admin = Depends(get_current_admin),
):
    """Queue batch category update (background)."""
    category = (payload.category or "").strip()
    return await _start_server_batch(
        "category",
        payload.server_ids,
        admin,
        params={"category": category},
    )


@router.post("/batch/install-agent", response_model=BatchJobStarted)
async def batch_install_agent(
    payload: BatchAgentAction,
    admin: Admin = Depends(get_current_admin),
):
    """Queue Agent install on multiple servers (background)."""
    return await _start_server_batch("install-agent", payload.server_ids, admin)


@router.post("/batch/upgrade-agent", response_model=BatchJobStarted)
async def batch_upgrade_agent(
    payload: BatchAgentAction,
    admin: Admin = Depends(get_current_admin),
):
    """Queue Agent upgrade on multiple servers (background)."""
    return await _start_server_batch("upgrade-agent", payload.server_ids, admin)


@router.post("/batch/detect-path", response_model=BatchJobStarted)
async def batch_detect_path(
    payload: BatchAgentAction,
    admin: Admin = Depends(get_current_admin),
):
    """Queue SSH target_path detection (background)."""
    return await _start_server_batch("detect-path", payload.server_ids, admin)


@router.post("/batch/uninstall-agent", response_model=BatchJobStarted)
async def batch_uninstall_agent(
    payload: BatchAgentAction,
    admin: Admin = Depends(get_current_admin),
):
    """Queue Agent uninstall on multiple servers (background)."""
    return await _start_server_batch("uninstall-agent", payload.server_ids, admin)


# ── Credential polling: add-by-ip + pending failure queue ──


def _pending_to_dict(row: PendingServer) -> dict:
    return {
        "id": row.id,
        "name": row.name,
        "domain": row.domain,
        "port": row.port,
        "agent_port": row.agent_port,
        "target_path": row.target_path or "",
        "category": row.category,
        "platform_id": row.platform_id,
        "node_id": row.node_id,
        "attempts": row.attempts or 0,
        "last_error": row.last_error or "",
        "last_attempt_at": str(row.last_attempt_at) if row.last_attempt_at else None,
        "created_at": str(row.created_at) if row.created_at else None,
    }


async def _unique_server_name(db: AsyncSession, base: str) -> str:
    from sqlalchemy import select as sa_select

    candidate = base.strip()[:100] or "server"
    result = await db.execute(sa_select(Server.name).where(Server.name == candidate))
    if not result.scalar_one_or_none():
        return candidate
    for i in range(2, 1000):
        name = f"{candidate[:90]}-{i}"
        result = await db.execute(sa_select(Server.name).where(Server.name == name))
        if not result.scalar_one_or_none():
            return name
    raise HTTPException(500, "无法生成唯一服务器名称")


async def _server_exists_for_domain(db: AsyncSession, domain: str) -> bool:
    from sqlalchemy import select as sa_select

    result = await db.execute(sa_select(Server.id).where(Server.domain == domain).limit(1))
    return result.scalar_one_or_none() is not None


async def _create_server_from_poll(
    db: AsyncSession,
    service: ServerService,
    *,
    poll_match,
    name: str,
    domain: str,
    port: int,
    agent_port: int,
    target_path: str | None,
    category: str | None,
    platform_id: int | None,
    node_id: int | None,
    admin: Admin,
    request: Request,
) -> Server:
    import secrets
    from datetime import datetime, timezone

    from server.infrastructure.database.crypto import encrypt_value

    unique_name = await _unique_server_name(db, name)
    data: dict = {
        "name": unique_name,
        "domain": domain,
        "port": port,
        "username": poll_match.username,
        "auth_method": poll_match.auth_method,
        "agent_port": agent_port,
        "target_path": target_path or "",
        "category": category,
        "platform_id": platform_id,
        "node_id": node_id,
        "connectivity": "ok",
        "last_checked_at": datetime.now(timezone.utc),
        "agent_api_key": f"nxs-{secrets.token_urlsafe(32)}",
    }
    if poll_match.auth_method == "password":
        data["preset_id"] = poll_match.preset_id
        data["password"] = encrypt_value(poll_match.password)
    else:
        data["ssh_key_preset_id"] = poll_match.ssh_key_preset_id
        data["ssh_key_private"] = encrypt_value(poll_match.ssh_key_private)
        data["ssh_key_public"] = poll_match.ssh_key_public or ""
        data["ssh_key_configured"] = True

    server = Server(**data)
    created = await service.create_server(server)

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="create_server",
        target_type="server",
        target_id=created.id,
        detail=(
            f"凭据轮询添加 名称={created.name} 地址={created.domain}:{created.port} "
            f"预设={poll_match.preset_name} 用户={poll_match.username}"
        ),
        ip_address=ip_address,
    ))
    return created


@router.get("/pending", response_model=dict)
async def list_pending_servers(
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """List servers that failed SSH credential polling."""
    from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl

    repo = PendingServerRepositoryImpl(db)
    rows = await repo.get_all()
    items = [_pending_to_dict(r) for r in rows]
    return {"items": items, "total": len(items)}


async def _try_add_by_ip(
    request: Request,
    *,
    domain: str,
    port: int,
    agent_port: int,
    target_path: Optional[str],
    category: Optional[str],
    platform_id: Optional[int],
    node_id: Optional[int],
    name: Optional[str],
    admin: Admin,
    service: ServerService,
    db: AsyncSession,
) -> AddByIpsBatchItemResult:
    """Poll credential presets for one host; create server or enqueue pending."""
    from server.application.services.credential_poller import (
        format_poll_errors,
        poll_credentials,
    )
    from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl

    domain = domain.strip()
    if await _server_exists_for_domain(db, domain):
        return AddByIpsBatchItemResult(
            domain=domain,
            status="skipped",
            message="已存在于服务器列表",
        )

    display_name = (name or domain).strip()
    poll = await poll_credentials(db, domain, port)

    if poll.success and poll.match:
        created = await _create_server_from_poll(
            db,
            service,
            poll_match=poll.match,
            name=display_name,
            domain=domain,
            port=port,
            agent_port=agent_port,
            target_path=target_path,
            category=category,
            platform_id=platform_id,
            node_id=node_id,
            admin=admin,
            request=request,
        )
        return AddByIpsBatchItemResult(
            domain=domain,
            status="created",
            server_id=created.id,
            matched_preset=poll.match.preset_name,
            matched_username=poll.match.username,
        )

    err_text = format_poll_errors(poll.errors)
    pending_repo = PendingServerRepositoryImpl(db)
    pending = await pending_repo.record_failure(
        name=display_name,
        domain=domain,
        port=port,
        agent_port=agent_port,
        target_path=target_path,
        category=category,
        platform_id=platform_id,
        node_id=node_id,
        last_error=err_text,
    )

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="add_server_by_ip_failed",
        target_type="pending_server",
        target_id=pending.id,
        detail=f"凭据轮询失败 地址={domain}:{port}",
        ip_address=ip_address,
    ))

    from server.api.schemas import CredentialPollErrorOut

    return AddByIpsBatchItemResult(
        domain=domain,
        status="pending",
        pending_id=pending.id,
        message=err_text,
        errors=[
            CredentialPollErrorOut(
                preset_type=e.preset_type,
                preset_name=e.preset_name,
                username=e.username,
                error=e.error,
            )
            for e in poll.errors
        ],
    )


@router.post("/add-by-ip", response_model=dict)
async def add_server_by_ip(
    request: Request,
    payload: AddByIpRequest,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Add server by IP only — poll all credential presets until SSH login succeeds."""
    item = await _try_add_by_ip(
        request,
        domain=payload.domain,
        port=payload.port,
        agent_port=payload.agent_port,
        target_path=payload.target_path,
        category=payload.category,
        platform_id=payload.platform_id,
        node_id=payload.node_id,
        name=payload.name,
        admin=admin,
        service=service,
        db=db,
    )

    if item.status == "skipped":
        raise HTTPException(409, f"服务器 {item.domain} 已存在于列表中")

    if item.status == "created":
        return {
            "success": True,
            "server": {"id": item.server_id, "domain": item.domain},
            "matched_preset": item.matched_preset,
            "matched_username": item.matched_username,
        }

    return {
        "success": False,
        "pending_id": item.pending_id,
        "errors": [e.model_dump() for e in (item.errors or [])],
        "message": item.message,
    }


@router.post("/add-by-ips-batch", response_model=AddByIpsBatchResult)
async def add_servers_by_ips_batch(
    request: Request,
    payload: AddByIpsBatchRequest,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Batch add servers: one IP/domain per line, credential preset polling, failures → pending."""
    from server.application.services.credential_poller import parse_batch_ip_lines

    try:
        parsed = parse_batch_ip_lines(payload.text, max_lines=MAX_BATCH_IP_LINES)
    except ValueError as e:
        raise HTTPException(400, str(e)) from e

    entries = parsed.entries
    if not entries:
        raise HTTPException(400, "未解析到有效 IP 或域名")

    items: list[AddByIpsBatchItemResult] = []
    created = pending = skipped = 0

    for entry in entries:
        item = await _try_add_by_ip(
            request,
            domain=entry.domain,
            port=payload.port,
            agent_port=payload.agent_port,
            target_path=payload.target_path,
            category=payload.category,
            platform_id=payload.platform_id,
            node_id=payload.node_id,
            name=entry.name,
            admin=admin,
            service=service,
            db=db,
        )
        items.append(item)
        if item.status == "created":
            created += 1
        elif item.status == "pending":
            pending += 1
        else:
            skipped += 1

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="add_servers_by_ips_batch",
        target_type="server",
        target_id=None,
        detail=(
            f"批量 IP 添加 输入{parsed.input_lines}行 去重{parsed.duplicates_removed} "
            f"实处理{len(entries)} 成功{created} 待连接{pending} 跳过{skipped}"
        ),
        ip_address=ip_address,
    ))

    return AddByIpsBatchResult(
        total=len(entries),
        created=created,
        pending=pending,
        skipped=skipped,
        input_lines=parsed.input_lines,
        duplicates_removed=parsed.duplicates_removed,
        items=items,
    )


@router.post("/pending/{pending_id}/retry", response_model=dict)
async def retry_pending_server(
    pending_id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Re-poll all credential presets for a pending server."""
    from server.application.services.credential_poller import (
        format_poll_errors,
        poll_credentials,
    )
    from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl

    pending_repo = PendingServerRepositoryImpl(db)
    pending = await pending_repo.get_by_id(pending_id)
    if not pending:
        raise HTTPException(404, "待连接服务器不存在")

    if await _server_exists_for_domain(db, pending.domain):
        await pending_repo.delete(pending_id)
        raise HTTPException(409, f"服务器 {pending.domain} 已存在于列表中，已从失败列表移除")

    poll = await poll_credentials(db, pending.domain, pending.port or 22)

    if poll.success and poll.match:
        created = await _create_server_from_poll(
            db,
            service,
            poll_match=poll.match,
            name=pending.name,
            domain=pending.domain,
            port=pending.port or 22,
            agent_port=pending.agent_port or 8601,
            target_path=pending.target_path,
            category=pending.category,
            platform_id=pending.platform_id,
            node_id=pending.node_id,
            admin=admin,
            request=request,
        )
        await pending_repo.delete(pending_id)
        return {
            "success": True,
            "server": _server_to_dict(created),
            "matched_preset": poll.match.preset_name,
            "matched_username": poll.match.username,
        }

    err_text = format_poll_errors(poll.errors)
    pending.attempts = (pending.attempts or 0) + 1
    pending.last_error = err_text
    from datetime import datetime, timezone

    pending.last_attempt_at = datetime.now(timezone.utc)
    await pending_repo.update(pending)

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="retry_pending_server_failed",
        target_type="pending_server",
        target_id=pending.id,
        detail=f"凭据轮询重试失败 地址={pending.domain}:{pending.port or 22} 第{pending.attempts}次",
        ip_address=ip_address,
    ))

    return {
        "success": False,
        "pending_id": pending.id,
        "errors": [{"preset_type": e.preset_type, "preset_name": e.preset_name, "username": e.username, "error": e.error} for e in poll.errors],
        "message": err_text,
    }


@router.delete("/pending/{pending_id}", status_code=204)
async def delete_pending_server(
    pending_id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl

    repo = PendingServerRepositoryImpl(db)
    pending = await repo.get_by_id(pending_id)
    if not pending:
        raise HTTPException(404, "待连接服务器不存在")
    if not await repo.delete(pending_id):
        raise HTTPException(404, "待连接服务器不存在")

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="delete_pending_server",
        target_type="pending_server",
        target_id=pending_id,
        detail=f"删除待连接 地址={pending.domain}:{pending.port or 22}",
        ip_address=ip_address,
    ))


@router.get("/{id}", response_model=dict)
async def get_server(
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
):
    """Get a single server by ID with live status from Redis"""
    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    server_data = _server_to_dict(server)

    redis = get_redis()
    try:
        heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{id}")
        _apply_heartbeat_overlay(server_data, server, heartbeat or None)
    except Exception as e:
        logger.warning(f"Redis read failed for server {id}: {e}")
        server_data["status"] = _derive_server_status(
            server_data.get("is_online"),
            server.agent_version,
            has_agent=_server_has_agent_monitoring(server),
        )

    return server_data


@router.get("/{id}/files-capability", response_model=dict)
async def get_files_capability(
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
):
    """Probe remote sudo capability for file manager (chmod/chown/write)."""
    from server.infrastructure.ssh.files_capability import probe_files_capability

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")
    return await probe_files_capability(server)


@router.post("/{id}/setup-files-sudo", response_model=dict)
async def setup_files_sudo(
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
):
    """Auto-configure passwordless sudoers for file manager on non-root servers.

    Uses the stored SSH password to run `sudo -S` and write
    /etc/sudoers.d/nexus-files on the target server.
    Requires the SSH user to have sudo access with password.
    """
    import textwrap
    from server.infrastructure.ssh.asyncssh_pool import ssh_pool
    from server.infrastructure.database.crypto import decrypt_value

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    ssh_user = (server.username or "root").strip() or "root"
    if ssh_user == "root":
        return {"ok": True, "message": "root 用户无需配置 sudoers，已拥有完整权限"}

    if not server.password:
        raise HTTPException(
            status_code=400,
            detail="该服务器未存储 SSH 密码，无法自动配置。请手动在目标机配置 sudoers。",
        )

    plain_password = decrypt_value(server.password)
    sudoers_content = textwrap.dedent(f"""\
        # nexus-files: auto-configured by Nexus file manager
        {ssh_user} ALL=(root) NOPASSWD: \\
            /bin/ls, /usr/bin/ls, \\
            /bin/cat, /usr/bin/cat, \\
            /usr/bin/tee, /bin/tee, \\
            /bin/mkdir, /usr/bin/mkdir, \\
            /bin/cp, /usr/bin/cp, \\
            /bin/mv, /usr/bin/mv, \\
            /bin/rm, /usr/bin/rm, \\
            /bin/chmod, /usr/bin/chmod, \\
            /bin/chown, /usr/bin/chown, \\
            /bin/bash, /usr/bin/bash
    """)

    sudoers_path = "/etc/sudoers.d/nexus-files"
    # Use heredoc via sudo -S to avoid exposing password in command arguments
    write_cmd = (
        f"printf %s {shlex.quote(sudoers_content)}"
        f" | sudo -S tee {shlex.quote(sudoers_path)} > /dev/null"
        f" && sudo chmod 440 {shlex.quote(sudoers_path)}"
        f" && sudo visudo -cf {shlex.quote(sudoers_path)}"
    )

    conn = None
    try:
        conn = await ssh_pool.acquire(server)
        # Feed password via stdin for sudo -S (avoids exposing password in process args)
        result = await asyncio.wait_for(
            conn.run(write_cmd, input=plain_password + "\n", timeout=15),
            timeout=20,
        )
        exit_code = result.exit_status if result.exit_status is not None else -1
        stderr_out = (result.stderr or "").strip()
        stdout_out = (result.stdout or "").strip()
        if exit_code != 0:
            detail = stderr_out or stdout_out or f"命令退出码 {exit_code}"
            raise HTTPException(
                status_code=500,
                detail=f"sudoers 写入失败：{detail[:300]}",
            )
    except HTTPException:
        raise
    except Exception as exc:
        raise HTTPException(
            status_code=500,
            detail=f"SSH 执行失败：{exc}",
        ) from exc
    finally:
        if conn is not None:
            await ssh_pool.release(server.id)

    return {
        "ok": True,
        "message": (
            f"sudoers 已配置：{sudoers_path}。"
            "现在文件管理器可使用 sudo -n 自动提权浏览/操作文件。"
        ),
    }


@router.post("/", response_model=dict, status_code=201)
async def create_server(
    request: Request,
    payload: ServerCreate,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Create a new server (sensitive fields encrypted at rest)"""
    from server.infrastructure.database.crypto import encrypt_value, decrypt_value
    from server.infrastructure.database.password_preset_repo import PasswordPresetRepositoryImpl
    from server.infrastructure.database.ssh_key_preset_repo import SshKeyPresetRepositoryImpl

    from server.utils.files_elevation import apply_files_elevation_payload

    data = payload.model_dump(exclude_none=True)
    apply_files_elevation_payload(data)

    # If preset_id is provided, resolve the preset password (server-side, no re-auth needed)
    if data.get("preset_id") and data.get("auth_method") == "password" and not data.get("password"):
        preset_repo = PasswordPresetRepositoryImpl(db)
        preset = await preset_repo.get_by_id(data["preset_id"])
        if preset:
            data["password"] = decrypt_value(preset.encrypted_pw)
            if preset.username and (not data.get("username") or data.get("username") == "root"):
                data["username"] = preset.username
        # If preset not found, fall through — user may provide password manually

    # If ssh_key_preset_id is provided, resolve the preset private key (server-side, no re-auth needed)
    if data.get("ssh_key_preset_id") and data.get("auth_method") == "key" and not data.get("ssh_key_private"):
        ssh_key_preset_repo = SshKeyPresetRepositoryImpl(db)
        ssh_key_preset = await ssh_key_preset_repo.get_by_id(data["ssh_key_preset_id"])
        if ssh_key_preset:
            data["ssh_key_private"] = decrypt_value(ssh_key_preset.encrypted_private_key)
            if ssh_key_preset.public_key and not data.get("ssh_key_public"):
                data["ssh_key_public"] = ssh_key_preset.public_key
            if ssh_key_preset.username and (not data.get("username") or data.get("username") == "root"):
                data["username"] = ssh_key_preset.username
        # If preset not found, fall through — user may provide key manually

    # Encrypt sensitive fields before storing
    if data.get("password"):
        data["password"] = encrypt_value(data["password"])
    if data.get("ssh_key_private"):
        data["ssh_key_private"] = encrypt_value(data["ssh_key_private"])
        data["ssh_key_configured"] = True

    # Auto-generate Agent API Key — used by /install-agent, no need to expose to frontend
    import secrets
    data["agent_api_key"] = f"nxs-{secrets.token_urlsafe(32)}"

    server = Server(**data)
    created = await service.create_server(server)

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="create_server",
        target_type="server",
        target_id=created.id,
        detail=f"名称={created.name} 域名={created.domain}",
        ip_address=ip_address,
    ))

    return _server_to_dict(created)


@router.put("/{id}", response_model=dict)
async def update_server(
    request: Request,
    id: int,
    payload: ServerUpdate,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Update an existing server (sensitive fields encrypted at rest)"""
    from server.infrastructure.database.crypto import encrypt_value, decrypt_value
    from server.infrastructure.database.password_preset_repo import PasswordPresetRepositoryImpl
    from server.infrastructure.database.ssh_key_preset_repo import SshKeyPresetRepositoryImpl

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    from server.utils.files_elevation import apply_files_elevation_payload

    # If preset_id is provided (and not None), resolve the preset password
    update_data = payload.model_dump(exclude_unset=True)
    apply_files_elevation_payload(update_data)
    if "preset_id" in update_data and update_data["preset_id"] is not None and not update_data.get("password"):
        preset_repo = PasswordPresetRepositoryImpl(db)
        preset = await preset_repo.get_by_id(update_data["preset_id"])
        if preset:
            update_data["password"] = decrypt_value(preset.encrypted_pw)
            if preset.username:
                update_data["username"] = preset.username

    # If ssh_key_preset_id is provided, resolve the preset private key
    if "ssh_key_preset_id" in update_data and update_data["ssh_key_preset_id"] is not None and not update_data.get("ssh_key_private"):
        ssh_key_preset_repo = SshKeyPresetRepositoryImpl(db)
        ssh_key_preset = await ssh_key_preset_repo.get_by_id(update_data["ssh_key_preset_id"])
        if ssh_key_preset:
            update_data["ssh_key_private"] = decrypt_value(ssh_key_preset.encrypted_private_key)
            if ssh_key_preset.public_key:
                update_data["ssh_key_public"] = ssh_key_preset.public_key
            if ssh_key_preset.username:
                update_data["username"] = ssh_key_preset.username

    # Explicit allowlist to prevent mass-assignment to sensitive columns
    # agent_api_key is excluded — must be set via dedicated /agent-key endpoint
    ALLOWED_FIELDS = {
        "name", "domain", "port", "username", "auth_method",
        "password", "ssh_key_path", "ssh_key_private", "ssh_key_public",
        "agent_port", "preset_id", "ssh_key_preset_id", "description", "target_path", "category",
        "platform_id", "node_id", "protocols", "extra_attrs",
    }

    # When auth_method changes, clear the opposite side's credentials
    new_auth = update_data.get("auth_method")
    if new_auth == "password":
        # Switching to password: clear key-side fields if not explicitly provided
        update_data.setdefault("ssh_key_private", None)
        update_data.setdefault("ssh_key_path", None)
        update_data.setdefault("ssh_key_public", None)
        update_data.setdefault("ssh_key_preset_id", None)
        # Also clear ssh_key_configured on the server object
        server.ssh_key_configured = False
    elif new_auth == "key":
        # Switching to key: clear password-side fields if not explicitly provided
        update_data.setdefault("password", None)
        update_data.setdefault("preset_id", None)

    for key, value in update_data.items():
        if key not in ALLOWED_FIELDS:
            continue
        # Encrypt sensitive fields on update
        if key == "password" and value:
            value = encrypt_value(value)
        elif key == "ssh_key_private" and value:
            value = encrypt_value(value)
            server.ssh_key_configured = True  # Mark key as configured when private key is set
        setattr(server, key, value)
    updated = await service.update_server(server)

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="update_server",
        target_type="server",
        target_id=id,
        detail=f"名称={updated.name}",
        ip_address=ip_address,
    ))

    return _server_to_dict(updated)


@router.delete("/{id}", status_code=204)
async def delete_server(
    request: Request,
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Delete a server"""
    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")
    result = await service.delete_server(id)
    if not result:
        raise HTTPException(status_code=404, detail="Server not found")

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="delete_server",
        target_type="server",
        target_id=id,
        detail=f"名称={server.name}",
        ip_address=ip_address,
    ))


# ── Agent API Key ──

@router.post("/{id}/agent-key", response_model=dict)
async def generate_agent_api_key(
    request: Request,
    id: int,
    payload: ApiKeyRevealRequest,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Generate a new per-server Agent API key (re-auth required)"""
    # Re-authenticate: require current password (same pattern as reveal_api_key)
    from server.infrastructure.database.admin_repo import AdminRepositoryImpl

    admin_repo = AdminRepositoryImpl(db)
    current = await admin_repo.get_by_id(admin.id)
    if not current or not bcrypt.checkpw(
        payload.current_password.encode(),
        current.password_hash.encode(),
    ):
        raise HTTPException(status_code=400, detail="当前密码错误")

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    import secrets
    new_key = f"nxs-{secrets.token_urlsafe(32)}"
    server.agent_api_key = new_key
    await service.update_server(server)

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="generate_agent_key",
        target_type="server",
        target_id=id,
        detail=f"为 {server.name} 生成新 Agent API Key",
        ip_address=ip_address,
    ))

    return {
        "server_id": id,
        "agent_api_key": new_key,
        "agent_api_key_preview": f"{new_key[:12]}...",
        "display_once": True,
        "message": "完整密钥仅在本响应中返回一次，请立即复制保存",
    }


# ── Agent Install ──

@router.post("/{id}/install-agent", response_model=dict)
async def install_agent_remote(
    request: Request,
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Install Nexus Agent on the managed server via SSH.

    Uses the server's stored SSH credentials to connect and run install.sh.
    Requires NEXUS_API_BASE_URL to be configured (used in the install command).
    """
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    from server.application.server_batch_common import agent_install_skip_result
    from server.utils.agent_version import get_central_agent_version

    skip = await agent_install_skip_result(
        server=server,
        server_id=id,
        server_name=server.name or str(id),
        central_version=get_central_agent_version(),
    )
    if skip is not None:
        return {
            "success": True,
            "server_id": id,
            "server_name": server.name,
            "stdout": skip.get("stdout", ""),
            "skipped": True,
        }

    # Validate prerequisites
    base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
    if not base_url:
        raise HTTPException(
            status_code=400,
            detail="NEXUS_API_BASE_URL 未配置，无法生成安装命令。请在系统设置中配置主站对外 URL。",
        )

    api_key = (server.agent_api_key or "").strip()
    if not api_key:
        import secrets
        api_key = f"nxs-{secrets.token_urlsafe(32)}"
        server.agent_api_key = api_key
        await service.update_server(server)

    agent_port = server.agent_port or 8601
    ssh_user = (server.username or "root").strip()

    # Build install command — download first, then execute (avoids pipe hiding curl errors)
    install_cmd = (
        f"TMP=$(mktemp) && curl -fsSL {shlex.quote(base_url)}/agent/install.sh -o \"$TMP\" && bash \"$TMP\" -- "
        f"--url {shlex.quote(base_url)} --key {shlex.quote(api_key)} "
        f"--id {id} --port {shlex.quote(str(agent_port))}"
        f"; RC=$?; rm -f \"$TMP\"; exit $RC"
    )

    # For non-root SSH users, pre-configure passwordless sudo so install.sh
    # can run apt-get / systemctl / mkdir /opt without prompting.
    install_cmd = _sudo_wrap(install_cmd, ssh_user)

    # Execute via SSH
    try:
        result = await exec_ssh_command(server, install_cmd, timeout=120)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") from e

    if result["exit_code"] != 0:
        raise HTTPException(
            status_code=400,
            detail=f"安装失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}",
        )

    # Check for "Status: FAILED" in stdout (install.sh exits 0 but service may not start)
    if "FAILED" in result.get("stdout", ""):
        raise HTTPException(
            status_code=400,
            detail="安装完成但 Agent 启动失败，请检查子服务器日志: journalctl -u nexus-agent",
        )

    # Audit
    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="install_agent",
        target_type="server",
        target_id=id,
        detail=f"Agent 远程安装: {server.name} ({server.domain})",
        ip_address=ip_address,
    ))

    return {
        "success": True,
        "server_id": id,
        "server_name": server.name,
        "stdout": result["stdout"][:3000],
        "exit_code": result["exit_code"],
    }


@router.post("/{id}/uninstall-agent", response_model=dict)
async def uninstall_agent_remote(
    request: Request,
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Uninstall Nexus Agent on the managed server via SSH.

    Stops the systemd service, removes the install directory and log file,
    then clears the agent metadata (is_online, agent_version) in the database.
    """
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
    if not base_url:
        raise HTTPException(
            status_code=400,
            detail="NEXUS_API_BASE_URL 未配置，无法下载卸载脚本。",
        )

    ssh_user = (server.username or "root").strip()

    # Build uninstall command — download uninstall.sh then execute
    uninstall_cmd = (
        f"TMP=$(mktemp) && curl -fsSL {shlex.quote(base_url)}/agent/uninstall.sh -o \"$TMP\" "
        f"&& bash \"$TMP\" ; RC=$?; rm -f \"$TMP\"; exit $RC"
    )
    uninstall_cmd = _sudo_wrap(uninstall_cmd, ssh_user)

    # Execute via SSH
    try:
        result = await exec_ssh_command(server, uninstall_cmd, timeout=60)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") from e

    if result["exit_code"] != 0:
        raise HTTPException(
            status_code=400,
            detail=f"卸载失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}",
        )

    # Clear agent metadata: Redis (real-time source) + MySQL (history)
    try:
        redis = get_redis()
        await redis.delete(f"{REDIS_KEY_PREFIX}{id}")
    except Exception:
        logger.warning(f"Failed to clear Redis heartbeat for server {id}", exc_info=True)
    server.is_online = False
    server.agent_version = None
    server.agent_api_key = None
    await db.commit()

    # Audit
    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="uninstall_agent",
        target_type="server",
        target_id=id,
        detail=f"Agent 远程卸载: {server.name} ({server.domain})",
        ip_address=ip_address,
    ))

    return {
        "success": True,
        "server_id": id,
        "server_name": server.name,
        "stdout": result["stdout"][:3000],
        "exit_code": result["exit_code"],
    }


@router.post("/{id}/agent-install-cmd", response_model=dict)
async def get_agent_install_cmd(
    id: int,
    payload: "ApiKeyRevealRequest",
    request: Request,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Return the curl install command for this server (re-auth required).

    Security: Only exposes per-server agent_api_key after password verification.
    Never falls back to global API_KEY — that would expose all servers' auth.
    """
    # Re-authenticate: require current password (same pattern as reveal_api_key)
    from server.infrastructure.database.admin_repo import AdminRepositoryImpl

    admin_repo = AdminRepositoryImpl(db)
    current = await admin_repo.get_by_id(admin.id)
    if not current or not bcrypt.checkpw(
        payload.current_password.encode(),
        current.password_hash.encode(),
    ):
        raise HTTPException(status_code=400, detail="当前密码错误")

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
    agent_port = server.agent_port or 8601

    # Only use per-server key — NEVER expose global API_KEY via this endpoint
    api_key = server.agent_api_key
    has_key = bool(api_key)
    cmd = None
    if base_url and api_key:
        import shlex
        cmd = (
            f"curl -fsSL {shlex.quote(base_url)}/agent/install.sh | bash -s -- "
            f"--url {shlex.quote(base_url)} --key {shlex.quote(api_key)} "
            f"--id {id} --port {shlex.quote(str(agent_port))}"
        )

    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="reveal_install_cmd",
        target_type="server",
        target_id=id,
        detail=f"查看 {server.name} 的安装命令",
        ip_address=ip_address,
    ))

    return {
        "server_id": id,
        "server_name": server.name,
        "base_url": base_url,
        "has_agent_key": has_key,
        "agent_port": agent_port,
        "install_cmd": cmd,
        "is_online": server.is_online,
    }


# ── Agent Upgrade ──

@router.post("/{id}/upgrade-agent", response_model=dict)
async def upgrade_agent(
    request: Request,
    id: int,
    admin: Admin = Depends(get_current_admin),
    service: ServerService = Depends(get_server_service),
    db: AsyncSession = Depends(get_db),
):
    """Upgrade Nexus Agent on the managed server via SSH.

    Downloads the latest agent.py from the central server and restarts
    the nexus-agent systemd service. Does NOT touch config.json.
    """
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command

    server = await service.get_server(id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
    if not base_url:
        raise HTTPException(
            status_code=400,
            detail="NEXUS_API_BASE_URL 未配置，无法构建下载地址",
        )

    from server.application.server_batch_common import agent_files_curl_cmds

    install_dir = "/opt/nexus-agent"
    venv_python = f"{install_dir}/.venv/bin/python"
    ssh_user = (server.username or "root").strip()
    agent_port = server.agent_port or 8601

    # Step 1: Check Python version in venv
    pyver_cmd = (
        f"{venv_python} -c '"
        "import sys; v=sys.version_info; "
        "print(f\"py_ok_{v.major}.{v.minor}.{v.micro}\") "
        "if v.major==3 and v.minor>=10 else sys.exit(1)'"
    )
    try:
        pyver = await exec_ssh_command(server, pyver_cmd, timeout=15)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") from e

    if pyver["exit_code"] != 0 or "py_ok_" not in pyver["stdout"]:
        ver_info = pyver["stderr"].replace("\n", "; ").strip()[:200] if pyver["stderr"] else "venv 不存在或 Python 版本 < 3.10"
        raise HTTPException(
            status_code=424,
            detail=(
                f"子机 Python 版本不满足要求（需要 3.10+），请手动修复后重试。\n"
                f"  venv 路径: {venv_python}\n"
                f"  详情: {ver_info}"
            ),
        )

    # Step 2: Ensure rsync (required for push/sync) — install if missing
    rsync_check = "command -v rsync >/dev/null 2>&1 || (apt-get update -qq && apt-get install -y -qq rsync) || (yum install -y -q rsync) || (dnf install -y -q rsync)"
    try:
        await exec_ssh_command(server, rsync_check, timeout=60)
    except Exception as rsync_err:
        logger.debug(f"rsync check/install skipped (non-blocking): {rsync_err}")

    # Step 3: Auto-update pip dependencies (locked versions)
    pip_cmd = (
        f"{venv_python} -m pip install -q "
        f"fastapi==0.115.6 uvicorn==0.34.0 httpx==0.28.1 psutil python-multipart==0.0.19"
    )

    # Step 4: Backup current agent.py
    backup_cmd = f"cp {install_dir}/agent.py {install_dir}/agent.py.bak"

    # Step 4: pip install + backup + download new agent.py → restart service
    # For non-root users, systemctl needs sudo (even after _sudo_wrap sets up NOPASSWD)
    systemctl_prefix = "sudo " if ssh_user != "root" else ""
    kill_port = f"fuser -k {agent_port}/tcp 2>/dev/null || true; "
    upgrade_cmd = _sudo_wrap(
        f"{pip_cmd} "
        f"&& {backup_cmd} "
        f"&& {agent_files_curl_cmds(base_url, install_dir)} "
        f"&& {kill_port}{systemctl_prefix}systemctl restart nexus-agent "
        f"&& sleep 2 "
        f"&& {systemctl_prefix}systemctl is-active nexus-agent "
        f"&& echo 'upgrade_ok'",
        ssh_user,
    )

    try:
        result = await exec_ssh_command(server, upgrade_cmd, timeout=120)
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") from e

    success = result["exit_code"] == 0 and "upgrade_ok" in result["stdout"]

    if not success:
        # Rollback: restore backup and restart
        rollback_cmd = _sudo_wrap(
            f"cp {install_dir}/agent.py.bak {install_dir}/agent.py "
            f"&& {systemctl_prefix}systemctl restart nexus-agent",
            ssh_user,
        )
        try:
            await exec_ssh_command(server, rollback_cmd, timeout=30)
        except Exception as rb_err:
            logger.warning(f"Rollback failed for server {id}: {rb_err}")
        raise HTTPException(
            status_code=400,
            detail=f"升级失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}，已回滚至旧版本",
        )

    # Audit
    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="upgrade_agent",
        target_type="server",
        target_id=id,
        detail=f"Agent 升级: {server.name} ({server.domain})",
        ip_address=ip_address,
    ))

    return {"success": True, "server_id": id, "server_name": server.name, "stdout": result["stdout"][:500]}



# ── Health Check ──

@router.post("/check", response_model=BatchJobStarted)
async def check_servers(
    payload: ServerCheck,
    admin: Admin = Depends(get_current_admin),
):
    """Queue SSH health check for selected servers (background)."""
    return await _start_server_batch("health-check", payload.server_ids, admin)



@router.get("/{id}/logs", response_model=list)
async def get_server_logs(
    id: int,
    limit: int = Query(50, ge=1, le=200),
    admin: Admin = Depends(get_current_admin),
    sync_service: SyncService = Depends(get_sync_service),
):
    """Get sync logs for a specific server"""
    logs = await sync_service.get_sync_logs(id, limit)
    return [_sync_log_to_dict(log) for log in logs]


# ── Helper functions ──

async def _build_server_list_with_overlay(redis, servers: list) -> list[dict]:
    """Batch Redis heartbeat overlay for server list rows."""
    if not servers:
        return []
    pipe = redis.pipeline()
    for server in servers:
        pipe.hgetall(f"{REDIS_KEY_PREFIX}{server.id}")
    heartbeats = await pipe.execute()
    result: list[dict] = []
    for server, heartbeat in zip(servers, heartbeats):
        server_data = _server_to_dict(server)
        try:
            _apply_heartbeat_overlay(server_data, server, heartbeat or None)
        except Exception as e:
            logger.warning(f"Redis read failed for server {server.id}: {e}")
            server_data["status"] = _derive_server_status(
                server_data.get("is_online"),
                server.agent_version,
                has_agent=_server_has_agent_monitoring(server),
            )
        result.append(server_data)
    return result


def _derive_server_status(is_online: bool | None, agent_version: str | None, *, has_agent: bool = False) -> str:
    """Map live/DB fields to frontend status: online | offline | unknown."""
    has_agent = has_agent or bool(agent_version and str(agent_version).strip())
    if has_agent or is_online:
        return "online" if is_online else "offline"
    return "unknown"


def _apply_heartbeat_overlay(server_data: dict, server: Server, heartbeat: dict | None) -> None:
    """Overlay Redis heartbeat onto server_data and refresh derived status."""
    has_agent = _server_has_agent_monitoring(server)
    if heartbeat:
        server_data["is_online"] = heartbeat.get("is_online") == "True"
        if heartbeat.get("system_info"):
            try:
                server_data["system_info"] = json.loads(heartbeat.get("system_info", "{}"))
            except (json.JSONDecodeError, TypeError):
                pass
        server_data["last_heartbeat"] = heartbeat.get(
            "last_heartbeat", server_data.get("last_heartbeat")
        )
        server_data["agent_version"] = heartbeat.get(
            "agent_version", server_data.get("agent_version", "")
        )
        try:
            server_data["time_drift_seconds"] = float(heartbeat.get("time_drift_seconds", 0))
        except (ValueError, TypeError):
            server_data["time_drift_seconds"] = 0.0
        server_data["drift_level"] = heartbeat.get("drift_level", "ok")
        server_data["_source"] = "redis"
    elif has_agent:
        server_data["is_online"] = False
        server_data["_source"] = "redis_miss"
    server_data["status"] = _derive_server_status(
        server_data.get("is_online"),
        server_data.get("agent_version") or server.agent_version,
        has_agent=has_agent,
    )


def _server_to_dict(server: Server) -> dict:
    """Convert Server model to API response dict (hide sensitive fields)"""
    from server.utils.files_elevation import get_files_elevation

    return {
        "id": server.id,
        "name": server.name,
        "domain": server.domain,
        "port": server.port,
        "username": server.username,
        "auth_method": server.auth_method,
        "password_set": bool(server.password),
        "ssh_key_path": server.ssh_key_path,
        "ssh_key_public": server.ssh_key_public or "",
        "ssh_key_private_set": bool(server.ssh_key_private),
        "agent_port": server.agent_port,
        "agent_api_key": (server.agent_api_key[:8] + "...") if server.agent_api_key and len(server.agent_api_key) > 8 else (server.agent_api_key or ""),
        "agent_api_key_set": bool(server.agent_api_key),
        "preset_id": server.preset_id,
        "ssh_key_preset_id": server.ssh_key_preset_id,
        "description": server.description,
        "target_path": server.target_path,
        "category": server.category,
        "platform_id": server.platform_id,
        "node_id": server.node_id,
        "protocols": server.protocols,
        "extra_attrs": server.extra_attrs,
        "files_elevation": get_files_elevation(server).value,
        "connectivity": server.connectivity,
        "ssh_key_configured": server.ssh_key_configured,
        "is_online": server.is_online,
        "status": _derive_server_status(
            server.is_online,
            server.agent_version,
            has_agent=_server_has_agent_monitoring(server),
        ),
        "last_heartbeat": str(server.last_heartbeat) if server.last_heartbeat else None,
        "last_checked_at": str(server.last_checked_at) if server.last_checked_at else None,
        "system_info": server.system_info,
        "agent_version": server.agent_version,
        "created_at": str(server.created_at) if server.created_at else None,
        "updated_at": str(server.updated_at) if server.updated_at else None,
    }


def _sync_log_to_dict(log, server_name: str = None) -> dict:
    """Convert SyncLog model to API response dict"""
    return {
        "id": log.id,
        "server_id": log.server_id,
        "server_name": server_name,
        "source_path": log.source_path,
        "target_path": log.target_path,
        "trigger_type": log.trigger_type,
        "operator": log.operator,
        "status": log.status,
        "sync_mode": log.sync_mode,
        "files_total": log.files_total,
        "files_transferred": log.files_transferred,
        "files_skipped": log.files_skipped,
        "bytes_transferred": log.bytes_transferred,
        "duration_seconds": log.duration_seconds,
        "error_message": log.error_message,
        "diff_summary": log.diff_summary,
        "push_permission": log.push_permission,
        "started_at": str(log.started_at) if log.started_at else None,
        "finished_at": str(log.finished_at) if log.finished_at else None,
    }