Nexus/server/api/sync_v2.py

"""Nexus — Sync Engine v2 API Routes (Step 4)

S1: POST /api/sync/files    — rsync file sync
P1: POST /api/sync/preview  — rsync dry-run (方案D: 智能提示预览)
"""

import io
import logging

from fastapi import APIRouter, Depends, HTTPException, Request

from server.api.schemas import SyncFiles, SyncBrowse, SyncBrowseLocal, SyncPreview, FileOperation, LocalFileOperation, LocalFilePreview, SyncVerify, SyncCancel, ValidateSourcePath, SyncDiagnose, FileSyncDiff
from server.api.dependencies import get_server_service
from server.application.services.sync_engine_v2 import SyncEngineV2
from server.infrastructure.database.server_repo import ServerRepositoryImpl
from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl
from server.api.auth_jwt import get_current_admin
from server.domain.models import Admin, AuditLog

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

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


async def _get_sync_engine(request: Request) -> SyncEngineV2:
    """DI factory for SyncEngineV2"""
    session = request.state.db
    return SyncEngineV2(
        server_repo=ServerRepositoryImpl(session),
        sync_log_repo=SyncLogRepositoryImpl(session),
        audit_repo=AuditLogRepositoryImpl(session),
        retry_repo=PushRetryJobRepositoryImpl(session),
    )


@router.post("/files")
async def sync_files(
    payload: SyncFiles,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """S1: File sync via rsync over SSH (with parallel execution)"""
    engine = await _get_sync_engine(request)
    return await engine.sync_files(
        server_ids=payload.server_ids,
        source_path=payload.source_path,
        target_path=payload.target_path,
        sync_mode=payload.sync_mode,
        operator=admin.username,
        batch_size=payload.batch_size,
        concurrency=payload.concurrency,
    )


@router.post("/preview")
async def preview_sync(
    payload: SyncPreview,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """P1: Dry-run rsync --stats against one representative server.

    No data is transferred. Returns file-change statistics and optional
    per-file list (verbose=true, capped at 200 lines).

    Full-sync mode (--delete) shows files_deleted — use this to confirm
    before running a destructive full sync.
    """
    engine = await _get_sync_engine(request)
    result = await engine.preview_sync(
        server_id=payload.server_id,
        source_path=payload.source_path,
        target_path=payload.target_path,
        sync_mode=payload.sync_mode,
        verbose=payload.verbose,
    )
    if "error" in result:
        raise HTTPException(status_code=400, detail=result["error"])

    await _audit_sync(
        "sync_preview", "server", payload.server_id,
        f"Dry-run preview: {payload.source_path} → server #{payload.server_id} ({payload.sync_mode})",
        admin.username, request,
    )
    return result


@router.post("/file-ops")
async def file_operation(
    payload: FileOperation,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Remote file operation (delete / rename / mkdir) via SSH exec.

    All paths are shlex-quoted server-side. Dirs use rm -rf; confirm
    client-side before sending delete on a directory. Audit-logged.
    """
    import shlex
    from server.infrastructure.database.server_repo import ServerRepositoryImpl
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command

    session = request.state.db
    server = await ServerRepositoryImpl(session).get_by_id(payload.server_id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    op = payload.operation
    p = payload.path.rstrip("/") or "/"

    if op == "delete":
        # rm -rf covers both files and directories
        cmd = f"rm -rf {shlex.quote(p)}"
        audit_detail = f"删除: {p}"
    elif op == "rename":
        if not payload.new_path:
            raise HTTPException(status_code=400, detail="new_path required for rename")
        np = payload.new_path.rstrip("/") or "/"
        cmd = f"mv {shlex.quote(p)} {shlex.quote(np)}"
        audit_detail = f"重命名: {p} → {np}"
    elif op == "mkdir":
        cmd = f"mkdir -p {shlex.quote(p)}"
        audit_detail = f"新建目录: {p}"
    else:
        raise HTTPException(status_code=400, detail="Invalid operation")

    result = await exec_ssh_command(server, cmd, timeout=30)
    if result["exit_code"] != 0:
        raise HTTPException(
            status_code=400,
            detail=result["stderr"][:300] or f"Command failed (exit {result['exit_code']})"
        )

    await _audit_sync(
        f"file_{op}", "server", payload.server_id,
        f"{audit_detail} 于 {server.name}", admin.username, request,
    )
    return {"success": True, "operation": op, "path": p}


@router.post("/browse")
async def browse_directory(
    payload: SyncBrowse,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Browse remote directory listing via SSH"""
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
    from server.infrastructure.database.server_repo import ServerRepositoryImpl

    session = request.state.db
    server_id = payload.server_id
    path = payload.path

    if not server_id:
        raise HTTPException(status_code=400, detail="server_id required")

    repo = ServerRepositoryImpl(session)
    server = await repo.get_by_id(server_id)
    if not server:
        raise HTTPException(status_code=404, detail="Server not found")

    # Use ls -la for detailed listing, parse output
    import shlex
    result = await exec_ssh_command(server, f"ls -la {shlex.quote(path)}", timeout=10)

    # Audit
    await _audit_sync("browse_directory", "server", server_id, f"Browse {server.name}:{path}", admin.username, request)

    if result["exit_code"] != 0:
        return {"path": path, "entries": [], "error": result["stderr"][:500]}

    entries = []
    for line in result["stdout"].split("\n")[1:]:  # Skip "total" line
        if not line.strip():
            continue
        parts = line.split(None, 8)
        if len(parts) >= 9:
            is_dir = parts[0].startswith("d")
            entries.append({
                "name": parts[8],  # Last field after split(None, 8) — handles spaces in names
                "is_dir": is_dir,
                "size": parts[4],
                "perms": parts[0],
                "owner": parts[2],
                "modified": " ".join(parts[5:8]),
            })

    return {"path": path, "entries": entries}


@router.post("/browse-local")
async def browse_local_directory(
    payload: SyncBrowseLocal,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Browse a local directory on the Nexus server (for upload file manager).

    Only allows browsing under /tmp/nexus_upload_* directories.
    Returns entries with name, is_dir, size.
    """
    import os

    path = payload.path.rstrip("/") or "/"

    # Security: only allow browsing under /tmp/nexus_upload_*
    real_path = os.path.realpath(path)
    if not real_path.startswith("/tmp/nexus_upload_"):
        raise HTTPException(status_code=403, detail="仅允许浏览上传临时目录")

    if not os.path.isdir(real_path):
        raise HTTPException(status_code=404, detail="目录不存在")

    entries = []
    try:
        for entry in sorted(os.scandir(real_path), key=lambda e: (not e.is_dir(), e.name.lower())):
            try:
                stat = entry.stat()
                entries.append({
                    "name": entry.name,
                    "is_dir": entry.is_dir(),
                    "size": stat.st_size if not entry.is_dir() else 0,
                })
            except OSError:
                continue
    except PermissionError:
        raise HTTPException(status_code=403, detail="无权限访问该目录")

    return {"path": real_path, "entries": entries}


@router.post("/local-file-ops")
async def local_file_operation(
    payload: LocalFileOperation,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Delete or rename a file/directory in the local upload staging area.

    Only allows operations under /tmp/nexus_upload_* directories.
    """
    import os
    import shutil

    path = payload.path.rstrip("/") or "/"

    # Security: only allow operations under /tmp/nexus_upload_*
    real_path = os.path.realpath(path)
    if not real_path.startswith("/tmp/nexus_upload_"):
        raise HTTPException(status_code=403, detail="仅允许操作上传临时目录中的文件")

    if payload.operation == "delete":
        if not os.path.exists(real_path):
            raise HTTPException(status_code=404, detail="文件或目录不存在")
        try:
            if os.path.isdir(real_path):
                shutil.rmtree(real_path)
            else:
                os.unlink(real_path)
        except OSError as e:
            raise HTTPException(status_code=500, detail=f"删除失败: {e}")

        await _audit_sync(
            "local_file_delete", "sync", 0,
            f"删除: {path}", admin.username, request,
        )
        return {"success": True, "operation": "delete", "path": path}

    elif payload.operation == "rename":
        if not payload.new_name:
            raise HTTPException(status_code=400, detail="new_name 必填")
        # Sanitize: strip slashes and path traversal
        safe_name = payload.new_name.replace("/", "_").replace("\\", "_").strip()
        if not safe_name or safe_name == "." or safe_name == "..":
            raise HTTPException(status_code=400, detail="无效的新名称")

        if not os.path.exists(real_path):
            raise HTTPException(status_code=404, detail="文件或目录不存在")

        parent_dir = os.path.dirname(real_path)
        new_path = os.path.join(parent_dir, safe_name)

        # Security: new path must also be under upload dir
        if not os.path.realpath(new_path).startswith("/tmp/nexus_upload_"):
            raise HTTPException(status_code=403, detail="目标路径不在上传临时目录内")

        if os.path.exists(new_path):
            raise HTTPException(status_code=400, detail="目标名称已存在")

        try:
            os.rename(real_path, new_path)
        except OSError as e:
            raise HTTPException(status_code=500, detail=f"重命名失败: {e}")

        await _audit_sync(
            "local_file_rename", "sync", 0,
            f"重命名: {path} → {safe_name}", admin.username, request,
        )
        return {"success": True, "operation": "rename", "path": path, "new_name": safe_name}


@router.post("/local-file-preview")
async def local_file_preview(
    payload: LocalFilePreview,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Preview file content in the local upload staging area.

    Only allows files under /tmp/nexus_upload_* directories.
    Returns up to 4KB of file content, base64-encoded.
    """
    import os
    import base64

    path = payload.path.rstrip("/") or "/"

    # Security: only allow previewing files under /tmp/nexus_upload_*
    real_path = os.path.realpath(path)
    if not real_path.startswith("/tmp/nexus_upload_"):
        raise HTTPException(status_code=403, detail="仅允许预览上传临时目录中的文件")

    if not os.path.isfile(real_path):
        raise HTTPException(status_code=404, detail="文件不存在或不是普通文件")

    # Size check: skip files over 1MB
    file_size = os.path.getsize(real_path)
    if file_size > 1_048_576:
        raise HTTPException(status_code=400, detail="文件过大，不支持预览（>1MB）")

    # Read up to 4KB
    max_read = 4096
    try:
        with open(real_path, "rb") as f:
            raw = f.read(max_read)
    except OSError as e:
        raise HTTPException(status_code=500, detail=f"读取失败: {e}") from None

    truncated = file_size > max_read

    # Detect encoding: try UTF-8 decode
    encoding_hint = "binary"
    try:
        raw.decode("utf-8")
        encoding_hint = "text"
    except (UnicodeDecodeError, ValueError):
        pass

    name = os.path.basename(real_path)

    await _audit_sync(
        "local_file_preview", "sync", 0,
        f"预览: {path}", admin.username, request,
    )

    return {
        "name": name,
        "size": file_size,
        "content_base64": base64.b64encode(raw).decode("ascii"),
        "truncated": truncated,
        "encoding_hint": encoding_hint,
    }


async def _audit_sync(action: str, target_type: str, target_id: int, detail: str, operator: str, request: Request):
    """Create audit log entry for sync operations"""
    try:
        session = request.state.db
        audit_repo = AuditLogRepositoryImpl(session)
        await audit_repo.create(AuditLog(
            admin_username=operator,
            action=action,
            target_type=target_type,
            target_id=target_id,
            detail=detail,
            ip_address=request.client.host if request.client else "",
        ))
    except Exception:
        logger.warning(f"Audit log write failed for {action} on {target_type}/{target_id}", exc_info=True)


@router.post("/cancel")
async def cancel_sync(
    payload: SyncCancel,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Cancel an in-progress push by setting a Redis flag.

    The sync engine checks this flag before each rsync execution.
    Already-completed servers retain their results; pending ones are skipped.
    """
    try:
        from server.infrastructure.redis.client import get_redis
        redis = get_redis()
        await redis.set(f"sync:cancel:{payload.batch_id}", "1", ex=3600)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"取消操作失败: {e}")

    await _audit_sync(
        "sync_cancel", "sync", 0,
        f"取消推送: batch_id={payload.batch_id}", admin.username, request,
    )

    return {"cancelled": True, "batch_id": payload.batch_id}


# ── H5: Validate Source Path ──

_FORBIDDEN_PATH_PREFIXES = ("/etc", "/root", "/home", "/var", "/boot", "/dev", "/proc", "/sys", "/run", "/srv")


@router.post("/validate-source-path")
async def validate_source_path(
    payload: ValidateSourcePath,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Validate a local directory path on the Nexus server as a push source.

    Security: rejects sensitive system paths (/etc, /root, /home, etc.)
    and paths with traversal components.
    """
    import os

    path = payload.path.strip()

    # Reject path traversal
    if ".." in path.split("/"):
        raise HTTPException(status_code=400, detail="路径不允许包含 '..'")

    real_path = os.path.realpath(path)

    # Reject sensitive paths
    for prefix in _FORBIDDEN_PATH_PREFIXES:
        if real_path == prefix or real_path.startswith(prefix + "/"):
            raise HTTPException(status_code=403, detail=f"不允许使用系统路径: {prefix}")

    if not os.path.exists(real_path):
        raise HTTPException(status_code=404, detail="路径不存在")

    if not os.path.isdir(real_path):
        raise HTTPException(status_code=400, detail="路径不是目录")

    # Count files and total size
    file_count = 0
    total_size = 0
    for root, _dirs, fnames in os.walk(real_path):
        for fn in fnames:
            file_count += 1
            try:
                total_size += os.path.getsize(os.path.join(root, fn))
            except OSError:
                pass
            if file_count >= 10000:
                break
        if file_count >= 10000:
            break

    await _audit_sync(
        "validate_source_path", "sync", 0,
        f"验证源路径: {path} ({file_count} 文件)",
        admin.username, request,
    )

    return {
        "valid": True,
        "is_dir": True,
        "path": real_path,
        "file_count": file_count,
        "size_bytes": total_size,
        "file_count_truncated": file_count >= 10000,
    }


# ── H4: Diagnose Push Failure ──

@router.post("/diagnose")
async def diagnose_push(
    payload: SyncDiagnose,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Diagnose why a push failed for a specific server.

    Checks SSH connectivity, disk space, target path permissions, and write access.
    """
    import shlex
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
    from server.infrastructure.database.server_repo import ServerRepositoryImpl

    session = request.state.db
    server = await ServerRepositoryImpl(session).get_by_id(payload.server_id)
    if not server:
        raise HTTPException(status_code=404, detail="服务器不存在")

    result = {
        "server_id": payload.server_id,
        "server_name": server.name,
        "ssh_ok": False,
        "disk_avail": None,
        "disk_used_pct": None,
        "path_exists": None,
        "path_perms": None,
        "path_writable": None,
        "errors": [],
    }

    # 1. SSH connectivity
    try:
        r = await exec_ssh_command(server, "echo ok", timeout=10)
        if r["exit_code"] == 0 and "ok" in r["stdout"]:
            result["ssh_ok"] = True
        else:
            result["errors"].append(f"SSH 连接异常: {r['stderr'][:200] or 'exit code ' + str(r['exit_code'])}")
            # Cannot continue if SSH fails
            await _audit_sync("diagnose", "server", payload.server_id, "诊断: SSH不通", admin.username, request)
            return result
    except Exception as e:
        result["errors"].append(f"SSH 连接失败: {e}")
        await _audit_sync("diagnose", "server", payload.server_id, "诊断: SSH失败", admin.username, request)
        return result

    dest = payload.target_path or server.target_path or "/tmp/sync"
    safe_dest = shlex.quote(dest)

    # 2. Disk space
    try:
        r = await exec_ssh_command(server, f"df -h {safe_dest} | tail -1", timeout=10)
        if r["exit_code"] == 0:
            parts = r["stdout"].strip().split()
            if len(parts) >= 6:
                result["disk_avail"] = parts[3]  # e.g. "50G"
                try:
                    result["disk_used_pct"] = int(parts[4].replace("%", ""))  # e.g. 72
                except ValueError:
                    pass
    except Exception as e:
        logger.debug(f"Diagnose disk check failed for server {payload.server_id}: {e}")
    try:
        r = await exec_ssh_command(server, f"ls -ld {safe_dest} 2>&1", timeout=10)
        if r["exit_code"] == 0:
            result["path_exists"] = True
            parts = r["stdout"].strip().split()
            if parts:
                result["path_perms"] = parts[0]  # e.g. "drwxr-xr-x"
        else:
            result["path_exists"] = False
            result["errors"].append(f"目标路径不存在: {dest}")
    except Exception as e:
        logger.debug(f"Diagnose path check failed for server {payload.server_id}: {e}")

    # 4. Write test
    if result["path_exists"]:
        try:
            test_file = f"{dest.rstrip('/')}/.nexus_diag_test_$$"
            r = await exec_ssh_command(
                server,
                f"touch {shlex.quote(test_file)} && rm -f {shlex.quote(test_file)}",
                timeout=10,
            )
            result["path_writable"] = r["exit_code"] == 0
            if r["exit_code"] != 0:
                result["errors"].append(f"目标路径不可写: {r['stderr'][:200]}")
        except Exception as e:
            result["errors"].append(f"写入测试失败: {e}")
            result["path_writable"] = False
    else:
        result["path_writable"] = False

    await _audit_sync(
        "diagnose", "server", payload.server_id,
        f"诊断: SSH={'✓' if result['ssh_ok'] else '✗'} 路径={'✓' if result['path_exists'] else '✗'} 写入={'✓' if result['path_writable'] else '✗'}",
        admin.username, request,
    )

    return result


# ── H3: File Diff View ──

_MAX_DIFF_SIZE = 102_400  # 100 KB


@router.post("/file-diff")
async def file_diff(
    payload: FileSyncDiff,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Compare a local file with its remote counterpart to show push diff.

    Security: only allows diffing files under /tmp/nexus_upload_* source paths.
    Reads up to 100KB from each side. Returns unified diff lines.
    """
    import os
    import difflib
    import shlex
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
    from server.infrastructure.database.server_repo import ServerRepositoryImpl

    # Security: source_path must be under /tmp/nexus_upload_*
    real_source = os.path.realpath(payload.source_path)
    if not real_source.startswith("/tmp/nexus_upload_"):
        raise HTTPException(status_code=403, detail="仅允许对比上传临时目录中的文件")

    # Resolve local file path
    local_file = os.path.realpath(os.path.join(payload.source_path, payload.relative_path))
    if not local_file.startswith(real_source):
        raise HTTPException(status_code=403, detail="文件路径越界")

    if not os.path.isfile(local_file):
        return {
            "file_name": os.path.basename(payload.relative_path),
            "local_exists": False,
            "remote_exists": None,
            "local_size": 0,
            "remote_size": None,
            "diff_lines": [],
            "truncated": False,
            "error": "本地文件不存在",
        }

    # Read local file
    local_size = os.path.getsize(local_file)
    try:
        with open(local_file, "r", encoding="utf-8", errors="replace") as f:
            local_text = f.read(_MAX_DIFF_SIZE)
    except OSError as e:
        raise HTTPException(status_code=500, detail=f"读取本地文件失败: {e}") from None

    local_lines = local_text.splitlines(keepends=True)
    local_truncated = local_size > _MAX_DIFF_SIZE

    # Get server + target path
    session = request.state.db
    server = await ServerRepositoryImpl(session).get_by_id(payload.server_id)
    if not server:
        raise HTTPException(status_code=404, detail="服务器不存在")

    dest = payload.target_path or server.target_path or "/tmp/sync"
    remote_file = f"{dest.rstrip('/')}/{payload.relative_path}"

    # Read remote file
    remote_exists = False
    remote_size = None
    remote_lines = []
    remote_truncated = False

    try:
        r = await exec_ssh_command(
            server,
            f"wc -c {shlex.quote(remote_file)} 2>/dev/null && cat {shlex.quote(remote_file)}",
            timeout=15,
        )
        if r["exit_code"] == 0:
            remote_exists = True
            output = r["stdout"]
            # First line is wc -c output: "  12345 /path/file"
            first_newline = output.index("\n") if "\n" in output else 0
            wc_line = output[:first_newline].strip()
            try:
                remote_size = int(wc_line.split()[0])
            except (ValueError, IndexError):
                remote_size = None
            remote_text = output[first_newline + 1:]
            if remote_size is not None and remote_size > _MAX_DIFF_SIZE:
                remote_truncated = True
            remote_lines = remote_text.splitlines(keepends=True)
        else:
            # File doesn't exist on remote — treat as empty (all local lines are additions)
            remote_exists = False
    except Exception as e:
        # SSH failed — report error
        logger.warning(f"File diff remote read failed for server {payload.server_id}: {e}")
        remote_exists = None

    # Compute unified diff
    diff_lines = []
    for line in difflib.unified_diff(
        remote_lines, local_lines,
        fromfile=f"remote/{payload.relative_path}",
        tofile=f"local/{payload.relative_path}",
        lineterm="",
    ):
        if line.startswith("+") and not line.startswith("+++"):
            diff_lines.append({"type": "add", "content": line[1:]})
        elif line.startswith("-") and not line.startswith("---"):
            diff_lines.append({"type": "del", "content": line[1:]})
        elif line.startswith("@@"):
            diff_lines.append({"type": "header", "content": line})
        elif line.startswith(" "):
            diff_lines.append({"type": "ctx", "content": line[1:]})

    await _audit_sync(
        "file_diff", "server", payload.server_id,
        f"文件对比: {payload.relative_path}",
        admin.username, request,
    )

    return {
        "file_name": os.path.basename(payload.relative_path),
        "local_exists": True,
        "remote_exists": remote_exists,
        "local_size": local_size,
        "remote_size": remote_size,
        "diff_lines": diff_lines,
        "truncated": local_truncated or remote_truncated,
    }


# ── File Upload via SFTP ──

MAX_UPLOAD_FILE_SIZE = 104_857_600  # 100 MB


@router.post("/upload")
async def upload_file(
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Upload a file to a remote server via SFTP.

    Multipart form fields:
      - server_id: int (required)
      - remote_path: str (target directory, required)
      - file: UploadFile (required)

    The file is uploaded to {remote_path}/{filename} on the target server.
    If the file already exists, it will be overwritten.
    """
    import asyncssh
    from server.infrastructure.ssh.asyncssh_pool import ssh_pool

    form = await request.form()
    server_id_raw = form.get("server_id")
    remote_path_raw = form.get("remote_path", "/tmp")
    file = form.get("file")

    if not server_id_raw:
        raise HTTPException(status_code=400, detail="server_id 必填")
    try:
        server_id = int(server_id_raw)
    except (ValueError, TypeError):
        raise HTTPException(status_code=400, detail="server_id 必须为数字")

    if not file or not hasattr(file, "filename") or not file.filename:
        raise HTTPException(status_code=400, detail="请选择要上传的文件")

    # Read file content with size check
    content = await file.read()
    if len(content) > MAX_UPLOAD_FILE_SIZE:
        raise HTTPException(status_code=400, detail=f"文件大小超过限制 ({MAX_UPLOAD_FILE_SIZE // (1024*1024)}MB)")

    remote_path = str(remote_path_raw).strip()
    filename = file.filename

    # Security: path traversal prevention
    if ".." in remote_path.split("/") or ".." in filename.split("/"):
        raise HTTPException(status_code=400, detail="路径不允许包含 '..'")

    # Sanitize filename — strip slashes to prevent path injection
    safe_filename = filename.replace("/", "_").replace("\\", "_").strip()
    if not safe_filename:
        raise HTTPException(status_code=400, detail="文件名无效")

    # Get server from DB
    session = request.state.db
    server = await ServerRepositoryImpl(session).get_by_id(server_id)
    if not server:
        raise HTTPException(status_code=404, detail="服务器不存在")

    # Build full remote path
    full_remote_path = f"{remote_path.rstrip('/')}/{safe_filename}"

    # Connect via asyncssh pool and use SFTP
    try:
        conn = await ssh_pool.acquire(server)
        try:
            async with conn.create_sftp_client() as sftp:
                await sftp.put(io.BytesIO(content), full_remote_path)
        finally:
            await ssh_pool.release(server.id)
    except asyncssh.PermissionDenied:
        raise HTTPException(status_code=403, detail=f"SSH 权限不足，无法写入 {full_remote_path}")
    except asyncssh.SFTPError as e:
        raise HTTPException(status_code=400, detail=f"SFTP 上传失败: {e}")
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}")

    # Audit
    await _audit_sync(
        "file_upload", "server", server_id,
        f"上传 {safe_filename} ({len(content)} bytes) → {server.name}:{full_remote_path}",
        admin.username, request,
    )

    return {
        "success": True,
        "server_id": server_id,
        "remote_path": full_remote_path,
        "filename": safe_filename,
        "size": len(content),
    }


# ── ZIP Upload & Extract (for push workflow) ──

MAX_ZIP_FILE_SIZE = 524_288_000  # 500 MB


@router.post("/upload-zip")
async def upload_zip(
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Upload a ZIP file to Nexus, extract it, return the source path for rsync push.

    Multipart form fields:
      - file: UploadFile (required, must be .zip)

    Returns {"source_path": "/tmp/nexus_upload_xxx/", "file_count": N}
    """
    import os
    import uuid
    import zipfile
    import shutil

    form = await request.form()
    file = form.get("file")

    if not file or not hasattr(file, "filename") or not file.filename:
        raise HTTPException(status_code=400, detail="请选择 ZIP 文件")

    filename = file.filename
    if not filename.lower().endswith(".zip"):
        raise HTTPException(status_code=400, detail="仅支持 .zip 格式")

    # Read with size check
    content = await file.read()
    if len(content) > MAX_ZIP_FILE_SIZE:
        raise HTTPException(status_code=400, detail=f"文件大小超过限制 ({MAX_ZIP_FILE_SIZE // (1024*1024)}MB)")
    if len(content) == 0:
        raise HTTPException(status_code=400, detail="文件为空")

    # Create temp directory
    upload_id = uuid.uuid4().hex[:12]
    extract_dir = f"/tmp/nexus_upload_{upload_id}"
    zip_path = f"{extract_dir}.zip"

    try:
        os.makedirs(extract_dir, exist_ok=True)

        # Save ZIP
        with open(zip_path, "wb") as f:
            f.write(content)

        # Validate and extract with zip slip protection
        file_count = 0
        with zipfile.ZipFile(zip_path, "r") as zf:
            for info in zf.infolist():
                # Zip slip: ensure no path traversal
                real_path = os.path.realpath(os.path.join(extract_dir, info.filename))
                if not real_path.startswith(os.path.realpath(extract_dir)):
                    raise HTTPException(status_code=400, detail=f"ZIP 包含非法路径: {info.filename}")
                if info.is_dir():
                    continue
                file_count += 1

            zf.extractall(extract_dir)

        # Clean up ZIP file
        os.unlink(zip_path)

        if file_count == 0:
            shutil.rmtree(extract_dir, ignore_errors=True)
            raise HTTPException(status_code=400, detail="ZIP 文件为空（无文件）")

        # Build file listing (relative paths, capped at 500)
        files = []
        for root, dirs, fnames in os.walk(extract_dir):
            for fn in fnames:
                rel = os.path.relpath(os.path.join(root, fn), extract_dir)
                files.append(rel)
                if len(files) >= 500:
                    break
            if len(files) >= 500:
                break
        files_truncated = file_count > len(files)

        # Audit
        await _audit_sync(
            "upload_zip", "sync", 0,
            f"上传 ZIP: {filename} ({len(content)} bytes) → {extract_dir} ({file_count} 个文件)",
            admin.username, request,
        )

        return {
            "success": True,
            "source_path": extract_dir,
            "file_count": file_count,
            "files": files,
            "files_truncated": files_truncated,
            "filename": filename,
            "size": len(content),
        }

    except HTTPException:
        shutil.rmtree(extract_dir, ignore_errors=True)
        try:
            os.unlink(zip_path)
        except OSError:
            pass
        raise
    except zipfile.BadZipFile:
        shutil.rmtree(extract_dir, ignore_errors=True)
        try:
            os.unlink(zip_path)
        except OSError:
            pass
        raise HTTPException(status_code=400, detail="无效的 ZIP 文件")
    except Exception as e:
        shutil.rmtree(extract_dir, ignore_errors=True)
        try:
            os.unlink(zip_path)
        except OSError:
            pass
        raise HTTPException(status_code=500, detail=f"解压失败: {e}")


# ── Post-Push Verification (sha256sum comparison) ──

@router.post("/verify")
async def verify_sync(
    payload: SyncVerify,
    request: Request,
    admin: Admin = Depends(get_current_admin),
):
    """Verify pushed files by comparing sha256sums between source and target servers.

    Walks the local source directory and computes sha256 for each file,
    then runs sha256sum on each target server via SSH and compares.
    Returns per-server results: matched / missing / mismatched file lists.
    """
    import asyncio
    import hashlib
    import shlex
    from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
    from server.infrastructure.database.server_repo import ServerRepositoryImpl

    session = request.state.db

    if not os.path.isdir(payload.source_path):
        raise HTTPException(status_code=400, detail=f"源路径不存在: {payload.source_path}")

    # Build local file manifest: {relative_path: sha256hex}
    local_files = {}
    for root, dirs, fnames in os.walk(payload.source_path):
        for fn in fnames:
            full = os.path.join(root, fn)
            rel = os.path.relpath(full, payload.source_path)
            if len(local_files) >= payload.max_files:
                break
            try:
                h = hashlib.sha256()
                with open(full, "rb") as f:
                    for chunk in iter(lambda: f.read(8192), b""):
                        h.update(chunk)
                local_files[rel] = h.hexdigest()
            except OSError:
                continue
        if len(local_files) >= payload.max_files:
            break

    if not local_files:
        raise HTTPException(status_code=400, detail="源目录无文件可校验")

    # Verify against each server concurrently
    sem = asyncio.Semaphore(5)
    results = {}

    async def _verify_one(sid: int):
        async with sem:
            server = await ServerRepositoryImpl(session).get_by_id(sid)
            if not server:
                results[sid] = {"server_id": sid, "server_name": "", "error": "服务器不存在"}
                return

            dest = payload.target_path or server.target_path or "/tmp/sync"
            # Run sha256sum on remote server for the same relative paths
            # Build a file list for sha256sum to check
            file_list = " ".join(shlex.quote(f) for f in local_files.keys())
            cmd = f"cd {shlex.quote(dest)} && sha256sum {file_list} 2>/dev/null"

            try:
                r = await exec_ssh_command(server, cmd, timeout=60)
            except Exception as e:
                results[sid] = {
                    "server_id": sid, "server_name": server.name,
                    "error": f"SSH 连接失败: {e}",
                }
                return

            # Parse sha256sum output: "hash  filename"
            remote_files = {}
            if r["exit_code"] == 0:
                for line in r["stdout"].strip().split("\n"):
                    parts = line.split(None, 1)
                    if len(parts) == 2:
                        remote_files[parts[1]] = parts[0]

            matched = []
            missing = []
            mismatched = []
            for rel, local_hash in local_files.items():
                if rel not in remote_files:
                    missing.append(rel)
                elif remote_files[rel] != local_hash:
                    mismatched.append(rel)
                else:
                    matched.append(rel)

            results[sid] = {
                "server_id": sid,
                "server_name": server.name,
                "matched": len(matched),
                "missing": len(missing),
                "mismatched": len(mismatched),
                "missing_files": missing[:50],
                "mismatched_files": mismatched[:50],
            }

    await asyncio.gather(*[_verify_one(sid) for sid in payload.server_ids])

    # Audit
    await _audit_sync(
        "sync_verify", "sync", 0,
        f"推送校验: {payload.source_path} → {len(payload.server_ids)} 台服务器",
        admin.username, request,
    )

    return {"results": results, "total_local_files": len(local_files)}