Nexus/server/api/settings.py

"""Nexus — Settings & Schedules API Routes
Presentation layer — system settings + push schedules + password presets + audit logs.

Security: All write operations require JWT authentication.
Immutable settings (secret_key, api_key, encryption_key, database_url) cannot be modified via API.
Sensitive values are masked in GET responses.
"""

from typing import Optional
from pathlib import Path
import asyncio
import html
import logging
import ipaddress
import re
import socket

import bcrypt
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, Field

from server.api.dependencies import get_db
from server.api.auth_jwt import get_current_admin, require_current_admin
from server.api.schemas import ScheduleCreate, ScheduleUpdate, PresetCreate, PresetUpdate, SettingUpdatePayload, ApiKeyRevealRequest, SshKeyPresetCreate, SshKeyPresetUpdate
from server.infrastructure.database.setting_repo import SettingRepositoryImpl
from server.infrastructure.database.push_schedule_repo import PushScheduleRepositoryImpl, PushRetryJobRepositoryImpl
from server.infrastructure.database.password_preset_repo import PasswordPresetRepositoryImpl
from server.infrastructure.database.ssh_key_preset_repo import SshKeyPresetRepositoryImpl
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
from server.domain.models import PushSchedule, PushRetryJob, PasswordPreset, SshKeyPreset, AuditLog, Admin

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

from sqlalchemy.ext.asyncio import AsyncSession

# Settings that cannot be modified via API (encryption consistency)
IMMUTABLE_KEYS = {"secret_key", "api_key", "encryption_key", "database_url"}

# Settings whose values are masked in GET responses
SENSITIVE_KEYS = {
    "secret_key",
    "api_key",
    "encryption_key",
    "telegram_bot_token",
    "telegram_offline_bot_token",
}

# S-02: Whitelist of keys accepted by PUT /{key} (superset of DB_OVERRIDE_MAP + notify toggles)
MUTABLE_KEYS: set[str] = {
    "system_name", "system_title", "db_pool_size", "db_max_overflow",
    "redis_url", "heartbeat_timeout", "api_base_url",
    "cpu_alert_threshold", "mem_alert_threshold", "disk_alert_threshold",
    "telegram_bot_token", "telegram_chat_id",
    "telegram_offline_bot_token", "telegram_offline_chat_id",
    "login_allowlist_enabled", "login_subscription_url",
    "login_subscription_ips", "login_manual_ips", "login_allowed_ips",
    "notify_alert_cpu", "notify_alert_mem", "notify_alert_disk",
    "notify_alert_offline",
    "notify_recovery", "notify_time_drift",
    "notify_system_redis", "notify_system_mysql", "notify_restart",
    "notify_script_complete",
    "script_exec_complete_sound",  # 脚本执行完成浏览器提示音 (off|beep_short|beep_double|chime_soft)
    "push_complete_sound",  # 文件推送完成浏览器提示音 (off|beep_short|beep_double|chime_soft)
    "editor_theme",  # Monaco editor theme preference
    "theme",  # 全站主题 (dark/light)
}

SETTING_ENUM_VALIDATORS: dict[str, frozenset[str]] = {
    "script_exec_complete_sound": frozenset({"off", "beep_short", "beep_double", "chime_soft"}),
    "push_complete_sound": frozenset({"off", "beep_short", "beep_double", "chime_soft"}),
    "theme": frozenset({"dark", "light"}),
}

# GET /{key} when row not yet in MySQL (PUT upserts on first save)
SETTING_DEFAULTS: dict[str, str] = {
    "script_exec_complete_sound": "beep_double",
    "push_complete_sound": "beep_double",
    "theme": "dark",
}

# S-02: Per-key validation rules: (type, min, max) for int settings
SETTING_VALIDATORS: dict[str, tuple] = {
    "db_pool_size": (int, 1, 1000),
    "db_max_overflow": (int, 1, 1000),
    "heartbeat_timeout": (int, 10, 600),
    "cpu_alert_threshold": (int, 1, 100),
    "mem_alert_threshold": (int, 1, 100),
    "disk_alert_threshold": (int, 1, 100),
}


# S-01: SSRF protection — check if hostname resolves to private/link-local IP
async def _is_private_host(hostname: str) -> bool:
    """Check if hostname resolves to any private/loopback/link-local IP."""
    if not hostname:
        return True
    # Direct IP check
    try:
        ip = ipaddress.ip_address(hostname)
        return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved
    except ValueError:
        pass  # Not a literal IP, do DNS resolution
    # F-2: Async DNS resolution (non-blocking)
    try:
        loop = asyncio.get_running_loop()
        infos = await loop.getaddrinfo(hostname, None, family=socket.AF_UNSPEC, type=socket.SOCK_STREAM)
        for _family, _, _, _, sockaddr in infos:
            ip = ipaddress.ip_address(sockaddr[0])
            if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
                return True
    except (socket.gaierror, ValueError, OSError):
        return True  # DNS failure → block
    return False

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


async def _verify_reauth(db: AsyncSession, admin: Admin, password: str) -> None:
    """F-4: Shared re-auth helper for reveal endpoints. Raises 400 on failure."""
    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(password.encode(), current.password_hash.encode()):
        raise HTTPException(status_code=400, detail="当前密码错误")


def _validate_ip_list(ips: list[str]) -> None:
    """Validate IP/CIDR/domain list. Raises 400 on invalid entry."""
    for raw in ips:
        ip = raw.strip()
        if not ip:
            continue
        if not (_IP_RE.match(ip) or _DOMAIN_RE.match(ip)):
            raise HTTPException(status_code=400, detail=f"无效的 IP/CIDR/域名: {ip}")


# ── System Settings ──

def _format_setting_response(key: str, value: str | None, updated_at) -> dict:
    """Build setting dict; sensitive keys never return plaintext."""
    if key in SENSITIVE_KEYS and value:
        return {
            "key": key,
            "value": "",
            "value_set": True,
            "updated_at": str(updated_at),
        }
    return {
        "key": key,
        "value": value,
        "value_set": bool(value) if key in SENSITIVE_KEYS else None,
        "updated_at": str(updated_at),
    }


@router.get("/", response_model=list)
async def list_settings(admin: Admin = Depends(get_current_admin), db: AsyncSession = Depends(get_db)):
    """List all system settings (sensitive values masked)"""
    repo = SettingRepositoryImpl(db)
    settings = await repo.get_all()
    return [_format_setting_response(s.key, s.value, s.updated_at) for s in settings]


@router.get("/ip-allowlist", response_model=dict)
async def get_ip_allowlist(
    admin: Admin = Depends(get_current_admin),
):
    """Return full allowlist state."""
    from server.config import settings as _settings
    from server.background.ip_allowlist_refresh import get_last_refresh_time

    allowlist_on = (_settings.LOGIN_ALLOWLIST_ENABLED or "false").lower() in ("true","1","yes","on")
    sub_ips_raw = _settings.LOGIN_SUBSCRIPTION_IPS or ""
    manual_raw = _settings.LOGIN_MANUAL_IPS or ""
    sub_ips = [ip.strip() for ip in sub_ips_raw.split(",") if ip.strip()]
    manual_ips = [ip.strip() for ip in manual_raw.split(",") if ip.strip()]

    return {
        "enabled": allowlist_on,
        "subscription_url": _settings.LOGIN_SUBSCRIPTION_URL or "",
        "subscription_ips": sub_ips,
        "manual_ips": manual_ips,
        "subscription_count": len(sub_ips),
        "manual_count": len(manual_ips),
        "total_count": len(sub_ips) + len(set(manual_ips) - set(sub_ips)),
        "last_refresh": get_last_refresh_time(),
    }


# ── Bing Daily Wallpaper — download 12+ images, rotate hourly ──
_WALLPAPER_DIR = Path(__file__).resolve().parent.parent.parent / "web" / "app" / "wallpapers"
_MAX_WALLPAPERS = 12  # keep 12 images (12 hours rotation before repeat)
_MAX_AGE_DAYS = 7     # auto-clean older than 7 days

_wallpaper_sync_lock = asyncio.Lock()


def _cleanup_old_wallpapers() -> int:
    """Remove wallpapers older than 7 days. Returns count removed."""
    from datetime import datetime, timedelta, timezone
    cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=_MAX_AGE_DAYS)
    removed = 0
    for f in sorted(_WALLPAPER_DIR.glob("*.jpg")):
        try:
            mtime = datetime.fromtimestamp(f.stat().st_mtime, tz=timezone.utc).replace(tzinfo=None)
            if mtime < cutoff:
                f.unlink()
                removed += 1
        except OSError as e:
            logger.debug("Wallpaper cleanup skip %s: %s", f.name, e)
    return removed


def _cached_wallpaper_urls() -> dict:
    """List locally cached wallpaper URLs (no outbound HTTP)."""
    _WALLPAPER_DIR.mkdir(parents=True, exist_ok=True)
    all_files = sorted(
        _WALLPAPER_DIR.glob("*.jpg"),
        key=lambda f: f.stat().st_mtime,
        reverse=True,
    )
    urls = [
        f"/app/wallpapers/{f.name}"
        for f in sorted(all_files[:_MAX_WALLPAPERS], key=lambda f: f.name, reverse=True)
    ]
    return {"urls": urls, "count": len(urls)}


async def _sync_bing_wallpapers_from_network() -> int:
    """Download recent Bing images to local cache. Returns count downloaded."""
    import httpx

    _WALLPAPER_DIR.mkdir(parents=True, exist_ok=True)
    downloaded = 0
    try:
        async with httpx.AsyncClient(timeout=12.0) as client:
            resp = await client.get(
                "https://cn.bing.com/HPImageArchive.aspx",
                params={"format": "js", "idx": 0, "n": 8, "mkt": "zh-CN"},
            )
            data = resp.json()
            for img in (data.get("images") or []):
                img_path = img.get("url")
                if not img_path:
                    continue
                date_str = img.get("startdate", "") or img.get("fullstartdate", "")[:8] or ""
                if not date_str:
                    import re
                    m = re.search(r"OHR\.(\w+)_", img_path)
                    date_str = m.group(1) if m else f"unknown_{downloaded}"

                cached = _WALLPAPER_DIR / f"{date_str}.jpg"
                if cached.exists() and cached.stat().st_size > 1024:
                    continue

                try:
                    img_resp = await client.get(f"https://cn.bing.com{img_path}")
                    if img_resp.status_code == 200:
                        cached.write_bytes(img_resp.content)
                        downloaded += 1
                except httpx.HTTPError as e:
                    logger.debug("Bing image download failed: %s", e)
    except httpx.HTTPError as e:
        logger.warning("Bing wallpaper sync failed: %s", e)

    _cleanup_old_wallpapers()
    all_files = sorted(_WALLPAPER_DIR.glob("*.jpg"), key=lambda f: f.stat().st_mtime, reverse=True)
    for f in all_files[_MAX_WALLPAPERS:]:
        try:
            f.unlink()
        except OSError as e:
            logger.debug("Wallpaper prune skip %s: %s", f.name, e)
    return downloaded


async def _ensure_wallpaper_cache() -> None:
    """Populate local cache when empty (login page has no admin to trigger POST /sync)."""
    if _cached_wallpaper_urls()["count"] > 0:
        return
    async with _wallpaper_sync_lock:
        if _cached_wallpaper_urls()["count"] > 0:
            return
        await _sync_bing_wallpapers_from_network()


@router.get("/bing-wallpapers", response_model=dict)
async def bing_wallpapers():
    """Return cached wallpaper URLs; lazy-sync from Bing when cache is empty."""
    await _ensure_wallpaper_cache()
    return _cached_wallpaper_urls()


@router.post("/bing-wallpapers/sync", response_model=dict)
async def bing_wallpapers_sync(admin: Admin = Depends(require_current_admin)):
    """Admin-only: fetch latest Bing images into local cache."""
    downloaded = await _sync_bing_wallpapers_from_network()
    result = _cached_wallpaper_urls()
    result["downloaded"] = downloaded
    return result


# Keep the old single-image endpoint for backward compatibility
@router.get("/bing-wallpaper", response_model=dict)
async def bing_wallpaper():
    """Return today's cached wallpaper URL (public, read-only)."""
    await _ensure_wallpaper_cache()
    _WALLPAPER_DIR.mkdir(parents=True, exist_ok=True)
    from datetime import datetime, timezone

    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    cached = _WALLPAPER_DIR / f"{today}.jpg"
    if cached.exists() and cached.stat().st_size > 1024:
        return {"url": f"/app/wallpapers/{today}.jpg"}

    all_files = sorted(_WALLPAPER_DIR.glob("*.jpg"), key=lambda f: f.stat().st_mtime, reverse=True)
    if all_files:
        return {"url": f"/app/wallpapers/{all_files[0].name}"}
    return {"url": None}


@router.get("/{key}", response_model=dict)
async def get_setting(key: str, admin: Admin = Depends(get_current_admin), db: AsyncSession = Depends(get_db)):
    """Get a single setting by key (sensitive values masked)"""
    repo = SettingRepositoryImpl(db)
    value = await repo.get(key)
    if value is None:
        default = SETTING_DEFAULTS.get(key)
        if default is None:
            raise HTTPException(status_code=404, detail="Setting not found")
        return _format_setting_response(key, default, None)
    return _format_setting_response(key, value, None)


@router.post("/api-key/reveal", response_model=dict)
async def reveal_api_key(
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Reveal global API_KEY for authenticated admin (JWT required)."""
    repo = SettingRepositoryImpl(db)
    value = await repo.get("api_key")
    if value is None:
        raise HTTPException(status_code=404, detail="API_KEY not found")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="reveal_api_key",
        target_type="setting",
        detail="查看全局 API Key",
        ip_address=request.client.host if request.client else "",
    ))
    return {"key": "api_key", "value": value}


@router.put("/{key}", response_model=dict)
async def set_setting(
    key: str,
    payload: SettingUpdatePayload,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Set a system setting value (immutable keys are rejected)

    Writes to DB AND hot-reloads the in-memory settings object so changes
    take effect immediately without server restart.
    """
    if key in IMMUTABLE_KEYS:
        raise HTTPException(status_code=403, detail=f"Setting '{key}' is immutable and cannot be modified via API")
    if key not in MUTABLE_KEYS:
        raise HTTPException(status_code=403, detail=f"未知设置项: {key}")

    val_str = str(payload.value)
    if key in SENSITIVE_KEYS and not val_str.strip():
        repo = SettingRepositoryImpl(db)
        current = await repo.get(key)
        if not current:
            raise HTTPException(status_code=400, detail=f"'{key}' 不能为空")
        return _format_setting_response(key, current, None)

    if key in SETTING_ENUM_VALIDATORS:
        allowed = SETTING_ENUM_VALIDATORS[key]
        if val_str not in allowed:
            raise HTTPException(
                status_code=400,
                detail=f"'{key}' 必须是: {', '.join(sorted(allowed))}",
            )

    # S-02: Validate value type and range for int settings
    if key in SETTING_VALIDATORS:
        expected_type, min_val, max_val = SETTING_VALIDATORS[key]
        try:
            int_val = int(val_str)
        except (ValueError, TypeError):
            raise HTTPException(status_code=400, detail=f"'{key}' 必须是整数") from None
        if int_val < min_val or int_val > max_val:
            raise HTTPException(status_code=400, detail=f"'{key}' 范围: {min_val}-{max_val}")

    repo = SettingRepositoryImpl(db)
    setting = await repo.set(key, str(payload.value))

    # Hot-reload: update in-memory settings so changes take effect immediately
    import logging
    _logger = logging.getLogger("nexus.settings")
    from server.config import settings as _settings
    from server.infrastructure.settings_broadcast import publish_setting_change

    if _settings.apply_db_override(key, setting.value):
        await publish_setting_change(key, setting.value)

    # Audit log
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="update_setting",
        target_type="setting",
        detail=f"键={key}",
        ip_address=request.client.host if request.client else "",
    ))

    if key in ("telegram_bot_token", "telegram_chat_id"):
        try:
            from server.application.services.ops_patrol_sync import sync_telegram_to_env

            sync_telegram_to_env(_settings.TELEGRAM_BOT_TOKEN, _settings.TELEGRAM_CHAT_ID)
        except Exception as exc:
            _logger.warning("Layer-3 Telegram env sync failed after %s update: %s", key, exc)

    return {"key": setting.key, "value": setting.value}


# ── Push Schedules ──

schedule_router = APIRouter(prefix="/api/schedules", tags=["schedules"])


@schedule_router.get("/", response_model=list)
async def list_schedules(admin: Admin = Depends(get_current_admin), db: AsyncSession = Depends(get_db)):
    """List all push schedules"""
    repo = PushScheduleRepositoryImpl(db)
    schedules = await repo.get_all()
    return [_schedule_to_dict(s) for s in schedules]


@schedule_router.post("/", response_model=dict, status_code=201)
async def create_schedule(
    payload: ScheduleCreate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Create a new push schedule"""
    repo = PushScheduleRepositoryImpl(db)
    schedule = PushSchedule(**payload.model_dump())
    created = await repo.create(schedule)

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

    return _schedule_to_dict(created)


@schedule_router.put("/{id}", response_model=dict)
async def update_schedule(
    id: int,
    payload: ScheduleUpdate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Update a push schedule"""
    repo = PushScheduleRepositoryImpl(db)
    schedule = await repo.get_by_id(id)
    if not schedule:
        raise HTTPException(status_code=404, detail="Schedule not found")
    # Explicit allowlist to prevent mass-assignment to sensitive columns
    ALLOWED_FIELDS = {
        "name", "source_path", "target_path", "server_ids", "run_mode", "cron_expr",
        "fire_at", "sync_mode", "enabled", "schedule_type",
        "script_id", "script_content", "exec_timeout", "long_task",
    }
    for key, value in payload.model_dump(exclude_unset=True).items():
        if key in ALLOWED_FIELDS:
            setattr(schedule, key, value)
    updated = await repo.update(schedule)

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

    return _schedule_to_dict(updated)


@schedule_router.delete("/{id}", status_code=204)
async def delete_schedule(
    id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Delete a push schedule"""
    repo = PushScheduleRepositoryImpl(db)
    result = await repo.delete(id)
    if not result:
        raise HTTPException(status_code=404, detail="Schedule not found")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="delete_schedule",
        target_type="schedule",
        target_id=id,
        ip_address=request.client.host if request.client else "",
    ))


# ── Password Presets ──

preset_router = APIRouter(prefix="/api/presets", tags=["presets"])


@preset_router.get("/", response_model=list)
async def list_presets(admin: Admin = Depends(get_current_admin), db: AsyncSession = Depends(get_db)):
    """List all password presets"""
    repo = PasswordPresetRepositoryImpl(db)
    presets = await repo.get_all()
    return [
        {"id": p.id, "name": p.name, "username": p.username or "root", "created_at": str(p.created_at)}
        for p in presets
    ]


@preset_router.post("/", response_model=dict, status_code=201)
async def create_preset(
    payload: PresetCreate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Create a password preset (password will be encrypted)"""
    from server.infrastructure.database.crypto import encrypt_value
    repo = PasswordPresetRepositoryImpl(db)
    encrypted = encrypt_value(payload.encrypted_pw)
    preset = PasswordPreset(
        name=payload.name,
        username=(payload.username or "root").strip() or "root",
        encrypted_pw=encrypted,
    )
    created = await repo.create(preset)

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

    return {"id": created.id, "name": created.name}


@preset_router.put("/{id}", response_model=dict)
async def update_preset(
    id: int,
    payload: PresetUpdate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Update a password preset (name and/or password — password encrypted)"""
    from server.infrastructure.database.crypto import encrypt_value

    repo = PasswordPresetRepositoryImpl(db)
    preset = await repo.get_by_id(id)
    if not preset:
        raise HTTPException(status_code=404, detail="Preset not found")

    updates = payload.model_dump(exclude_unset=True)
    new_pw = updates.pop("encrypted_pw", None)
    if updates.get("name"):
        preset.name = updates["name"]
    if updates.get("username"):
        preset.username = updates["username"].strip() or "root"
    if new_pw:
        preset.encrypted_pw = encrypt_value(new_pw)

    await repo.update(preset)

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

    return {"id": preset.id, "name": preset.name}


@preset_router.post("/{id}/reveal", response_model=dict)
async def reveal_preset(
    id: int,
    payload: ApiKeyRevealRequest,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Reveal a password preset's decrypted value (re-auth + audit-logged)."""
    from server.infrastructure.database.crypto import decrypt_value
    await _verify_reauth(db, admin, payload.current_password)

    repo = PasswordPresetRepositoryImpl(db)
    preset = await repo.get_by_id(id)
    if not preset:
        raise HTTPException(status_code=404, detail="Preset not found")

    try:
        plaintext = decrypt_value(preset.encrypted_pw)
    except ValueError:
        raise HTTPException(status_code=500, detail="解密失败") from None

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

    return {"id": preset.id, "name": preset.name, "password": plaintext}


@preset_router.delete("/{id}", status_code=204)
async def delete_preset(
    id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Delete a password preset"""
    repo = PasswordPresetRepositoryImpl(db)
    result = await repo.delete(id)
    if not result:
        raise HTTPException(status_code=404, detail="Preset not found")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="delete_preset",
        target_type="preset",
        target_id=id,
        ip_address=request.client.host if request.client else "",
    ))


# ── SSH Key Presets ──

ssh_key_preset_router = APIRouter(prefix="/api/ssh-key-presets", tags=["ssh-key-presets"])


@ssh_key_preset_router.get("/", response_model=list)
async def list_ssh_key_presets(admin: Admin = Depends(get_current_admin), db: AsyncSession = Depends(get_db)):
    """List all SSH key presets"""
    repo = SshKeyPresetRepositoryImpl(db)
    presets = await repo.get_all()
    return [
        {
            "id": p.id,
            "name": p.name,
            "username": p.username or "root",
            "public_key": p.public_key or "",
            "private_key_set": bool(p.encrypted_private_key),
            "created_at": str(p.created_at),
        }
        for p in presets
    ]


@ssh_key_preset_router.post("/", response_model=dict, status_code=201)
async def create_ssh_key_preset(
    payload: SshKeyPresetCreate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Create an SSH key preset (private key will be encrypted)"""
    from server.infrastructure.database.crypto import encrypt_value
    from server.utils.ssh_key_pem import prepare_ssh_key_preset_fields

    repo = SshKeyPresetRepositoryImpl(db)
    try:
        private_pem, public_key = prepare_ssh_key_preset_fields(
            payload.private_key, payload.public_key
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    encrypted_private = encrypt_value(private_pem)
    preset = SshKeyPreset(
        name=payload.name,
        username=(payload.username or "root").strip() or "root",
        encrypted_private_key=encrypted_private,
        public_key=public_key,
    )
    created = await repo.create(preset)

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

    return {"id": created.id, "name": created.name}


@ssh_key_preset_router.put("/{id}", response_model=dict)
async def update_ssh_key_preset(
    id: int,
    payload: SshKeyPresetUpdate,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Update an SSH key preset (name and/or private key — private key encrypted)"""
    from server.infrastructure.database.crypto import encrypt_value

    repo = SshKeyPresetRepositoryImpl(db)
    preset = await repo.get_by_id(id)
    if not preset:
        raise HTTPException(status_code=404, detail="SSH key preset not found")

    updates = payload.model_dump(exclude_unset=True)
    new_private_key = updates.pop("private_key", None)
    if updates.get("name"):
        preset.name = updates["name"]
    if updates.get("username"):
        preset.username = updates["username"].strip() or "root"
    if new_private_key:
        from server.utils.ssh_key_pem import prepare_ssh_key_preset_fields

        try:
            private_pem, derived_public = prepare_ssh_key_preset_fields(
                new_private_key,
                updates.get("public_key") if "public_key" in updates else preset.public_key,
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        preset.encrypted_private_key = encrypt_value(private_pem)
        if "public_key" in updates or not (preset.public_key or "").strip():
            preset.public_key = derived_public
    elif "public_key" in updates:
        preset.public_key = updates["public_key"]

    await repo.update(preset)

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

    return {"id": preset.id, "name": preset.name}


@ssh_key_preset_router.post("/{id}/reveal", response_model=dict)
async def reveal_ssh_key_preset(
    id: int,
    payload: ApiKeyRevealRequest,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Reveal an SSH key preset's decrypted private key (re-auth + audit-logged)."""
    from server.infrastructure.database.crypto import decrypt_value
    await _verify_reauth(db, admin, payload.current_password)

    repo = SshKeyPresetRepositoryImpl(db)
    preset = await repo.get_by_id(id)
    if not preset:
        raise HTTPException(status_code=404, detail="SSH key preset not found")

    try:
        plaintext = decrypt_value(preset.encrypted_private_key)
    except ValueError:
        raise HTTPException(status_code=500, detail="解密失败") from None

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

    return {"id": preset.id, "name": preset.name, "private_key": plaintext, "public_key": preset.public_key or ""}


@ssh_key_preset_router.delete("/{id}", status_code=204)
async def delete_ssh_key_preset(
    id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Delete an SSH key preset"""
    repo = SshKeyPresetRepositoryImpl(db)
    result = await repo.delete(id)
    if not result:
        raise HTTPException(status_code=404, detail="SSH key preset not found")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="delete_ssh_key_preset",
        target_type="ssh_key_preset",
        target_id=id,
        ip_address=request.client.host if request.client else "",
    ))


# ── Alert History ──

alert_history_router = APIRouter(prefix="/api/alert-history", tags=["alerts"])


@alert_history_router.get("/", response_model=dict)
async def list_alert_history(
    server_id: Optional[int] = None,
    alert_type: Optional[str] = Query(None, alias="type"),
    is_recovery: Optional[bool] = None,
    status: Optional[str] = Query(None, description="active|recovered — maps to is_recovery"),
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    page: Optional[int] = Query(None, ge=1),
    per_page: Optional[int] = Query(None, ge=1, le=200),
    limit: int = Query(50, ge=1, le=200),
    offset: int = Query(0, ge=0),
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Paginated alert/recovery history with filters."""
    import math
    from sqlalchemy import select, func
    from server.domain.models import AlertLog, Server

    if per_page is not None:
        limit = per_page
    if page is not None and per_page is not None:
        offset = (page - 1) * per_page

    filters = []
    if server_id:
        filters.append(AlertLog.server_id == server_id)
    if alert_type:
        filters.append(AlertLog.alert_type == alert_type)
    recovery_filter = is_recovery
    if status == "active":
        recovery_filter = False
    elif status == "recovered":
        recovery_filter = True
    if recovery_filter is not None:
        filters.append(AlertLog.is_recovery == recovery_filter)
    from server.utils.display_time import beijing_day_range_utc, parse_operator_calendar_date

    try:
        if date_from:
            start_utc, _ = beijing_day_range_utc(parse_operator_calendar_date(date_from))
            filters.append(AlertLog.created_at >= start_utc)
        if date_to:
            _, end_utc = beijing_day_range_utc(parse_operator_calendar_date(date_to))
            filters.append(AlertLog.created_at <= end_utc)
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e)) from e

    count_q = select(func.count(AlertLog.id))
    data_q = (
        select(AlertLog, Server.name.label("srv_name"))
        .join(Server, AlertLog.server_id == Server.id, isouter=True)
        .order_by(AlertLog.created_at.desc())
        .offset(offset).limit(limit)
    )
    if filters:
        count_q = count_q.where(*filters)
        data_q = data_q.where(*filters)

    total = int((await db.execute(count_q)).scalar() or 0)
    rows = (await db.execute(data_q)).all()

    def _to_dict(log, srv_name):
        return {
            "id": log.id,
            "server_id": log.server_id,
            "server_name": log.server_name or srv_name or f"#{log.server_id}",
            "alert_type": log.alert_type,
            "value": log.value,
            "is_recovery": log.is_recovery,
            "created_at": str(log.created_at) if log.created_at else None,
        }

    current_page = (offset // limit) + 1 if limit else 1
    pages = math.ceil(total / limit) if limit and total > 0 else 1
    return {
        "items": [_to_dict(log, sn) for log, sn in rows],
        "total": total,
        "limit": limit,
        "offset": offset,
        "page": page if page is not None else current_page,
        "per_page": per_page if per_page is not None else limit,
        "pages": pages,
    }


@alert_history_router.get("/stats", response_model=dict)
async def alert_stats(
    days: int = 7,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Top alerting servers + daily counts for the last N days."""
    from sqlalchemy import func, select
    from server.domain.models import AlertLog

    # Top 5 servers by alert count
    top_q = (
        select(AlertLog.server_id, AlertLog.server_name, func.count(AlertLog.id).label("cnt"))
        .where(AlertLog.is_recovery == False)
        .group_by(AlertLog.server_id, AlertLog.server_name)
        .order_by(func.count(AlertLog.id).desc())
        .limit(5)
    )
    top_rows = (await db.execute(top_q)).all()

    # Top 5 alert types by count
    type_q = (
        select(AlertLog.alert_type, func.count(AlertLog.id).label("cnt"))
        .where(AlertLog.is_recovery == False)
        .group_by(AlertLog.alert_type)
        .order_by(func.count(AlertLog.id).desc())
        .limit(5)
    )
    type_rows = (await db.execute(type_q)).all()

    from datetime import date, datetime, timedelta, timezone
    from sqlalchemy import case

    days_val = max(1, int(days))
    cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days_val)
    daily_q = (
        select(
            func.date(AlertLog.created_at).label("day"),
            func.sum(case((AlertLog.is_recovery == False, 1), else_=0)).label("alerts"),
            func.sum(case((AlertLog.is_recovery == True, 1), else_=0)).label("recoveries"),
        )
        .where(AlertLog.created_at >= cutoff)
        .group_by(func.date(AlertLog.created_at))
        .order_by(func.date(AlertLog.created_at))
    )
    daily_rows = (await db.execute(daily_q)).all()

    from server.infrastructure.redis.client import get_redis
    import logging

    daily = [{"day": str(r.day), "alerts": int(r.alerts), "recoveries": int(r.recoveries)} for r in daily_rows]
    today_str = str(date.today())
    today_alerts = 0
    today_recovered = 0
    for row in daily:
        if row["day"] == today_str:
            today_alerts = row["alerts"]
            today_recovered = row["recoveries"]
            break

    active_count = 0
    try:
        redis = get_redis()
        cursor = 0
        while True:
            cursor, keys = await redis.scan(cursor, match="alerts:*", count=200)
            for key in keys:
                active_count += int(await redis.scard(key) or 0)
            if cursor == 0:
                break
    except Exception as e:
        logging.getLogger(__name__).warning("alert_stats: Redis active count failed: %s", e)

    top_servers = [
        {"server_id": r.server_id, "server_name": r.server_name or f"#{r.server_id}", "count": r.cnt}
        for r in top_rows
    ]
    top_server_name = top_servers[0]["server_name"] if top_servers else None

    return {
        "top_types": [{"alert_type": r.alert_type, "count": r.cnt} for r in type_rows],
        "top_servers": top_servers,
        "daily": daily,
        "today": today_alerts,
        "active": active_count,
        "recovered": today_recovered,
        "top_server": top_server_name,
    }


# ── Layer-3 Host Patrol (health_monitor.sh) ──

@router.get("/ops-patrol/status", response_model=dict)
async def ops_patrol_status(
    admin: Admin = Depends(get_current_admin),
):
    """Report Telegram readiness for Layer-3 cron patrol and env sync state."""
    from server.config import settings as _settings
    from server.application.services.ops_patrol_sync import read_telegram_env_status

    app_configured = bool(_settings.TELEGRAM_BOT_TOKEN and _settings.TELEGRAM_CHAT_ID)
    env_status = read_telegram_env_status()
    return {
        "telegram_app_configured": app_configured,
        "telegram_layer3_ready": app_configured,
        "telegram_env_synced": env_status["env_synced"],
        "env_status": env_status,
        "db_fallback_enabled": True,
        "cron_install_command": "sudo nx cron",
        "health_log_path": "/var/log/nexus_health.log",
    }


@router.post("/ops-patrol/sync-telegram", response_model=dict)
async def ops_patrol_sync_telegram(
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Sync DB Telegram credentials to .env files readable by health_monitor.sh."""
    from server.config import settings as _settings
    from server.application.services.ops_patrol_sync import (
        read_telegram_env_status,
        sync_telegram_to_env,
    )

    if not _settings.TELEGRAM_BOT_TOKEN or not _settings.TELEGRAM_CHAT_ID:
        raise HTTPException(status_code=400, detail="请先在上方保存 Bot Token 与 Chat ID")

    result = sync_telegram_to_env(_settings.TELEGRAM_BOT_TOKEN, _settings.TELEGRAM_CHAT_ID)
    if result.get("skipped"):
        raise HTTPException(status_code=400, detail="Telegram 未配置，无法同步")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="sync_patrol_telegram",
        target_type="setting",
        detail=f"app_env={result.get('app_env')} host_env_prod={result.get('host_env_prod')}",
        ip_address=request.client.host if request.client else "",
    ))

    env_status = read_telegram_env_status()
    return {
        "synced": result,
        "env_status": env_status,
        "telegram_layer3_ready": True,
    }


# ── Telegram Test + Chat ID Detection ──

async def _telegram_clear_webhook_if_set(client, base_url: str) -> None:
    """Remove active webhook so getUpdates can receive messages."""
    wh_resp = await client.get(f"{base_url}/getWebhookInfo")
    if wh_resp.status_code != 200:
        raise HTTPException(status_code=502, detail=f"Telegram getWebhookInfo 返回 {wh_resp.status_code}")
    wh_data = wh_resp.json()
    if not wh_data.get("ok"):
        desc = wh_data.get("description") or "getWebhookInfo 失败"
        raise HTTPException(status_code=502, detail=f"Telegram API: {desc}")
    if wh_data.get("result", {}).get("url"):
        del_resp = await client.post(
            f"{base_url}/deleteWebhook",
            params={"drop_pending_updates": "false"},
        )
        del_data = del_resp.json()
        if not del_data.get("ok"):
            desc = del_data.get("description") or "deleteWebhook 失败"
            raise HTTPException(status_code=502, detail=f"无法清除 Webhook: {desc}")


@router.post("/telegram/test", response_model=dict)
async def telegram_test_send(
    admin: Admin = Depends(get_current_admin),
):
    """Send a test Telegram message to verify Bot Token + Chat ID configuration."""
    import httpx
    from server.config import settings as _settings
    from server.infrastructure.telegram import TELEGRAM_API_BASE

    if not _settings.TELEGRAM_BOT_TOKEN:
        raise HTTPException(status_code=400, detail="TELEGRAM_BOT_TOKEN 未配置")
    if not _settings.TELEGRAM_CHAT_ID:
        raise HTTPException(status_code=400, detail="TELEGRAM_CHAT_ID 未配置，请先检测并填写")

    from server.utils.display_time import format_beijing_now

    now = format_beijing_now()
    sys_name = html.escape((_settings.SYSTEM_NAME or "Nexus").strip())
    message = (
        f"✅ <b>【Telegram 配置测试】消息发送成功</b>\n"
        f"平台：{sys_name}\n"
        f"具体情况：Bot Token 与 Chat ID 配置正确，Telegram 通道可用\n"
        f"将接收的通知类型：\n"
        f"• 子机 CPU/内存/磁盘超阈值告警与恢复\n"
        f"• 子机离线告警（可另配专用 Bot，见设置页）\n"
        f"• Redis / MySQL 连接异常与恢复\n"
        f"• 子机时钟偏差、文件推送结果\n"
        f"• Layer 3 宿主机巡检 /health 失败与自动重启\n"
        f"说明：可在设置页关闭单项 NOTIFY_* 开关以停止对应类型推送\n"
        f"时间：{now}"
    )

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.post(
                f"{TELEGRAM_API_BASE}/bot{_settings.TELEGRAM_BOT_TOKEN}/sendMessage",
                json={
                    "chat_id": _settings.TELEGRAM_CHAT_ID,
                    "text": message,
                    "parse_mode": "HTML",
                },
            )
            data = resp.json()
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"请求失败: {e}") from e

    if resp.status_code == 200 and data.get("ok"):
        return {"success": True}
    desc = data.get("description") or f"HTTP {resp.status_code}"
    raise HTTPException(status_code=502, detail=f"Telegram API: {desc}")


@router.post("/telegram/reveal-token", response_model=dict)
async def reveal_telegram_token(
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Reveal Telegram Bot Token for authenticated admin (JWT required)."""
    repo = SettingRepositoryImpl(db)
    value = await repo.get("telegram_bot_token")
    if value is None:
        raise HTTPException(status_code=404, detail="telegram_bot_token not found")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="reveal_telegram_token",
        target_type="setting",
        detail="查看 Telegram Bot Token",
        ip_address=request.client.host if request.client else "",
    ))
    return {"key": "telegram_bot_token", "value": value}


async def _telegram_fetch_chats(token: str) -> dict:
    import httpx
    from server.infrastructure.telegram import TELEGRAM_API_BASE

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            base_url = f"{TELEGRAM_API_BASE}/bot{token}"
            await _telegram_clear_webhook_if_set(client, base_url)
            resp = await client.get(
                f"{base_url}/getUpdates",
                params={"limit": 50},
            )
        if resp.status_code != 200:
            raise HTTPException(status_code=502, detail=f"Telegram API 返回 {resp.status_code}")
        data = resp.json()
        if not data.get("ok"):
            desc = data.get("description") or "getUpdates 失败"
            raise HTTPException(status_code=502, detail=f"Telegram API: {desc}")
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"请求失败: {e}") from e

    seen: set[int] = set()
    chats: list[dict] = []
    for update in data.get("result", []):
        msg = update.get("message") or update.get("channel_post") or {}
        chat = msg.get("chat", {})
        cid = chat.get("id")
        if cid and cid not in seen:
            seen.add(cid)
            chats.append({
                "id": cid,
                "type": chat.get("type", ""),
                "title": chat.get("title") or chat.get("first_name") or "",
                "username": chat.get("username") or "",
            })
        if len(chats) >= 5:
            break

    return {"chats": chats, "count": len(chats)}


@router.get("/telegram/chats", response_model=dict)
async def telegram_get_chats(
    admin: Admin = Depends(get_current_admin),
):
    """Call getUpdates to detect recent chats and extract unique chat_ids.

    User must send at least one message to the Bot before calling this.
    Returns up to 5 distinct chats from recent messages.
    """
    from server.config import settings as _settings

    token = (_settings.TELEGRAM_BOT_TOKEN or "").strip()
    if not token:
        raise HTTPException(status_code=400, detail="TELEGRAM_BOT_TOKEN 未配置")
    return await _telegram_fetch_chats(token)


def _offline_telegram_token_or_400() -> str:
    from server.config import settings as _settings

    token = (_settings.TELEGRAM_OFFLINE_BOT_TOKEN or "").strip()
    if not token:
        raise HTTPException(status_code=400, detail="telegram_offline_bot_token 未配置")
    return token


def _offline_telegram_credentials_or_400() -> tuple[str, str]:
    """Resolve offline-dedicated Bot; 400 if not fully configured."""
    from server.config import settings as _settings

    token = _offline_telegram_token_or_400()
    chat_id = (_settings.TELEGRAM_OFFLINE_CHAT_ID or "").strip()
    if not chat_id:
        raise HTTPException(status_code=400, detail="telegram_offline_chat_id 未配置，请先检测并填写")
    return token, chat_id


@router.post("/telegram/offline/test", response_model=dict)
async def telegram_offline_test_send(
    admin: Admin = Depends(get_current_admin),
):
    """Send a test message via offline-dedicated Telegram Bot (not the default channel)."""
    import httpx
    from server.config import settings as _settings
    from server.infrastructure.telegram import TELEGRAM_API_BASE
    from server.utils.display_time import format_beijing_now

    token, chat_id = _offline_telegram_credentials_or_400()
    now = format_beijing_now()
    sys_name = html.escape((_settings.SYSTEM_NAME or "Nexus").strip())
    message = (
        f"✅ <b>【离线告警 Telegram 测试】消息发送成功</b>\n"
        f"平台：{sys_name}\n"
        f"具体情况：离线专用 Bot Token 与 Chat ID 配置正确\n"
        f"将接收的通知：子机由在线变为离线时的 Telegram 推送（须开启「子机离线告警」开关）\n"
        f"说明：未配置离线专用 Bot 时，离线告警回退至默认 Telegram 通道\n"
        f"时间：{now}"
    )

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.post(
                f"{TELEGRAM_API_BASE}/bot{token}/sendMessage",
                json={
                    "chat_id": chat_id,
                    "text": message,
                    "parse_mode": "HTML",
                },
            )
            data = resp.json()
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"请求失败: {e}") from e

    if resp.status_code == 200 and data.get("ok"):
        return {"success": True}
    desc = data.get("description") or f"HTTP {resp.status_code}"
    raise HTTPException(status_code=502, detail=f"Telegram API: {desc}")


@router.post("/telegram/offline/reveal-token", response_model=dict)
async def reveal_offline_telegram_token(
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Reveal offline-dedicated Telegram Bot Token (JWT + audit)."""
    repo = SettingRepositoryImpl(db)
    value = await repo.get("telegram_offline_bot_token")
    if value is None:
        raise HTTPException(status_code=404, detail="telegram_offline_bot_token not found")

    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="reveal_offline_telegram_token",
        target_type="setting",
        detail="查看离线告警 Telegram Bot Token",
        ip_address=request.client.host if request.client else "",
    ))
    return {"key": "telegram_offline_bot_token", "value": value}


@router.get("/telegram/offline/chats", response_model=dict)
async def telegram_offline_get_chats(
    admin: Admin = Depends(get_current_admin),
):
    """Detect chats for offline-dedicated Bot via getUpdates."""
    token = _offline_telegram_token_or_400()
    return await _telegram_fetch_chats(token)


# ── IP Allowlist ──

_IP_RE = re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(/\d{1,2})?$")
_DOMAIN_RE = re.compile(r"^[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z]{2,})+$")


class IpAllowlistRequest(BaseModel):
    ips: list[str] = Field(..., min_length=1, max_length=100, description="IP or hostname list")


class SubscriptionParseRequest(BaseModel):
    url: str = Field(..., min_length=1, description="Proxy subscription URL to fetch and parse")


@router.post("/ip-allowlist/parse-subscription", response_model=dict)
async def parse_subscription(
    payload: SubscriptionParseRequest,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Fetch a proxy subscription URL, decode it, and extract node host/IP list.

    Supports SS, VMess, VLESS, Trojan, Hysteria2 formats.
    Returns the parsed host list for the user to review before saving.
    """
    from urllib.parse import urlparse
    import httpx
    from server.infrastructure.subscription_parser import parse_subscription_content

    url = payload.url.strip()
    parsed = urlparse(url)
    if parsed.scheme not in ("http", "https"):
        raise HTTPException(status_code=400, detail="URL 必须以 http:// 或 https:// 开头")

    # S-01: SSRF protection — block private/link-local/loopback IPs
    hostname = parsed.hostname or ""
    if await _is_private_host(hostname):
        raise HTTPException(status_code=400, detail="不允许访问内网地址")

    MAX_BODY = 1_000_000  # 1 MB
    MAX_REDIRECTS = 3
    try:
        async with httpx.AsyncClient(timeout=15.0, follow_redirects=False) as client:
            resp = await client.get(url)
            # Manual redirect following with SSRF check on each hop
            redirects = 0
            while resp.is_redirect and redirects < MAX_REDIRECTS:
                redirects += 1
                location = resp.headers.get("location", "")
                if not location:
                    break
                # Resolve relative redirects
                redirect_url = str(httpx.URL(location).resolve_with(url))
                redir_parsed = urlparse(redirect_url)
                redir_host = redir_parsed.hostname or ""
                if await _is_private_host(redir_host):
                    raise HTTPException(status_code=400, detail="订阅链接重定向到内网地址，已阻止")
                url = redirect_url
                resp = await client.get(redirect_url)

            if resp.status_code != 200:
                raise HTTPException(status_code=400, detail=f"订阅链接返回 HTTP {resp.status_code}")
            # S-01: Limit response body size
            if len(resp.content) > MAX_BODY:
                raise HTTPException(status_code=400, detail="订阅内容超过 1MB 限制")
            raw = resp.text
    except httpx.TimeoutException:
        raise HTTPException(status_code=400, detail="请求订阅链接超时（15s）") from None
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"获取订阅失败: {e}") from e

    hosts = parse_subscription_content(raw)
    if not hosts:
        raise HTTPException(status_code=400, detail="未能从订阅内容中解析出任何节点 IP / 域名")

    # Audit
    ip_address = request.client.host if request.client else ""
    audit_repo = AuditLogRepositoryImpl(db)
    from server.domain.models import AuditLog
    await audit_repo.create(AuditLog(
        admin_username=admin.username, action="parse_subscription",
        target_type="setting", detail=f"解析订阅: {url[:80]} → {len(hosts)} 个节点",
        ip_address=ip_address,
    ))

    return {"hosts": hosts, "count": len(hosts)}


class IpAllowlistSaveRequest(BaseModel):
    manual_ips: list[str] = Field(default_factory=list)
    subscription_url: Optional[str] = None   # None = don't change; "" = clear
    enabled: Optional[bool] = None           # None = don't change


def _allowlist_save_response(*, manual_count: int, refresh, has_subscription: bool) -> dict:
    """Build POST /ip-allowlist response with subscription refresh outcome."""
    if not has_subscription:
        return {
            "success": True,
            "manual_count": manual_count,
            "subscription_ips": [],
            "subscription_count": 0,
            "last_refresh": None,
            "refresh_ok": True,
        }
    return {
        "success": True,
        "manual_count": manual_count,
        "subscription_ips": list(refresh.subscription_ips),
        "subscription_count": refresh.subscription_count,
        "last_refresh": refresh.last_refresh or None,
        "refresh_ok": refresh.ok,
        "refresh_error": refresh.error if not refresh.ok else None,
    }


@router.post("/ip-allowlist", response_model=dict)
async def set_ip_allowlist(
    payload: IpAllowlistSaveRequest,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Save manual IPs, subscription URL, and/or enabled toggle."""
    from server.config import settings as _settings
    from server.background.ip_allowlist_refresh import RefreshResult, do_refresh
    from server.infrastructure.settings_broadcast import publish_setting_changes

    if payload.manual_ips:
        _validate_ip_list(payload.manual_ips)

    repo = SettingRepositoryImpl(db)
    manual_cleaned = [ip.strip() for ip in payload.manual_ips if ip.strip()]
    manual_val = ",".join(manual_cleaned)
    await repo.set("login_manual_ips", manual_val)
    changes: dict[str, str] = {"login_manual_ips": manual_val}

    if payload.enabled is not None:
        val = "true" if payload.enabled else "false"
        await repo.set("login_allowlist_enabled", val)
        changes["login_allowlist_enabled"] = val

    if payload.subscription_url is not None:
        sub_url = payload.subscription_url.strip()
        await repo.set("login_subscription_url", sub_url)
        changes["login_subscription_url"] = sub_url

    if payload.subscription_url is not None:
        sub_url_now = payload.subscription_url.strip()
    else:
        sub_url_now = (_settings.LOGIN_SUBSCRIPTION_URL or "").strip()

    refresh = RefreshResult(ok=True)
    if not sub_url_now:
        await repo.set("login_allowed_ips", manual_val)
        await repo.set("login_subscription_ips", "")
        changes["login_allowed_ips"] = manual_val
        changes["login_subscription_ips"] = ""

    await publish_setting_changes(changes)

    if sub_url_now:
        refresh = await do_refresh()

    audit_repo = AuditLogRepositoryImpl(db)
    from server.domain.models import AuditLog
    ip_address = request.client.host if request.client else ""
    enabled_str = "" if payload.enabled is None else f"; 白名单开关={'开' if payload.enabled else '关'}"
    await audit_repo.create(AuditLog(
        admin_username=admin.username, action="update_ip_allowlist",
        target_type="setting",
        detail=f"手动IP {len(manual_cleaned)} 条{enabled_str}",
        ip_address=ip_address,
    ))

    return _allowlist_save_response(
        manual_count=len(manual_cleaned),
        refresh=refresh,
        has_subscription=bool(sub_url_now),
    )


@router.post("/ip-allowlist/toggle", response_model=dict)
async def toggle_ip_allowlist(
    request: Request,
    enabled: bool,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Quick toggle for allowlist enforcement (no IP list changes)."""
    val = "true" if enabled else "false"
    repo = SettingRepositoryImpl(db)
    await repo.set("login_allowlist_enabled", val)

    from server.infrastructure.settings_broadcast import publish_setting_changes
    await publish_setting_changes({"login_allowlist_enabled": val})

    audit_repo = AuditLogRepositoryImpl(db)
    from server.domain.models import AuditLog
    ip_address = request.client.host if request.client else ""
    await audit_repo.create(AuditLog(
        admin_username=admin.username, action="toggle_ip_allowlist",
        target_type="setting",
        detail=f"登录IP白名单: {'启用' if enabled else '关闭'}",
        ip_address=ip_address,
    ))
    return {"success": True, "enabled": enabled}


@router.post("/ip-allowlist/manual", response_model=dict)
async def add_manual_ips(
    payload: IpAllowlistRequest,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Append IPs to the manual list and trigger allowlist rebuild."""
    from server.config import settings as _settings
    _validate_ip_list(payload.ips)

    existing = [ip.strip() for ip in (_settings.LOGIN_MANUAL_IPS or "").split(",") if ip.strip()]
    new_ips = [ip.strip() for ip in payload.ips if ip.strip()]
    merged = list(dict.fromkeys(existing + new_ips))
    val = ",".join(merged)
    repo = SettingRepositoryImpl(db)
    await repo.set("login_manual_ips", val)

    from server.infrastructure.settings_broadcast import publish_setting_changes
    from server.background.ip_allowlist_refresh import do_refresh
    await publish_setting_changes({"login_manual_ips": val})
    await do_refresh()

    # S-07: Audit log
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="add_manual_ips",
        target_type="setting",
        detail=f"添加 {len(new_ips)} 条手动 IP: {', '.join(new_ips[:5])}",
        ip_address=request.client.host if request.client else "",
    ))

    return {"success": True, "manual_ips": merged}


@router.delete("/ip-allowlist/ip", response_model=dict)
async def remove_allowlist_ip(
    ip: str,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Remove one IP from the manual list and rebuild."""
    from server.config import settings as _settings

    existing = [i.strip() for i in (_settings.LOGIN_MANUAL_IPS or "").split(",") if i.strip()]
    remaining = [i for i in existing if i != ip.strip()]
    val = ",".join(remaining)
    repo = SettingRepositoryImpl(db)
    await repo.set("login_manual_ips", val)

    from server.infrastructure.settings_broadcast import publish_setting_changes
    from server.background.ip_allowlist_refresh import do_refresh
    await publish_setting_changes({"login_manual_ips": val})
    await do_refresh()

    # S-07: Audit log
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="remove_allowlist_ip",
        target_type="setting",
        detail=f"移除白名单 IP: {ip}",
        ip_address=request.client.host if request.client else "",
    ))

    return {"success": True, "removed": ip, "remaining": remaining}


# ── Audit Logs ──

audit_router = APIRouter(prefix="/api/audit", tags=["audit"])


@audit_router.get("/", response_model=dict)
async def list_audit_logs(
    action: Optional[str] = None,
    admin_username: Optional[str] = None,
    date_from: Optional[str] = None,   # ISO date: YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS
    date_to: Optional[str] = None,
    limit: int = 50,
    offset: int = 0,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """List audit logs with pagination, action filter, admin filter, and date range."""
    repo = AuditLogRepositoryImpl(db)
    try:
        logs, total = await repo.query(
            action=action,
            admin_username=admin_username,
            date_from=date_from,
            date_to=date_to,
            limit=limit,
            offset=offset,
        )
    except ValueError as e:
        raise HTTPException(status_code=422, detail=str(e)) from e
    return {
        "total": total,
        "limit": limit,
        "offset": offset,
        "items": await _audit_items_enriched(db, logs),
    }


async def _audit_items_enriched(db, logs: list[AuditLog]) -> list[dict]:
    """Attach target_name for server rows (audit UI shows name instead of #id)."""
    from sqlalchemy import select

    from server.domain.models import Server

    server_ids = {
        int(log.target_id)
        for log in logs
        if log.target_type == "server" and log.target_id
    }
    name_by_id: dict[int, str] = {}
    if server_ids:
        rows = (
            await db.execute(
                select(Server.id, Server.name).where(Server.id.in_(server_ids))
            )
        ).all()
        name_by_id = {int(sid): (name or f"server-{sid}") for sid, name in rows}

    items: list[dict] = []
    for log in logs:
        row = _audit_to_dict(log)
        if log.target_type == "server" and log.target_id:
            tid = int(log.target_id)
            row["target_name"] = name_by_id.get(tid)
        items.append(row)
    return items


# ── Retry Jobs ──

retry_router = APIRouter(prefix="/api/retries", tags=["retries"])


@retry_router.get("/", response_model=list)
async def list_retry_jobs(
    limit: int = 100,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """List all retry jobs (not just pending)"""
    repo = PushRetryJobRepositoryImpl(db)
    jobs = await repo.get_all(limit)
    return [_retry_to_dict(j) for j in jobs]


@retry_router.post("/{id}/retry", response_model=dict)
async def retry_job(
    id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Manually retry a failed/pending job"""
    repo = PushRetryJobRepositoryImpl(db)
    job = await repo.get_by_id(id)
    if not job:
        raise HTTPException(status_code=404, detail="Retry job not found")

    # Reset the job to pending state for immediate retry
    job = await repo.update_status(
        id, "pending",
        retry_count=0,
        next_retry_at=None,
        last_error=None,
    )

    # Audit log
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="retry_job",
        target_type="retry",
        target_id=id,
        detail=f"手动重试任务 #{id}",
        ip_address=request.client.host if request.client else "",
    ))

    return _retry_to_dict(job)


@retry_router.delete("/{id}", status_code=204)
async def delete_retry_job(
    id: int,
    request: Request,
    admin: Admin = Depends(get_current_admin),
    db: AsyncSession = Depends(get_db),
):
    """Delete a retry job"""
    repo = PushRetryJobRepositoryImpl(db)
    job = await repo.get_by_id(id)
    if not job:
        raise HTTPException(status_code=404, detail="Retry job not found")

    # Audit log
    audit_repo = AuditLogRepositoryImpl(db)
    await audit_repo.create(AuditLog(
        admin_username=admin.username,
        action="delete_retry",
        target_type="retry",
        target_id=id,
        detail=f"删除重试任务 #{id}",
        ip_address=request.client.host if request.client else "",
    ))

    await repo.delete(id)


# ── Helper functions ──

def _schedule_to_dict(schedule: PushSchedule) -> dict:
    from server.background.schedule_runner import compute_schedule_next_run

    next_run_dt = compute_schedule_next_run(schedule)
    return {
        "id": schedule.id,
        "name": schedule.name,
        "source_path": schedule.source_path,
        "target_path": getattr(schedule, "target_path", None),
        "server_ids": schedule.server_ids,
        "run_mode": getattr(schedule, "run_mode", None) or "once",
        "cron_expr": schedule.cron_expr,
        "fire_at": str(schedule.fire_at) if getattr(schedule, "fire_at", None) else None,
        "sync_mode": schedule.sync_mode or "incremental",
        "enabled": schedule.enabled,
        "schedule_type": getattr(schedule, "schedule_type", None) or "push",
        "script_id": getattr(schedule, "script_id", None),
        "script_content": getattr(schedule, "script_content", None),
        "exec_timeout": getattr(schedule, "exec_timeout", 60),
        "long_task": getattr(schedule, "long_task", False),
        "last_run_at": str(schedule.last_run_at) if schedule.last_run_at else None,
        "next_run": str(next_run_dt) if next_run_dt else None,
        "created_at": str(schedule.created_at) if schedule.created_at else None,
    }


def _audit_to_dict(log: AuditLog) -> dict:
    return {
        "id": log.id,
        "admin_username": log.admin_username,
        "action": log.action,
        "target_type": log.target_type,
        "target_id": log.target_id,
        # 前端历史字段别名
        "resource_type": log.target_type,
        "resource_id": log.target_id,
        "detail": log.detail,
        "ip_address": log.ip_address,
        "created_at": str(log.created_at) if log.created_at else None,
    }


def _retry_to_dict(job: PushRetryJob) -> dict:
    return {
        "id": job.id,
        "server_id": job.server_id,
        "server_name": job.server_name,
        "operator": job.operator,
        "source_path": job.source_path,
        "target_path": job.target_path,
        "status": job.status,
        "retry_count": job.retry_count,
        "max_retries": job.max_retries,
        "next_retry_at": str(job.next_retry_at) if job.next_retry_at else None,
        "last_error": job.last_error,
        "created_at": str(job.created_at) if job.created_at else None,
    }