"""
AstroHub v2.0 - 统一路由聚合

整合 M1-M11 所有模块路由到 /api/v1/ 下:
- /api/v1/health            - 全局健康检查
- /api/v1/discovery/sadp    - SADP 设备发现
- /api/v1/devices/*         - 设备管理(真实)
- /api/v1/ptz/*             - PTZ 控制(真实)
- /api/v1/streams/*         - 流服务
- /api/v1/calibration/*     - 校准
- /api/v1/ascom/*           - ASCOM 设备
- /api/v1/settings          - 系统设置

Author: 雅痞张@南方天文
"""

from __future__ import annotations

import asyncio
import concurrent.futures
import time
from datetime import UTC, datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING, Any

from fastapi import APIRouter, Query, Request
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel

# v10.55: PTZ 专用线程池 — 避免与 ISAPI proxy / 录像 / SADP 等共享默认线程池
# v1.1.25: 单 FIFO 命令通道 — move/stop/home 全部经单线程池提交，
# 全局到达顺序=提交顺序（单设备独占架构下无并发损失）。
# 根治 v10.59/v11.21 双线程池"只互斥不排序"导致 stop 反超 move、
# continuous 成为最后指令的设备失控漂移。
_ptz_executor = concurrent.futures.ThreadPoolExecutor(
    max_workers=1, thread_name_prefix="PTZ-SEQ"
)
# v10.59 名称保留（同通道别名），避免其余调用点遗漏改造
_ptz_stop_executor = _ptz_executor

if TYPE_CHECKING:
    from src.advanced.monitor import SystemMonitor

# v8.80: 三星校准
from src.advanced.calibration_3stars import CalibrationSolver, CoordinateConverter

# v9.75: 点击指向
from src.advanced.click_point import (
    MIN_CALIBRATION_DISTANCE_DEG,
    finish_session,
    get_session,
    reset_session,
)
from src.advanced.device_path import normalize_mac

# v8.102: SkyAtlas 天体目标选择
from src.advanced.skyatlas import coord_convert_service, get_skyatlas, skyatlas_goto_service
from src.astro_move.astro_goto import AstroGoto
from src.astro_move.astro_tracking import (
    get_tracking_status,
    set_tracking_rate,
    start_tracking,
    stop_tracking,
)
from src.astro_move.rd2az import CelestialResolver, get_staralign_manager

# v9.35: 高度曲线图
from src.console.altitude_chart import get_altitude_chart_data
from src.core.device_manager import DeviceManager
from src.core.device_reader import get_device_reader
from src.core.ptz_controller import PTZDeviceController

# ================================================================ #
#  核心模块导入
# ================================================================ #
from src.main.constants import VERSION, VERSION_NUM

# v6.40: 操作日志
from src.operation_logger import log_api, log_error, log_info, log_warn
from src.ptz.isapi.client import ISAPIClient
from src.ptz.isapi.ptz import ISAPIPTZController

# ================================================================ #
#  路由前缀定义
# ================================================================ #

API_V1_PREFIX = "/api/v1"

# 主 API 路由器
api_router = APIRouter(prefix=API_V1_PREFIX, tags=["API v1"])

# 健康检查路由器
health_router = APIRouter(prefix="/api/v1", tags=["System"])


# ================================================================ #
#  v6.40: 操作日志 API
# ================================================================ #

class OperationLogRequest(BaseModel):
    """操作日志请求."""
    level: str = "INFO"
    module: str = "web"
    action: str = ""
    detail: str | dict = ""

@health_router.post("/log/operation", summary="记录操作日志")
async def log_operation_api(req: OperationLogRequest) -> dict:
    """前端发送操作日志到后端."""
    from src.operation_logger import log_operation
    log_operation(req.level, req.module, req.action, req.detail)
    return {"success": True}


# ================================================================ #
#  Pydantic 请求模型
# ================================================================ #

class AddDeviceRequest(BaseModel):
    """手动添加设备请求。"""
    ip: str
    port: int = 80
    username: str = "admin"
    password: str
    name: str = ""
    model: str = ""
    mac: str = ""
    gateway: str = ""
    subnet_mask: str = ""

class ConnectDeviceRequest(BaseModel):
    """连接设备请求。"""
    username: str = "admin"
    password: str = ""
    port: int = 80

class ModifyNetworkRequest(BaseModel):
    """修改网络配置请求。"""
    ip: str
    subnet_mask: str = "255.255.255.0"
    gateway: str

class SADPIpModifyRequest(BaseModel):
    """SADP IP 修改请求 (含循环验证)。"""
    mac: str
    username: str = "admin"  # 设备用户名
    password: str
    new_ip: str
    original_ip: str = ""
    subnet_mask: str = "255.255.255.0"
    gateway: str = ""

class DeviceCredentialsRequest(BaseModel):
    """v1.1.26: 设备凭据设置请求。"""
    username: str
    password: str

class PTZMoveRequest(BaseModel):
    """PTZ 移动请求。

    speed: 1-7 档位,默认 4 档
    档位映射: 1→15, 2→30, 3→45, 4→60, 5→75, 6→90, 7→100
    """
    direction: str
    speed: int = 4  # 默认 4 档

class PTZAbsoluteRequest(BaseModel):
    """绝对坐标移动请求。

    设计约束: 绝对移动不暴露 speed 给前端，内部固定 50。
    """
    pan: float
    tilt: float
    zoom: float | None = None


class PTZRecordRequest(BaseModel):
    """PTZ 录像请求。"""
    target_name: str = ""


class TelescopeSlewRequest(BaseModel):
    """望远镜 Slew 请求。"""
    ra: float
    dec: float


# ================================================================ #
#  操作日志 (最多 10 条)
# ================================================================ #

operations_log: list[dict[str, str]] = []

def _read_lines(filepath, n=None):
    """Read file lines in thread (for asyncio.to_thread)."""
    with open(filepath, encoding="utf-8") as f:
        return f.readlines() if n is None else f.readlines()[-n:]

def _read_file(filepath):
    """Read entire file in thread."""
    with open(filepath, encoding="utf-8") as f:
        return f.read()

def _write_file(filepath, content):
    """Write file in thread."""
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(content)


def _write_binary(filepath, data: bytes):
    """Write binary data in thread."""
    with open(filepath, "wb") as f:
        f.write(data)


def _normalize_mac(mac: str) -> str:
    """统一 MAC 为无分隔符小写格式（含合法性校验，委托 device_path.normalize_mac）。"""
    from src.advanced.device_path import normalize_mac
    return normalize_mac(mac, strict=True)


# ================================================================ #
#  管理器注入
# ================================================================ #

_managers: dict[str, Any] = {}


def set_managers(**kwargs: Any) ->PTZDeviceController | None:
    """注入所有管理器实例。"""
    _managers.update(kwargs)


def _resolve_device_id_to_ip(mgr: PTZDeviceController, device_id: str) -> str | None:
    """解析 device_id 为实际 IP 地址。

    device_id 可能是 IP 地址或 MAC 地址。
    - 如果是 IP: 直接返回
    - 如果是 MAC: 从 SADP 发现缓存或已存储设备中查找 IP
    """
    import re
    # 判断是否为 MAC 地址格式 (冒号、横杠分隔或无分隔符)
    mac_clean = normalize_mac(device_id)
    # 无分隔符12位十六进制 = MAC
    if len(mac_clean) == 12 and re.match(r'^[0-9a-f]{12}$', mac_clean):
        # v1.1.25: 连接时固化缓存优先（_active_mac/_active_ip 零 IO）——
        # 原路径每次 list_stored_devices() 文件 IO 在事件循环内实测高达 743ms，
        # 把 move/stop 竞态窗口拉到秒级
        active_ip = mgr.get_active_ip(device_id)
        if active_ip:
            return active_ip
        # 是 MAC 地址,尝试从发现缓存中查找
        discovered = mgr.get_discovered_devices()
        for dev in discovered:
            dev_mac = normalize_mac(dev.get("mac", ""))
            if dev_mac == mac_clean:
                return dev.get("ip")
        # 从已存储设备中查找
        stored = mgr.list_stored_devices()
        for dev in stored:
            dev_mac = normalize_mac(dev.get("mac", ""))
            if dev_mac == mac_clean:
                ip = dev.get("ip", "")
                if ip:
                    return ip
        return None
    # 看起来像 IP 地址,直接返回
    return device_id


def _resolve_device_id_to_mac(mgr: PTZDeviceController, device_id: str) -> str:
    """v10.05: 解析 device_id 为真实 MAC 地址。

    device_id 可能是 IP 或 MAC:
    - 如果是 MAC (12 位 hex): 直接返回 clean 格式
    - 如果是 IP: 从 registry / ISAPIPTZController 缓存中查找对应 MAC
    - 都找不到: 返回原始 device_id（保持向后兼容）
    """
    import re

    mac_clean = normalize_mac(device_id)

    # 已经是合法 MAC 格式（12 位 hex）
    if len(mac_clean) == 12 and re.match(r'^[0-9a-f]{12}$', mac_clean):
        return mac_clean

    # 看起来像 IP，尝试从已存储设备中查找对应 MAC
    if "." in device_id:
        # 从 registry 查找（通过 DeviceManager）
        dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
        if dm:
            for dev in dm.list_devices():
                if dev.ip == device_id:
                    return dev.mac

        # 从 ISAPIPTZController 缓存查找
        if mgr._active_ip == device_id and mgr._active_mac:
            return mgr._active_mac

    # 无法解析，返回原始值
    return device_id


# ================================================================ #
#  健康检查端点
# ================================================================ #


@health_router.get("/health", summary="全局健康检查")
async def global_health() -> dict:
    """全局健康检查端点 (GET /api/v1/health)。"""
    module_keys = (
        "ptz_controller",
        "device_manager",
    )

    module_health: dict[str, dict[str, Any]] = {}
    for key in module_keys:
        module_health[key] = {
            "status": "initialized" if _managers.get(key) is not None else "not_initialized",
            "present": key in _managers,
        }

    all_present = all(m["present"] for m in module_health.values())

    return {
        "status": "healthy" if all_present else "degraded",
        "modules": module_health,
        "timestamp": datetime.now(UTC).isoformat(),
    }


@health_router.get("/version", summary="获取版本号")
async def get_version() -> dict:
    """获取系统版本号 (GET /api/v1/version)。"""
    return {
        "version": VERSION,
        "version_num": VERSION_NUM,
    }


# === OTA 更新 API ===


@health_router.get("/update/check", summary="检查 OTA 更新")
async def update_check() -> dict:
    """检查是否有新版本可用 (GET /api/v1/update/check)。

    同时返回整包更新和文件级更新信息。
    """
    from src.updater.update_checker import (
        check_all_updates,
        get_cached_file_updates,
        get_cached_update_info,
    )

    cached_full = get_cached_update_info()
    cached_files = get_cached_file_updates()

    if cached_full is not None and cached_files is not None:
        result = _update_info_to_dict(cached_full)
        result["file_updates"] = [_file_update_to_dict(f) for f in cached_files]
        return result

    # 缓存未就绪，实时查询
    combined = await check_all_updates()
    result = _update_info_to_dict(combined.full_package)
    result["file_updates"] = [_file_update_to_dict(f) for f in combined.file_updates]
    return result


@health_router.get("/update/status", summary="获取更新执行状态")
async def update_status() -> dict:
    """获取当前更新执行状态（v1.1.34: 后台下载/安装状态机字段）。"""
    from src.updater.update_applier import get_update_status

    return get_update_status()


@health_router.post("/update/download", summary="后台下载安装包")
async def update_download() -> dict:
    """v1.1.34: 后台下载整包安装包到系统下载目录。

    立即返回；下载在服务端后台任务继续，弹窗关闭/请求断开不影响。
    幂等：downloading 中返回当前状态；同版本已下载返回 downloaded。
    """
    from src.updater.update_applier import start_background_download
    from src.updater.update_checker import check_for_update, get_cached_update_info

    cached = get_cached_update_info()
    if cached is None:
        cached = await check_for_update()

    if not cached.update_available:
        return {"success": False, "message": "没有可用更新"}
    if not cached.download_url:
        return {"success": False, "message": "下载地址为空"}

    status = start_background_download(
        download_url=cached.download_url,
        expected_sha256=cached.sha256,
        version=cached.latest_version,
    )
    return {"success": True, **status}


@health_router.post("/update/install", summary="启动安装向导")
async def update_install() -> dict:
    """v1.1.34: 启动可见 NSIS 安装向导（需已下载完成）。

    安装器 .onInit 会结束本程序与占用进程，前端展示引导弹窗。
    """
    from src.updater.update_applier import install_update

    return install_update()


@health_router.post("/update/apply", summary="执行文件级 OTA 更新")
async def update_apply(request: Request) -> dict:
    """应用文件级更新（POST /api/v1/update/apply）。

    v1.1.34: 整包更新拆分为 /update/download + /update/install，
    本端点仅接受 mode=files（下载文件 → 校验 → 替换 → 更新 file_versions.json）。
    """
    from src.updater.update_applier import apply_file_updates
    from src.updater.update_checker import check_file_updates, get_cached_file_updates

    try:
        body = await request.json()
    except Exception:
        body = {}

    mode = body.get("mode", "files")
    if mode != "files":
        return {
            "success": False,
            "message": "整包更新请使用 /update/download + /update/install",
        }

    cached_files = get_cached_file_updates()
    if cached_files is None:
        cached_files = await check_file_updates()

    if not cached_files:
        return {"success": False, "message": "没有文件级更新"}

    file_entries = [
        {
            "path": f.path,
            "target": f.target,
            "version": f.version,
            "sha256": f.sha256,
            "requires_restart": f.requires_restart,
            "restart_target": f.restart_target,
        }
        for f in cached_files
    ]

    result = await apply_file_updates(file_entries)
    return {
        "success": result.success,
        "status": result.status,
        "message": result.message,
    }


def _update_info_to_dict(info: object) -> dict:
    """将 UpdateInfo 转为 API 响应字典。"""
    return {
        "update_available": info.update_available,
        "latest_version": info.latest_version,
        "latest_version_num": info.latest_version_num,
        "current_version": info.current_version,
        "download_url": info.download_url,
        "file_size": info.file_size,
        "sha256": info.sha256,
        "changelog": info.changelog,
        "release_date": info.release_date,
        "error": info.error,
    }


def _file_update_to_dict(info: object) -> dict:
    """将 FileUpdateInfo 转为 API 响应字典。"""
    return {
        "path": info.path,
        "target": info.target,
        "version": info.version,
        "sha256": info.sha256,
        "size": info.size,
        "changelog": info.changelog,
        "requires_restart": info.requires_restart,
        "restart_target": info.restart_target,
        "local_sha256": getattr(info, "local_sha256", ""),
        "integrity_mismatch": getattr(info, "integrity_mismatch", False),
    }


@api_router.get("/diagnostics/eventloop", summary="事件循环冻结诊断", tags=["Diagnostics"])
async def eventloop_diagnostics(duration: float = 5.0) -> dict:
    """在事件循环内部测量冻结，并记录冻结期间的任务。

    Args:
        duration: 诊断持续时间（秒），默认 5 秒
    """
    import asyncio
    import time

    results = []
    total_checks = 0
    start_time = time.time()

    while time.time() - start_time < duration:
        total_checks += 1
        t0 = time.perf_counter()
        await asyncio.sleep(0.01)  # 应该几乎立即返回
        actual = (time.perf_counter() - t0) * 1000

        if actual > 100:  # 如果超过 100ms，记录为冻结
            # 获取当前所有任务
            tasks = []
            for task in asyncio.all_tasks():
                if not task.done():
                    # 获取任务的协程名称
                    coro = task.get_coro()
                    coro_name = coro.__name__ if hasattr(coro, '__name__') else str(coro)
                    tasks.append({
                        "name": coro_name,
                        "task_id": id(task),
                    })

            results.append({
                "timestamp": time.time(),
                "freeze_ms": actual,
                "active_tasks": tasks[:10],  # 最多记录 10 个任务
                "total_tasks": len(tasks),
            })

    # 统计
    freezes = [r for r in results if r["freeze_ms"] > 100]
    severe_freezes = [r for r in results if r["freeze_ms"] > 1000]

    # 分析最常见的任务
    task_counts = {}
    for r in freezes:
        for t in r["active_tasks"]:
            name = t["name"]
            task_counts[name] = task_counts.get(name, 0) + 1

    top_tasks = sorted(task_counts.items(), key=lambda x: x[1], reverse=True)[:10]

    return {
        "success": True,
        "duration": duration,
        "total_checks": total_checks,
        "freezes_over_100ms": len(freezes),
        "freezes_over_1000ms": len(severe_freezes),
        "max_freeze_ms": max((r["freeze_ms"] for r in results), default=0),
        "avg_freeze_ms": sum(r["freeze_ms"] for r in freezes) / len(freezes) if freezes else 0,
        "top_tasks_during_freeze": [{"task": t[0], "count": t[1]} for t in top_tasks],
        "freeze_details": freezes[:20],  # 最多返回 20 个冻结记录
    }


@api_router.get("/localhost", summary="本机信息", tags=["System"])
async def get_localhost_info() -> dict:
    """获取本机系统信息 (hostname, CPU, RAM, GPU, IP, gateway).

    首次调用时会自动收集并保存到 data/reports/localhost.json
    """
    from src.advanced.startup import check_localhost_exists, get_localhost_info, run_startup

    # v1.1.49: 硬件采集/文件读取移入线程池，避免阻塞事件循环。原同步调用会冻结 loop ~1.6s，
    # 令启动期并发的一批 API 全部排队（/geo /ptz/connected 等轻量端点也被拖到 ~1.5s）。
    need_collect = not await asyncio.to_thread(check_localhost_exists)
    if not need_collect:
        saved = await asyncio.to_thread(get_localhost_info) or {}
        need_collect = saved.get("version") != "v1.0.8"
    if need_collect:
        try:
            info = await asyncio.to_thread(run_startup)
            return {"success": True, "data": info, "first_run": True}
        except Exception as e:
            return {"success": False, "message": f"收集本机信息失败: {e}"}

    # 读取已保存的信息
    info = await asyncio.to_thread(get_localhost_info)
    if info:
        return {"success": True, "data": info}
    else:
        return {"success": False, "message": "无法读取本机信息"}


# ================================================================ #
#  SADP 发现端点
# ================================================================ #


@api_router.get("/discovery/sadp", summary="SADP 设备发现", tags=["Discovery"])
async def sadp_discover(bind_ip: str = "0.0.0.0") -> dict:
    """通过 SADP 多播发现局域网内 PTZ 设备。

    BUG-003 修复: 直接使用 SADPManager,不依赖 PTZDeviceController。
    """
    import asyncio
    start = time.time()

    # 优先使用 PTZDeviceController（如果已初始化）
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if mgr:
        devices = await asyncio.to_thread(mgr.discover_devices, bind_ip=bind_ip)
    else:
        # 直接使用 SADPManager 不依赖 PTZDeviceController
        try:
            from src.core.sadp_discovery import SADPManager
            sadp_mgr = SADPManager()
            raw_devices = await asyncio.to_thread(sadp_mgr.discover_devices, timeout=3, bind_ip=bind_ip)
            # 转换为 SADP 标准格式
            devices = []
            for d in raw_devices:
                devices.append({
                    "mac": d.get("mac", ""),
                    "ip": d.get("ip", ""),
                    "subnet_mask": d.get("subnet_mask", ""),
                    "gateway": d.get("gateway", ""),
                    "model": d.get("model", ""),
                    "serial_number": d.get("serial_number", ""),
                    "device_name": d.get("device_name", ""),
                    "firmware_version": d.get("firmware_version", ""),
                    "activated": d.get("activated", False),
                    "is_hikvision": d.get("is_hikvision", False),
                })
        except Exception as e:
            return {"success": False, "message": f"SADP 发现异常: {e}", "devices": []}

    elapsed = round(time.time() - start, 1)

    # BUG-015 修复: 统一 MAC 地址格式为横杠小写 (24-0f-9b-76-41-93)
    for dev in devices:
        mac = dev.get("mac", "")
        if mac:
            clean = normalize_mac(mac)
            if len(clean) == 12:
                dev["mac"] = "-".join(clean[i:i+2] for i in range(0, 12, 2))

    return {
        "success": True,
        "message": f"发现 {len(devices)} 台设备 (耗时 {elapsed}s)",
        "devices": devices,
        "count": len(devices),
    }


# ========== SADP 发现 API ==========


# ================================================================ #
#  设备管理端点
# ================================================================ #


# v1.1.26: 检查 info.json 中是否存有凭据（解决卸载重装后认证失败）
def _has_creds_in_info(mac: str) -> bool:
    """v1.1.26: 读取 data/devices/{mac}/info.json 检查 username/password 是否存在。"""
    import json as _json_local

    from src.advanced.device_path import get_device_dir
    try:
        info_path = get_device_dir(mac) / "info.json"
        if info_path.exists():
            _info = _json_local.loads(info_path.read_text(encoding="utf-8"))
            return bool(_info.get("username") and _info.get("password"))
    except Exception:
        pass
    return False


def _dedupe_same_ip_list(devices_list: list[dict]) -> list[dict]:
    """v1.1.35: 同 IP 重复注册项收敛——多 MAC 同 IP 时保留有凭据者
    （其次已连接者），合并丢弃其余并记日志（生产同 IP 双 MAC 条目是
    快速连接选错凭据候选/列表重复行的根因）。"""
    slots: list[dict] = []
    pos: dict[str, int] = {}
    dropped: list[str] = []
    for dev in devices_list:
        ip = dev.get("ip", "")
        if not ip:
            slots.append(dev)
            continue
        if ip not in pos:
            pos[ip] = len(slots)
            slots.append(dev)
            continue
        cur = slots[pos[ip]]
        better = (dev.get("has_credentials") and not cur.get("has_credentials")) or (
            dev.get("status") == "已连接" and cur.get("status") != "已连接"
        )
        if better:
            dropped.append(cur.get("mac", ""))
            slots[pos[ip]] = dev
        else:
            dropped.append(dev.get("mac", ""))
    if dropped:
        log_api("devices_dedupe", {"same_ip_dropped": dropped})
    return slots


@api_router.get("/devices", summary="设备列表", tags=["Devices"])
async def list_devices() -> dict:
    """获取设备列表(MAC唯一标识,SADP更新IP/型号).

    v1.1.25: 状态三态化——已连接(ISAPI 会话)/在线(可达性缓存)/离线；
    可达性来自 DeviceReader 后台低频 sweep 缓存，请求路径不发探测。
    """
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "data": [], "total": 0}

    reader = get_device_reader()
    reach = reader.get_reachability() if reader else {}

    def _status_for(ip: str, is_connected: bool) -> tuple[str, bool]:
        reach_ok = bool(reach.get(ip, (False, 0.0))[0]) if ip else False
        if is_connected:
            return "已连接", True
        return ("在线", True) if reach_ok else ("离线", False)

    stored = mgr.list_stored_devices()
    discovered = mgr.get_discovered_devices()

    # 以MAC为唯一键,存储设备为基础
    devices: dict[str, dict[str, Any]] = {}
    for dev in stored:
        mac = _normalize_mac(dev.get("mac", "") or "")
        if mac:
            ip = dev.get("ip", "")
            # v10.42: 检查设备是否已连接（ISAPI 客户端存在）
            is_connected = ip in mgr._clients if ip else False
            status, reach_ok = _status_for(ip, is_connected)
            devices[mac] = {
                "mac": mac,
                "ip": ip,
                "name": dev.get("name", "") or dev.get("device_name", ""),
                "model": dev.get("model", ""),
                "gateway": dev.get("gateway", ""),
                "subnet_mask": dev.get("subnet_mask", ""),
                "has_credentials": _has_creds_in_info(mac),
                "device_type": "已知设备",
                "status": status,
                "reachable": reach_ok,
            }

    # SADP发现:更新或新增
    for sadp in discovered:
        mac = _normalize_mac(sadp.get("mac", "") or "")
        if not mac:
            continue
        if mac in devices:
            # 更新:SADP数据为准(IP可能变了)，同时更新网关
            devices[mac]["ip"] = sadp.get("ip", "") or devices[mac]["ip"]
            devices[mac]["model"] = sadp.get("model", "") or devices[mac]["model"]
            if sadp.get("gateway"):
                devices[mac]["gateway"] = sadp["gateway"]
            if sadp.get("subnet_mask"):
                devices[mac]["subnet_mask"] = sadp["subnet_mask"]
            # 更新连接状态
            ip = devices[mac]["ip"]
            status, reach_ok = _status_for(ip, ip in mgr._clients)
            devices[mac]["status"] = status
            devices[mac]["reachable"] = reach_ok
        else:
            # 新增:首次发现，标记为新设备
            ip = sadp.get("ip", "")
            status, reach_ok = _status_for(ip, ip in mgr._clients)
            devices[mac] = {
                "mac": mac,
                "ip": ip,
                "name": sadp.get("device_name", "") or sadp.get("name", ""),
                "model": sadp.get("model", "") or "PTZ",
                "gateway": sadp.get("gateway", ""),
                "subnet_mask": sadp.get("subnet_mask", ""),
                "has_credentials": False,
                "device_type": "新设备",
                "status": status,
                "reachable": reach_ok,
            }

    data = _dedupe_same_ip_list(list(devices.values()))
    return {"success": True, "data": data, "total": len(data)}


@api_router.post("/devices/refresh", summary="刷新设备状态", tags=["Devices"])
async def refresh_devices() -> dict:
    """并行ping所有设备,更新在线状态(timeout=1s)。"""
    from src.core.ptz_controller import check_reachable

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "data": [], "total": 0}

    stored = mgr.list_stored_devices()
    discovered = mgr.get_discovered_devices()

    # 以MAC为唯一键
    devices: dict[str, dict[str, Any]] = {}
    for dev in stored:
        mac = _normalize_mac(dev.get("mac", "") or "")
        if mac:
            devices[mac] = {
                "mac": mac,
                "ip": dev.get("ip", ""),
                "name": dev.get("name", "") or dev.get("device_name", ""),
                "model": dev.get("model", ""),
                "gateway": dev.get("gateway", ""),
                "subnet_mask": dev.get("subnet_mask", ""),
                "has_credentials": _has_creds_in_info(mac),
                "device_type": "已知设备",
                "status": "离线",
            }

    for sadp in discovered:
        mac = _normalize_mac(sadp.get("mac", "") or "")
        if not mac:
            continue
        if mac in devices:
            # 已知设备：SADP 更新 IP/网关
            stored_gateway = devices[mac]["gateway"]
            stored_subnet = devices[mac]["subnet_mask"]
            devices[mac]["ip"] = sadp.get("ip", "") or devices[mac]["ip"]
            devices[mac]["model"] = sadp.get("model", "") or devices[mac]["model"]
            if sadp.get("gateway"):
                devices[mac]["gateway"] = sadp["gateway"]
            if sadp.get("subnet_mask"):
                devices[mac]["subnet_mask"] = sadp["subnet_mask"]
            # v1.1.21: SADP 提供的网关/子网掩码落盘（原仅进响应不持久化，
            # 导致扫描时可见、刷新后丢失）；空值不覆盖语义由 upsert 保证
            if (not stored_gateway and sadp.get("gateway")) or (
                not stored_subnet and sadp.get("subnet_mask")
            ):
                mgr.config.upsert_device(mac, {
                    "gateway": sadp.get("gateway", ""),
                    "subnet_mask": sadp.get("subnet_mask", ""),
                })
        else:
            # 新设备：仅 SADP 发现，未保存凭据
            devices[mac] = {
                "mac": mac,
                "ip": sadp.get("ip", ""),
                "name": sadp.get("device_name", "") or sadp.get("name", ""),
                "model": sadp.get("model", "") or "PTZ",
                "gateway": sadp.get("gateway", ""),
                "subnet_mask": sadp.get("subnet_mask", ""),
                "has_credentials": False,
                "device_type": "新设备",
                "status": "离线",
            }

    # 并行ping检测状态（v1.1.25: 400ms 超时暖 ARP + check_reachable 已含 TCP 回退）
    devices_list = list(devices.values())
    ips = [d["ip"] for d in devices_list if d["ip"]]

    loop = asyncio.get_running_loop()
    reachability = await asyncio.gather(
        *[loop.run_in_executor(None, check_reachable, ip, 0.4) for ip in ips]
    )

    # 更新状态：在线/离线
    ip_status = dict(zip(ips, reachability, strict=False))
    # v1.1.25: 探测结果回写 DeviceReader 可达性缓存（/devices 三态数据源）
    _reader = get_device_reader()
    if _reader:
        _reader.update_reach_cache(ip_status)
    for dev in devices_list:
        ip = dev.get("ip", "")
        if ip and ip_status.get(ip, False):
            dev["status"] = "在线"
        else:
            dev["status"] = "离线"

    # 检查 ISAPI 已连接状态（优先级最高）
    connected_ips = mgr.list_controllers()
    for dev in devices_list:
        if dev.get("ip") in connected_ips:
            dev["status"] = "已连接"

    # v1.1.35: 同 IP 重复项收敛（与 /devices 同语义）
    devices_list = _dedupe_same_ip_list(devices_list)

    return {"success": True, "data": devices_list, "total": len(devices_list)}


@api_router.get("/devices/active", summary="获取上次连接的设备", tags=["Devices"])
async def get_active_device() -> dict:
    """v6.30: 获取上次连接的设备(用于快速连接)。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "active": False, "device": None, "message": "PTZDeviceController未初始化"}

    device = mgr.get_connected_device()
    if device is None:
        return {"success": False, "active": False, "device": None, "message": "无上次连接设备"}

    return {"success": True, "active": True, "device": device}


@api_router.post("/devices/active", summary="设置上次连接的设备", tags=["Devices"])
async def set_active_device(mac: str) -> dict:
    """v6.30: 设置上次连接的设备(用于快速连接)。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # 从 devices/{mac}/info.json 读取设备信息
    device = mgr.config.get_device(mac)
    if device is None:
        return {"success": False, "message": f"设备不存在: {mac}"}

    # 更新 registry.json 的 active_device 和 last_connected
    mgr.config.upsert_device(mac, {"mac": mac})

    return {
        "success": True,
        "message": f"上次连接设备已设置: {device.get('model', 'Unknown')} @ {device.get('ip')}",
        "device": device
    }


@api_router.post("/devices", summary="注册设备", tags=["Devices"])
async def register_device(req: AddDeviceRequest) -> dict:
    """手动添加设备(IP + 凭据)并自动设为活跃。

    v7.17: 必须有 MAC 地址才能注册,如果没有则从 SADP 发现中查找。
    """
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # v7.17: 如果没有 MAC,从 SADP 发现中查找
    device_mac = req.mac
    device_model = req.model
    device_name = req.name

    if not device_mac:
        # 检查 SADP 发现中是否有该 IP 的设备
        sadp_devices = mgr._discovered
        for mac, info in sadp_devices.items():
            if info.ip == req.ip:
                device_mac = mac
                device_model = device_model or getattr(info, 'model', None)
                device_name = device_name or getattr(info, 'device_name', None)
                break

        if not device_mac:
            return {"success": False, "message": f"未发现设备 {req.ip},请先执行 SADP 发现"}

    # v1.1.43: 密码落盘前 ISAPI 单次校验（防错误密码注入 → 防设备锁定）
    verify = await asyncio.to_thread(
        mgr.verify_password_once, req.ip, req.username or "admin",
        req.password, req.port,
    )
    if not verify.get("success"):
        err_type = verify.get("error_type", "unknown")
        msg = verify.get("message", "密码错误")
        if err_type == "auth":
            msg = "密码错误，请重新输入"
        log_error("register_device", {"ip": req.ip, "error_type": err_type, "message": msg})
        return {
            "success": False,
            "error_code": "AUTH_FAILED" if err_type == "auth" else "VERIFY_FAILED",
            "error_type": err_type,
            "message": msg,
        }

    # 使用正确的 MAC 保存
    mgr.save_credentials(
        ip=req.ip,
        username=req.username,
        password=req.password,
        port=req.port,
        mac=device_mac,
        model=device_model,
        name=device_name,
        gateway=req.gateway,
        subnet_mask=req.subnet_mask,
    )

    # 注册到 DeviceManager
    dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
    if dm:
        dm.register_device(
            mac=device_mac,  # v7.17: 使用真实 MAC
            ip=req.ip,
            name=device_name or f"PTZ-{req.ip}",
            model=device_model or "PTZ",
        )
        # v10.40: 通过 DeviceManager 设为活跃设备
        dm.set_active_device(device_mac)

    # 更新设备配置文件
    mgr.config.upsert_device(device_mac, {"ip": req.ip, "model": device_model or "", "port": req.port, "mac": device_mac})

    log_info("device", "register", {"ip": req.ip, "mac": device_mac, "name": device_name})
    return {
        "success": True,
        "message": f"设备已保存: {device_mac} @ {req.ip}",
        "device": {
            "ip": req.ip,
            "port": req.port,
            "name": device_name,
            "model": device_model,
            "mac": device_mac,
        },
    }


@api_router.post("/devices/{mac}/connect", summary="连接设备", tags=["Devices"])
async def connect_device(mac: str, request: Request, req: ConnectDeviceRequest | None = None) -> dict:
    """连接 PTZ 设备并认证。

    device_id 可以是 IP 地址或 MAC 地址。如果是 MAC,自动解析为 IP。
    使用 admin 作为默认用户名,密码必须由用户提供或已保存在配置中。

    BUG-014 修复: ISAPI 连接前 socket 预检,5 秒超时快速失败。
    """
    log_api("connect", {"mac": mac})

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        log_error("connect", {"mac": mac, "error": "PTZDeviceController未初始化"})
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # 解析 device_id: 如果看起来像 MAC,从发现缓存中解析 IP
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        log_error("connect", {"mac": mac, "error": "无法解析设备标识"})
        return {"success": False, "message": f"无法解析设备标识: {mac},请提供 IP 地址或确保设备已通过 SADP 发现"}

    # v10.42: 如果未提供凭据或密码为空，自动从存储获取
    if req is None or not req.password:
        creds = mgr.get_credentials(target_ip)
        if not creds:
            log_error("connect", {"mac": mac, "error": "设备未保存凭据"})
            return {"success": False, "message": "设备未保存凭据,请先提供用户名和密码"}
        username = creds["username"]
        password = creds["password"]
        port = creds.get("port", 80)
    else:
        username = req.username or "admin"
        password = req.password
        port = req.port

    # v1.1.35: 凭据取证日志——每次连接记录凭据来源，生产认证失败复现时一次 grep 定位
    log_api("connect", {
        "mac": mac, "ip": target_ip,
        "cred_source": "provided" if (req is not None and req.password) else "saved",
    })

    result = await asyncio.to_thread(mgr.connect_device, target_ip, username, password, port)
    if result and result.get("success"):
        log_info("connect", {"mac": mac, "ip": target_ip, "status": "success"})
        # v1.1.35: 请求体带密码且连接成功→必落盘（修"输入凭据"弹窗只连不存缺口）
        if req is not None and req.password:
            try:
                _persist_mac = _resolve_device_id_to_mac(mgr, mac)
                if _persist_mac:
                    mgr.save_credentials(
                        ip=target_ip, username=username, password=req.password,
                        port=port, mac=_persist_mac,
                    )
            except Exception as e:
                log_api("credential_persist", {"mac": mac, "error": str(e)})
        # v1.1.25: 手动连接成功后清除启动重连失败消息（防 toast 每次刷新复现）
        request.app.state.reconnect_message = ""
        # v1.1.34: 存储网关缺失时后台补偿扫描一次（SADP 落盘，不阻塞响应）
        try:
            _bf_mac = _resolve_device_id_to_mac(mgr, mac)
            _bf_info = mgr.config.get_device_by_mac(_bf_mac) if _bf_mac else None
            if _bf_mac and not (_bf_info or {}).get("gateway"):
                asyncio.create_task(asyncio.to_thread(mgr.backfill_gateway_via_scan))
        except Exception as e:
            log_api("gateway_backfill", {"error": str(e)})
        # v10.05: 解析为真实 MAC 后更新 DeviceManager 状态
        dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
        if dm:
            resolved_mac = _resolve_device_id_to_mac(mgr, mac)
            dm.update_status(resolved_mac, "online")
        # v10.09: 连接成功后立即触发一次轮询，填充 DeviceReader 缓存
        reader = get_device_reader()
        if reader:
            reader.poll_once()
            # v11.30: 启动 PTZ 位置事件订阅（alertStream 推送链路）
            reader.refresh_event_subscription()
        # v10.67: 连接成功后更新 OBS GStreamer 源的 RTSP 地址
        # v1.0.8: 从本地 PTZ 设备信息读取视频编码，供媒体源自适应 H264/H265
        # v1.1.4: 移除 `if obs_mgr.connected` 守卫——未连接时 on_device_connected
        # 内部已有 supervise_init + 30s 等待自愈，守卫会造成 RTSP 回写遗漏窗
        from src.obs.obs_manager import get_obs_manager
        try:
            obs_mgr = get_obs_manager()
            _creds = mgr.get_credentials(target_ip) or {}
            _codec = _creds.get("video_codec", "H.265")
            asyncio.create_task(
                obs_mgr.on_device_connected(target_ip, username, password, _codec)
            )
        except Exception as e:
            log_api("obs_update", {"error": str(e)})
        return result

    log_error("connect", {
        "mac": mac, "ip": target_ip, "status": "failed",
        "message": (result or {}).get("message", ""),
    })
    return result or {"success": False, "message": "连接失败"}


@api_router.post("/devices/{mac}/disconnect", summary="断开设备", tags=["Devices"])
async def disconnect_device(mac: str) -> dict:
    """断开 PTZ 设备连接."""
    log_api("disconnect", {"mac": mac})

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    await asyncio.to_thread(mgr.disconnect_device, mac)

    # v11.30: 断连后停止事件订阅
    _reader = get_device_reader()
    if _reader:
        _reader.refresh_event_subscription()

    dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
    if dm:
        resolved_mac = _resolve_device_id_to_mac(mgr, mac)
        dm.update_status(resolved_mac, "offline")

    log_info("disconnect", {"mac": mac, "status": "success"})
    return {"success": True, "message": f"设备已断开: {mac}"}


@api_router.delete("/devices/{mac}", summary="删除设备", tags=["Devices"])
async def delete_device(mac: str) -> dict:
    """删除设备。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # 先断开
    await asyncio.to_thread(mgr.disconnect_device, mac)

    # 移除凭据
    await asyncio.to_thread(mgr.remove_credentials, mac)

    dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
    if dm:
        dm.unregister_device(mac)

    log_info("device", "delete", {"device_id": mac})
    return {
        "success": True,
        "message": f"设备已删除: {mac}",
    }


@api_router.post("/devices/{mac}/credentials", summary="设置设备凭据", tags=["设备"])
async def set_device_credentials(mac: str, req: DeviceCredentialsRequest) -> dict:
    """v1.1.28: 统一凭据保存路径，写入 ptz_config.json + 内存缓存。

    修复: 原实现仅写 info.json，get_credentials() 不读 info.json，
    导致重启后凭据丢失、连接失败。
    """
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # 1. 从已有设备信息获取 IP（info.json 或 config）
    device_info = mgr.get_device_by_mac(mac)
    ip = ""
    port = 80
    model = ""
    name = ""
    if device_info:
        ip = device_info.get("ip", "")
        port = device_info.get("port", 80)
        model = device_info.get("model", "")
        name = device_info.get("name", "") or device_info.get("device_name", "")

    if not ip:
        return {"success": False, "message": f"设备 {mac} 无 IP 地址，请先添加设备"}

    # v1.1.43: 密码落盘前 ISAPI 单次校验（防错误密码注入 → 防设备锁定）
    verify = await asyncio.to_thread(
        mgr.verify_password_once, ip, req.username or "admin",
        req.password, port,
    )
    if not verify.get("success"):
        err_type = verify.get("error_type", "unknown")
        msg = verify.get("message", "密码错误")
        if err_type == "auth":
            msg = "密码错误，请重新输入"
        log_error("set_credentials", {"mac": mac, "ip": ip, "error_type": err_type, "message": msg})
        return {
            "success": False,
            "error_code": "AUTH_FAILED" if err_type == "auth" else "VERIFY_FAILED",
            "error_type": err_type,
            "message": msg,
        }

    # 2. 统一通过 save_credentials 写入 ptz_config.json + 内存缓存
    mgr.save_credentials(
        ip=ip,
        username=req.username,
        password=req.password,
        port=port,
        mac=mac,
        model=model,
        name=name,
    )

    log_info("device", "credentials_set", {"mac": mac, "ip": ip, "username": req.username})
    return {"success": True, "message": "凭据已保存"}


@api_router.get("/devices/{mac}/info", summary="设备详细信息", tags=["Devices"])
async def get_device_info(mac: str) -> dict:
    """获取设备详细信息(通过 ISAPI)。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # 解析 device_id (MAC) 到 IP
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": "无法解析设备标识: 请提供 IP 地址或确保设备已连接"}

    info = await asyncio.to_thread(mgr.get_device_info, target_ip)
    if "error" in info:
        return {"success": False, "message": info["error"]}

    return {"success": True, "data": info}


@api_router.put("/devices/{mac}/network", summary="修改设备网络配置", tags=["Devices"])
async def modify_network(mac: str, req: ModifyNetworkRequest) -> dict:
    """修改设备网络配置(通过 SADP 协议)。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # v10.74: 移到线程池避免阻塞事件循环（SADP 协议调用可能阻塞 5-30 秒）
    result = await asyncio.to_thread(
        mgr.modify_device_network,
        ip=mac,
        new_ip=req.ip,
        subnet_mask=req.subnet_mask,
        gateway=req.gateway,
    )
    return result


# ================================================================ #
#  SADP IP 修改 & 系统信息端点
# ================================================================ #


@api_router.post("/sadp/{mac}/modify-ip", summary="SADP 修改设备 IP (含自动验证循环)", tags=["SADP"])
async def sadp_modify_ip(mac: str, req: SADPIpModifyRequest) -> dict:
    """通过 SADP DLL 修改设备 IP,含自动重试和扫描确认循环。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    try:
        result = mgr.modify_reachable_with_retry(
            mac=mac,
            username=req.username or "admin",
            password=req.password,
            original_ip=req.original_ip or "",
            target_ip=req.new_ip,
            subnet_mask=req.subnet_mask,
            gateway=req.gateway,
        )
        # 修改成功后保存凭据（修复 BUG 5: 凭据未保存导致修改后无法连接）
        if result.get("success"):
            new_ip = req.new_ip
            username = req.username or "admin"
            mgr.save_credentials(
                ip=new_ip,
                username=username,
                password=req.password,
                port=80,
                mac=req.mac,
            )
            # 更新内存中的 _discovered 缓存 IP
            norm_mac = normalize_mac(req.mac)
            for dev in mgr._discovered.values():
                dev_mac_clean = normalize_mac(getattr(dev, "mac", ""))
                if dev_mac_clean == norm_mac:
                    dev.ip = new_ip
            log_api("sadp_modify_ip", {"mac": mac, "new_ip": new_ip, "username": username, "status": "success"})
        return result
    except Exception as e:
        import traceback
        return {"success": False, "message": f"修改IP异常: {str(e)}", "detail": traceback.format_exc()}


@api_router.get("/sadp/error-codes", summary="SADP 错误码参考表", tags=["SADP"])
async def sadp_error_codes() -> JSONResponse:
    """返回 SADP 错误码中文对照表。"""
    try:
        from src.core.sadp_discovery import SADP_ERROR_CODES
        return JSONResponse(
            content={"success": True, "codes": SADP_ERROR_CODES},
            ensure_ascii=False,
        )
    except ImportError:
        return JSONResponse(
            content={"success": False, "message": "SADP 模块未加载"},
            ensure_ascii=False,
        )


@api_router.post("/sadp/auto-reconnect", summary="已知设备自动重连", tags=["SADP"])
async def sadp_auto_reconnect(req: AutoReconnectDevicesRequest | None = None) -> dict:
    """扫描 SADP 设备 -> MAC匹配已保存设备 -> IP不可达时自动修改 IP。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    bind_ip = "0.0.0.0"
    target_ip = None
    if req:
        bind_ip = req.bind_ip or "0.0.0.0"
        target_ip = req.target_ip or None

    # v10.74: 移到线程池避免阻塞事件循环（SADP 扫描+重连可能阻塞 10-60 秒）
    result = await asyncio.to_thread(
        mgr.auto_reconnect_known_devices,
        bind_ip=bind_ip,
        target_ip=target_ip,
    )
    return result


class AutoReconnectDevicesRequest(BaseModel):
    """已知设备自动重连请求。"""
    bind_ip: str = "0.0.0.0"
    target_ip: str = ""


# ================================================================
#  v6.0 连接状态 API
# ================================================================

@api_router.get("/ptz/connected", summary="获取已连接设备状态 (v10.40)", tags=["PTZ"])
async def get_connected_device(request: Request) -> dict:
    """v10.40: 获取当前连接状态和上次连接设备信息。

    返回:
    - connected: 当前是否已连接
    - device: 上次连接的设备信息
    - mac_mismatch: 启动时 MAC 验证失败（设备已更换）
    """
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化", "connected": False, "device": None, "mac_mismatch": False}

    # v10.40: 通过 DeviceManager 检查 active_device
    connected = False
    if dm:
        active_mac = dm.get_active_device_mac()
        if active_mac and mgr._controllers:
            connected = True

    # 获取上次连接的设备信息
    device = mgr.get_connected_device()

    # v10.40: 读取 MAC 不匹配状态
    mac_mismatch = getattr(request.app.state, "mac_mismatch", False)
    # v1.1.21: 启动自动重连失败原因（前端 toast 展示真因）
    reconnect_message = getattr(request.app.state, "reconnect_message", "") or ""
    # v1.1.43: 自动重连密码错误终局标志（前端 toast 区分“密码错误”与“不可达”）
    reconnect_auth_failed = getattr(request.app.state, "reconnect_auth_failed", False)

    if not device:
        return {"success": True, "connected": connected, "device": None, "mac_mismatch": mac_mismatch,
                "reconnect_message": reconnect_message, "reconnect_auth_failed": reconnect_auth_failed}

    return {"success": True, "connected": connected, "device": device, "mac_mismatch": mac_mismatch,
            "reconnect_message": reconnect_message, "reconnect_auth_failed": reconnect_auth_failed}

@api_router.get("/system/info", summary="系统硬件信息", tags=["System"])
async def get_system_info() -> dict:
    """获取本机系统硬件信息 (hostname/CPU/RAM/GPU/VRAM)。"""
    try:
        from src.ptz.core.system_info import collect_system_info
        # v10.54-fix: 同步系统信息收集移到线程池，避免阻塞事件循环
        info = await asyncio.to_thread(collect_system_info)
        return {"success": True, "data": info}
    except Exception as e:
        return {"success": False, "message": str(e)}


# v1.1.0: GPU 探测结果内存缓存 (timestamp, count, names)，60s TTL
_gpu_probe_cache: tuple[float, int, list[str]] | None = None


def _write_gpu_info_to_local(gpu_count: int, gpu_names: list[str]) -> None:
    """v1.1.0: GPU 探测成功回写 local.json（供其他消费方；失败不影响主流程）。"""
    import json as _json

    from src.config_paths import DATA_DIR

    local_path = DATA_DIR / "config" / "local.json"
    try:
        cfg: dict = {}
        if local_path.exists():
            cfg = _json.loads(local_path.read_text(encoding="utf-8"))
        cfg["gpu_count"] = gpu_count
        cfg["gpu_names"] = gpu_names
        cfg["gpu_detected_at"] = datetime.now().isoformat(timespec="seconds")
        tmp = local_path.with_suffix(".tmp")
        tmp.write_text(_json.dumps(cfg, indent=2, ensure_ascii=False), encoding="utf-8")
        tmp.replace(local_path)
    except Exception as e:
        log_warn("system", "gpu_cache_write_failed", str(e))


@api_router.get("/system/capabilities", summary="功能可用性检测 (v10.02)", tags=["System"])
async def get_system_capabilities() -> dict:
    """检测高级功能可用性（禁止静默降级，前端根据结果灰显并提示）。

    - gpu_available: v1.1.0 实时探测优先（60s TTL 内存缓存），
      local.json 仅作探测失败回退（修复 gpu_count 恒 0 误报）
    - obs_plugin_installed: OBS 插件目录下存在 astrohub-obs.dll
    - capabilities: gpu_stack/gpu_stretch 需 GPU+插件；histogram 为 CPU 计算始终可用
    """
    import json
    from pathlib import Path

    from src.config_paths import DATA_DIR
    from src.obs.obs_config import load_obs_config

    warnings: list[str] = []

    # v1.1.0: GPU 实时探测优先（修复 local.json gpu_count 恒 0 导致有 GPU 却报无 GPU）
    global _gpu_probe_cache
    gpu_count = 0
    gpu_names: list[str] = []
    now = time.time()
    if _gpu_probe_cache is not None and now - _gpu_probe_cache[0] < 60:
        gpu_count, gpu_names = _gpu_probe_cache[1], _gpu_probe_cache[2]
    else:
        try:
            from src.ptz.core.system_info import get_gpu_info
            info = await asyncio.to_thread(get_gpu_info)
            gpu_count = int(info.get("gpu_count", 0))
            gpu_names = list(info.get("gpu_names", []) or [])
        except Exception:
            gpu_count = 0
        if gpu_count > 0:
            _gpu_probe_cache = (now, gpu_count, gpu_names)
            _write_gpu_info_to_local(gpu_count, gpu_names)
        else:
            # 探测失败回退: 读 local.json 旧缓存
            local_path = DATA_DIR / "config" / "local.json"
            if local_path.exists():
                try:
                    local_cfg = json.loads(local_path.read_text(encoding="utf-8"))
                    gpu_count = int(local_cfg.get("gpu_count", 0))
                    gpu_names = list(local_cfg.get("gpu_names", []) or [])
                except Exception:
                    pass
    gpu_available = gpu_count > 0
    if not gpu_available:
        warnings.append("无 GPU 检测到：实时叠加/动态范围压缩已禁用（基础功能不受影响）")

    # OBS 插件检测
    obs_cfg = load_obs_config()
    plugin_file = Path(obs_cfg.plugin_path) / "astrohub-obs.dll"
    obs_plugin_installed = plugin_file.exists()
    if not obs_plugin_installed:
        warnings.append("astrohub-obs.dll 未安装：OBS 滤镜功能（叠加/拉伸）不可用")

    advanced_ok = gpu_available and obs_plugin_installed
    return {
        "success": True,
        "data": {
            "gpu_available": gpu_available,
            "gpu_names": gpu_names,
            "obs_plugin_installed": obs_plugin_installed,
            "plugin_path": obs_cfg.plugin_path,
            "capabilities": {
                "gpu_stack": advanced_ok,
                "gpu_stretch": advanced_ok,
                "histogram": True,  # CPU 计算，始终可用
            },
            "warnings": warnings,
        },
    }


@api_router.get("/system/nics", summary="网络接口列表", tags=["System"])
async def get_nics() -> dict:
    """获取所有网络接口列表,按优先级排序。"""
    try:
        from src.core.net_detector import get_all_nics
        nics = get_all_nics()
        return {"success": True, "data": nics, "total": len(nics)}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/system/default-ip", summary="推荐目标 IP", tags=["System"])
async def get_default_ip() -> dict:
    """获取推荐的目标 IP 地址。"""
    try:
        from src.core.net_detector import suggest_target_ip
        ip = suggest_target_ip()
        return {"success": True, "data": {"ip": ip}}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/system/operations", summary="操作日志 (最近10条)", tags=["System"])
async def get_operations_log() -> dict:
    """获取最近 10 条操作日志。"""
    return {"success": True, "data": list(reversed(operations_log)), "total": len(operations_log)}


@api_router.get("/stats/io", summary="设备 IO 统计", tags=["System"])
async def get_io_stats():
    """获取 IO 监控统计（IOMonitor + DeviceReader）。"""
    from src.core.io_monitor import IOMonitor

    # IOMonitor 统计（实时速率）
    monitor = IOMonitor.get_instance()
    io_stats = monitor.get_stats()

    # DeviceReader 统计（累计计数）
    reader = get_device_reader()
    reader_counters = reader.get_counters() if reader else {}

    return {
        "success": True,
        "data": io_stats,
        "counters": reader_counters,
    }


@api_router.get("/diagnostics/latency", summary="延迟诊断数据 (v10.54-diag)", tags=["Diagnostics"])
async def get_latency_diagnostics() -> dict:
    """v10.54-diag: 获取 PTZ 延迟诊断数据（调用链各层计时 + 事件循环 + 线程池）。"""
    from src.diagnostics.event_loop_monitor import get_event_loop_monitor, get_thread_pool_monitor
    from src.diagnostics.latency_tracker import get_latency_tracker

    tracker = get_latency_tracker()
    el_monitor = get_event_loop_monitor()
    tp_monitor = get_thread_pool_monitor()

    # 拍摄线程池快照
    tp_snapshot = tp_monitor.take_snapshot()

    return {
        "success": True,
        "latency_stats": tracker.get_stats(),
        "recent_records": tracker.get_recent(20),
        "event_loop": el_monitor.get_report(),
        "thread_pool": {
            "snapshot": tp_snapshot,
            "report": tp_monitor.get_report(),
        },
    }


# v10.54-fix: 全局 SystemMonitor 单例，确保 GPU 缓存生效
_system_monitor: SystemMonitor | None = None


def _get_system_monitor() -> SystemMonitor:
    """获取全局 SystemMonitor 单例。"""
    global _system_monitor
    if _system_monitor is None:
        from src.advanced.monitor import SystemMonitor
        _system_monitor = SystemMonitor(device_reader=get_device_reader())
    return _system_monitor


# v10.55-fix: /stats/hardware 端点缓存，避免频繁采集导致 GIL 竞争冻结事件循环
_hardware_stats_cache: dict | None = None
_hardware_stats_cache_time: float = 0.0
_hardware_stats_cache_ttl: float = 10.0  # 缓存 10 秒


@api_router.get("/stats/hardware", summary="硬件监控统计 (v10.50)", tags=["System"])
async def get_hardware_stats() -> dict:
    """设备管理页硬件监控（5秒刷新，离开停止）。

    不产生任何 ISAPI 设备 IO，所有数据来自本地 psutil + nvidia-smi/wmic。
    返回: CPU/GPU/RAM/网卡IO/存储盘IO/进程资源/ISAPI统计。

    v10.55-fix: 10秒缓存 + to_thread，避免阻塞事件循环。
    """
    global _hardware_stats_cache, _hardware_stats_cache_time

    # 缓存命中：直接返回，零开销
    import time as _time
    now = _time.time()
    if _hardware_stats_cache is not None and (now - _hardware_stats_cache_time) < _hardware_stats_cache_ttl:
        return _hardware_stats_cache

    monitor = _get_system_monitor()

    def _collect_all_sync() -> dict:
        """在后台线程中执行所有阻塞操作。"""
        import json as _json

        import psutil

        from src.ptz.core.system_info import get_ram_detail, snap_ram_gb

        # 1) 读取配置
        nic_name = None
        record_path = None
        cpu_model = ""
        try:
            from src.config_paths import DATA_DIR
            local_path = DATA_DIR / "config" / "local.json"
            if local_path.exists():
                local_cfg = _json.loads(local_path.read_text(encoding="utf-8"))
                selected_nic = local_cfg.get("selected_nic")
                if isinstance(selected_nic, dict):
                    nic_name = selected_nic.get("name")
                elif isinstance(selected_nic, str) and selected_nic:
                    nic_name = selected_nic
                # v1.0.8: 已存配置命中虚拟网卡黑名单或不存在时回退物理网卡
                if nic_name:
                    import psutil as _psutil

                    from src.core.net_detector import get_default_nic, is_virtual_nic
                    if is_virtual_nic(nic_name) or nic_name not in _psutil.net_if_addrs():
                        nic_name = get_default_nic() or nic_name
                cpu_model = local_cfg.get("cpu_model", "")
            if not cpu_model:
                localhost_path = DATA_DIR / "reports" / "localhost.json"
                if localhost_path.exists():
                    localhost_cfg = _json.loads(localhost_path.read_text(encoding="utf-8"))
                    cpu_model = localhost_cfg.get("cpu_model", "")
            storage_path = DATA_DIR / "config" / "storage.json"
            if storage_path.exists():
                storage_cfg = _json.loads(storage_path.read_text(encoding="utf-8"))
                record_path = storage_cfg.get("record_path")
        except Exception:
            pass

        # 2) SystemMonitor 同步采集（GPU + psutil）
        data = monitor.collect_all_sync(nic_name=nic_name, record_path=record_path)

        # 3) psutil CPU/内存 + RAM 频率（wmic）
        cpu_usage = psutil.cpu_percent(interval=0)
        mem = psutil.virtual_memory()
        try:
            ram_speed_mhz = get_ram_detail().get("speed_mhz", 0)
        except Exception:
            ram_speed_mhz = 0

        # 4) ISAPI 统计（纯内存）
        from src.core.io_monitor import IOMonitor
        isapi_stats = IOMonitor.get_instance().get_stats().get("isapi", {})

        return {
            "success": True,
            "data": {
                "cpu": {"model": cpu_model, "usage_percent": round(cpu_usage, 1)},
                "ram": {
                    "total_gb": snap_ram_gb(mem.total / (1024 ** 3)),
                    "speed_mhz": ram_speed_mhz,
                    "usage_percent": round(mem.percent, 1),
                    "used_gb": round(mem.used / (1024 ** 3), 1),
                },
                "gpu": data["gpu"],
                "nic": data["net_io"],
                "storage": data["storage_io"],
                "processes": data["processes"],
                "isapi": isapi_stats,
            },
        }

    result = await asyncio.to_thread(_collect_all_sync)

    # 更新缓存
    _hardware_stats_cache = result
    _hardware_stats_cache_time = _time.time()

    return result


@health_router.get("/log/operations/file", summary="从文件读取操作日志")
async def get_operation_log_file(lines: int = 50) -> dict:
    """从日志文件读取最近N条操作日志."""
    from datetime import datetime

    from src.operation_logger import LOG_DIR

    today = datetime.now().strftime("%Y%m%d")
    log_file = LOG_DIR / f"operation_{today}.log"

    if not log_file.exists():
        return {"success": True, "data": [], "total": 0}

    try:
        all_lines = await asyncio.to_thread(_read_lines, log_file)

        recent = all_lines[-lines:] if len(all_lines) > lines else all_lines
        recent.reverse()

        entries = []
        for line in recent:
            line = line.strip()
            if not line:
                continue
            entries.append(line)

        return {"success": True, "data": entries, "total": len(entries)}
    except Exception as e:
        return {"success": False, "message": str(e), "data": [], "total": 0}


# ================================================================ #
#  PTZ 控制端点
# ================================================================ #


@api_router.post("/ptz/{mac}/move", summary="PTZ 移动", tags=["PTZ"])
async def ptz_move(mac: str, req: PTZMoveRequest) -> dict:
    """PTZ 方向移动控制。v9.77: to_thread 避免阻塞事件循环。"""
    import asyncio
    import time as _time
    _t_entry = _time.perf_counter()
    try:
        mgr: PTZDeviceController | None = _managers.get("ptz_controller")
        if not mgr:
            return {"success": False, "message": "PTZDeviceController未初始化"}

        _t_resolve = _time.perf_counter()
        target_ip = _resolve_device_id_to_ip(mgr, mac)
        _resolve_ms = (_time.perf_counter() - _t_resolve) * 1000
        if not target_ip:
            return {"success": False, "message": f"无法解析设备标识: {mac}"}

        # v1.1.25: 入队前捕获 generation——stop 提交会 bump，
        # worker 执行时落后则自废（防网络乱序导致 move 晚于 stop 到达）
        _cmd_gen = mgr.current_cmd_gen()

        _t_to_thread = _time.perf_counter()
        # v10.55: 使用 PTZ 专用线程池；v1.1.25: 单 FIFO 通道保证与 stop 的顺序
        import functools
        _loop = asyncio.get_event_loop()
        result = await _loop.run_in_executor(
            _ptz_executor,
            functools.partial(mgr.ptz_move, target_ip, direction=req.direction,
                              speed=req.speed, cmd_gen=_cmd_gen),
        )
        _to_thread_ms = (_time.perf_counter() - _t_to_thread) * 1000
        _total_ms = (_time.perf_counter() - _t_entry) * 1000

        # v10.54-diag: 路由层分段计时
        from src.logger import get_logger as _get_logger
        _get_logger(__name__).info(
            f"PTZ_MOVE TIMING: resolve={_resolve_ms:.1f}ms to_thread={_to_thread_ms:.1f}ms total={_total_ms:.1f}ms"
        )
        try:
            from src.diagnostics.latency_tracker import get_latency_tracker
            get_latency_tracker().record(
                "router_move",
                resolve_ms=_resolve_ms,
                to_thread_ms=_to_thread_ms,
                total_ms=_total_ms,
            )
        except ImportError:
            pass

        return result
    except Exception as e:
        import traceback
        traceback.print_exc()
        return {"success": False, "message": f"PTZ 移动异常: {e}"}


@api_router.post("/ptz/{mac}/home", summary="PTZ 归位", tags=["PTZ"])
async def ptz_gotohome(mac: str) -> dict:
    """PTZ 归位(预置点 10)。"""
    import asyncio
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    result = await asyncio.to_thread(mgr.ptz_home, target_ip)
    if result.get("success"):
        log_info("ptz", "home", {"device": mac})
    return result


@api_router.post("/ptz/{mac}/stop", summary="PTZ 停止", tags=["PTZ"])
async def ptz_stop(mac: str) -> dict:
    """PTZ 停止所有移动。v9.77: to_thread 避免阻塞事件循环。"""
    import asyncio
    import time as _time
    _t_entry = _time.perf_counter()
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    _t_resolve = _time.perf_counter()
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    _resolve_ms = (_time.perf_counter() - _t_resolve) * 1000
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    _t_to_thread = _time.perf_counter()
    # v10.59: stop 使用独立线程池，不与 move 共享排队
    _loop = asyncio.get_event_loop()
    result = await _loop.run_in_executor(_ptz_stop_executor, mgr.ptz_stop, target_ip)
    _to_thread_ms = (_time.perf_counter() - _t_to_thread) * 1000
    _total_ms = (_time.perf_counter() - _t_entry) * 1000

    # v10.54-diag: 路由层分段计时
    from src.logger import get_logger as _get_logger
    _get_logger(__name__).info(
        f"PTZ_STOP TIMING: resolve={_resolve_ms:.1f}ms to_thread={_to_thread_ms:.1f}ms total={_total_ms:.1f}ms"
    )
    try:
        from src.diagnostics.latency_tracker import get_latency_tracker
        get_latency_tracker().record(
            "router_stop",
            resolve_ms=_resolve_ms,
            to_thread_ms=_to_thread_ms,
            total_ms=_total_ms,
        )
    except ImportError:
        pass

    return result


class FocusModeRequest(BaseModel):
    """对焦模式请求"""
    mode: str  # manual, auto, semiauto


@api_router.post("/ptz/{mac}/focus/mode", summary="设置对焦模式", tags=["PTZ"])
async def set_focus_mode(mac: str, req: FocusModeRequest) -> dict:
    """设置对焦模式(手动/自动/半自动)。委托 ISAPIPTZController.set_focus_mode()。"""
    import asyncio

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl = mgr._controllers.get(target_ip)
    if not ctrl:
        return {"success": False, "message": "设备未连接"}

    try:
        ok = await asyncio.to_thread(ctrl.set_focus_mode, req.mode)
        if ok:
            log_info("ptz", "focus_mode", {"device": mac, "mode": req.mode})
            return {"success": True, "message": f"对焦模式已设置为 {req.mode}"}
        return {"success": False, "message": "设置对焦模式失败"}
    except Exception as e:
        return {"success": False, "message": f"设置对焦模式失败: {e}"}


@api_router.get("/ptz/{mac}/focus/mode", summary="获取对焦模式", tags=["PTZ"])
async def get_focus_mode(mac: str) -> dict:
    """获取当前对焦模式。委托 ISAPIPTZController.get_focus_mode()。"""
    import asyncio

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl = mgr._controllers.get(target_ip)
    if not ctrl:
        return {"success": False, "message": "设备未连接"}

    try:
        mode = await asyncio.to_thread(ctrl.get_focus_mode)
        return {"success": True, "mode": mode}
    except Exception as e:
        return {"success": False, "message": f"获取对焦模式失败: {e}"}


@api_router.get("/ptz/{mac}/presets", summary="获取预置点列表", tags=["PTZ"])
async def ptz_list_presets(mac: str) -> dict:
    """获取设备所有预置点列表."""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    return await asyncio.to_thread(mgr.ptz_list_presets, target_ip)


@api_router.post("/ptz/{mac}/preset/{preset_id}", summary="预置位", tags=["PTZ"])
async def ptz_preset(mac: str, preset_id: int) -> dict:
    """移动到指定预置点。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # BUG-A 修复: 支持 MAC 地址作为 device_id
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    result = await asyncio.to_thread(mgr.ptz_preset, target_ip, preset_id=preset_id)
    if result.get("success"):
        log_info("ptz", "preset_goto", {"device": mac, "preset_id": preset_id})
    return result


@api_router.post("/ptz/{mac}/preset/{preset_id}/set", summary="保存预置位", tags=["PTZ"])
async def ptz_set_preset(mac: str, preset_id: int, name: str = "") -> dict:
    """设置当前位置为指定预置点。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    result = await asyncio.to_thread(mgr.set_preset, target_ip, preset_id=preset_id, name=name)
    if result.get("success"):
        log_info("ptz", "preset_set", {"device": mac, "preset_id": preset_id, "name": name})
    return result


@api_router.post("/ptz/{mac}/absolute", summary="绝对位置", tags=["PTZ"])
async def ptz_absolute(mac: str, req: PTZAbsoluteRequest) -> dict:
    """绝对坐标移动(Pan/Tilt/Zoom)。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # BUG-A 修复: 支持 MAC 地址作为 device_id
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    return await asyncio.to_thread(
        mgr.ptz_absolute,
        target_ip,
        pan=req.pan,
        tilt=req.tilt,
        zoom=req.zoom,
    )


@api_router.get("/ptz/{mac}/position", summary="获取 PTZ 位置", tags=["PTZ"])
async def get_ptz_position(mac: str) -> dict:
    """获取 PTZ 当前位置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    pos = reader.get_position()
    if not pos:
        return {"success": False, "message": "未获取到位置数据"}

    result = {"success": True, "data": pos}

    # 检查是否有校准数据,计算真实 alt/az
    try:
        mgr = _managers.get("ptz_controller")
        if mgr:
            target_ip = _resolve_device_id_to_ip(mgr, mac)
            if target_ip:
                session = get_staralign_manager().sessions.get(target_ip)
                if session and session.get("solver", None) and session["solver"]._solved:
                    solver = session["solver"]
                    pan = pos.get("pan", 0)
                    tilt = pos.get("tilt", 0)
                    true_az, true_alt = solver.ptz_to_true(pan / 10, tilt / 10)
                    result["data"]["az"] = round(true_az, 4)
                    result["data"]["alt"] = round(true_alt, 4)
                    result["data"]["calibrated"] = True
                else:
                    result["data"]["calibrated"] = False
    except Exception:
        result["data"]["calibrated"] = False

    return result


@api_router.get("/device/{mac}/initial-state", summary="获取设备初始状态", tags=["Device"])
async def get_device_initial_state(mac: str) -> dict:
    """v10.03: 返回设备启动时的完整状态快照，供前端初始化渲染。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    position = reader.get_position()
    image = reader.get_image_settings()
    osd = reader.get_osd_info()
    config = reader.get("device_config")

    return {
        "success": True,
        "data": {
            "position": position,
            "image": image,
            "osd": osd,
            "config": config,
        },
    }


# 海康 ISAPI PTZ OSD 显示端点 — v11.12: 业务逻辑迁入 ISAPIPTZController


@api_router.get("/ptz/{mac}/osd/ptz", summary="获取 PTZ 坐标 OSD 状态", tags=["PTZ"])
async def ptz_osd_ptz_get(mac: str) -> dict:
    """获取 PTZ 坐标 OSD 状态。v11.12: 委托给 ISAPIPTZController.get_osd_ptz_status。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}

    try:
        result = await asyncio.to_thread(ctrl.get_osd_ptz_status)
        if result.get("success"):
            result["data"] = {"enabled": result["enabled"], "device_id": mac, "ip": target_ip}
        return result
    except Exception as e:
        return {"success": False, "message": f"OSD 读取异常: {e}"}


@api_router.post("/ptz/{mac}/osd/ptz", summary="切换 PTZ 坐标 OSD", tags=["PTZ"])
async def ptz_osd_ptz_toggle(mac: str, body: dict = None) -> dict:
    """切换 PTZ 坐标 OSD。v11.12: 委托给 ISAPIPTZController.toggle_osd_ptz。"""
    enabled = (body or {}).get("enabled", True) if body else True

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}

    try:
        result = await asyncio.to_thread(ctrl.toggle_osd_ptz, enabled)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("osd_ptz", {"enabled": enabled})
            log_info("ptz", "osd_ptz", {"device": mac, "enabled": enabled})
            return {"success": True, "message": f"PTZ 坐标 OSD 已{'开启' if enabled else '关闭'}", "enabled": enabled}
        return result
    except Exception as e:
        return {"success": False, "message": f"PTZ OSD 切换异常: {e}"}


@api_router.post("/ptz/{mac}/osd/info", summary="切换 OSD 信息显示", tags=["PTZ"])
async def ptz_osd_info_toggle(mac: str, body: dict = None) -> dict:
    """切换 OSD 信息显示。v11.12: 委托给 ISAPIPTZController.toggle_osd_info。"""
    enabled = (body or {}).get("enabled", True) if body else True

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}

    try:
        result = await asyncio.to_thread(ctrl.toggle_osd_info, enabled)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("osd_info", {"enabled": enabled})
            log_info("ptz", "osd_info", {"device": mac, "enabled": enabled})
            return {"success": True, "message": f"OSD 信息显示已{'开启' if enabled else '关闭'}", "enabled": enabled}
        return result
    except Exception as e:
        return {"success": False, "message": f"OSD 信息切换异常: {e}"}


# ============================================================
# 画面调整
# ============================================================

class ImageAdjustRequest(BaseModel):
    """画面调整请求"""
    brightness: int | None = None  # 0-100
    contrast: int | None = None    # 0-100
    saturation: int | None = None  # 0-100


@api_router.get("/ptz/{mac}/image/settings", summary="获取画面设置", tags=["Image"])
async def get_image_settings(mac: str) -> dict:
    """获取当前画面参数(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    return {"success": True, "data": image_data}


@api_router.put("/ptz/{mac}/image/settings", summary="更新画面设置", tags=["Image"])
async def update_image_settings(mac: str, req: ImageAdjustRequest) -> dict:
    """更新画面参数。委托 ISAPIPTZController.set_color()。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl = mgr._controllers.get(target_ip)
    if not ctrl:
        return {"success": False, "message": "设备未连接"}

    try:
        result = await asyncio.to_thread(
            ctrl.set_color, req.brightness, req.contrast, req.saturation
        )
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("color", result["data"])
            log_info("image", "color", {"device": mac, "updated": result["data"]})
            return {"success": True, "message": "画面设置已更新", "data": result["data"]}
        return result
    except Exception as e:
        return {"success": False, "message": f"更新画面设置异常: {e}"}


@api_router.post("/ptz/{mac}/capture", summary="PTZ 截图(ISAPI 原生)", tags=["PTZ"])
async def ptz_capture(
    mac: str,
    click_x: float | None = Query(None, description="点击位置x (0~1)"),
    click_y: float | None = Query(None, description="点击位置y (0~1)"),
    crop_size: int = Query(100, description="裁剪区域像素尺寸")
) -> dict:
    """通过海康 ISAPI 原生方法截取 PTZ 设备当前视频帧。

    使用: GET /ISAPI/Streaming/Channels/{channel}/picture
    不再依赖 ffmpeg。
    """
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}

    client = mgr._clients.get(target_ip)
    if not client:
        return {"success": False, "message": f"ISAPI client not found for {target_ip}"}

    try:
        # v10.52: 使用 asyncio.to_thread 避免阻塞事件循环
        import asyncio
        jpeg_bytes = await asyncio.to_thread(client.capture_picture)
        if not jpeg_bytes:
            return {"success": False, "message": "ISAPI 截图失败:设备未返回图像数据"}

        # JPEG 头验证
        if jpeg_bytes[:3] != b"\xff\xd8\xff":
            return {"success": False, "message": "返回的数据不是有效的 JPEG"}

        # v9.75: 点击指向截图 - 裁剪指定区域
        crop_base64 = None
        if click_x is not None and click_y is not None:
            # v10.73: cv2 图像处理是 CPU 密集操作，移到线程池避免阻塞事件循环
            def _process_crop() -> dict:
                import base64

                import cv2
                import numpy as np

                bgr = cv2.imdecode(np.frombuffer(jpeg_bytes, np.uint8), cv2.IMREAD_COLOR)
                if bgr is None:
                    return {"error": "图像解码失败"}
                img_h, img_w = bgr.shape[:2]
                px = int(click_x * img_w)
                py = int(click_y * img_h)
                half = crop_size // 2
                x1 = max(0, px - half)
                y1 = max(0, py - half)
                x2 = min(img_w, px + half)
                y2 = min(img_h, py + half)
                cropped = bgr[y1:y2, x1:x2]
                _, buf = cv2.imencode('.jpg', cropped)
                crop_bytes = buf.tobytes()
                crop_b64 = base64.b64encode(crop_bytes).decode('utf-8')
                return {"crop_base64": crop_b64}

            crop_result = await asyncio.to_thread(_process_crop)
            if "error" in crop_result:
                return {"success": False, "message": crop_result["error"]}
            crop_base64 = crop_result["crop_base64"]

        # v7.105: 保存文件(使用配置路径 + 日期子目录)
        from datetime import datetime

        from src.core.file_naming import generate_filename

        record_path = _resolve_storage_path()
        from pathlib import Path
        record_dir = Path(record_path)

        # 日期子目录
        date_str = datetime.now().strftime("%Y-%m-%d")
        date_dir = record_dir / date_str
        date_dir.mkdir(parents=True, exist_ok=True)

        # v7.108: 获取设备名称(优先 name,回退 device_name,再回退 model)
        creds = mgr.get_credentials(target_ip)
        device_name = (creds.get("name", "") or creds.get("device_name", "") or creds.get("model", "") or "").replace(" ", "_").replace("/", "_").replace("\\", "_") or target_ip

        filename = generate_filename(target_name=device_name, extension=".jpg")
        filepath = date_dir / filename
        filepath.write_bytes(jpeg_bytes)
        file_size = filepath.stat().st_size

        log_info("ptz", "capture", {"device": mac, "filename": filename})
        result = {
            "success": True,
            "message": f"截图成功: {mac}",
            "data": {
                "mac": mac,
                "image_path": str(filepath),
                "image_size": file_size,
                "filename": filename,
                "relative_path": f"{date_str}/{filename}",
                "format": "jpeg",
                "verified": True,
            },
        }

        # v9.75: 返回裁剪区域的base64
        if crop_base64:
            result["data"]["image_base64"] = crop_base64
            result["data"]["crop_x"] = click_x
            result["data"]["crop_y"] = click_y
            result["data"]["crop_size"] = crop_size

        return result
    except Exception as e:
        return {"success": False, "message": f"截图异常: {e}"}


# ================================================================ #
#  PTZ 录像端点
# ================================================================ #


@api_router.post("/ptz/{mac}/record/start", summary="启动录像(ISAPI 官方)", tags=["PTZ"])
async def ptz_record_start(mac: str, req: PTZRecordRequest | None = None) -> dict:
    """v7.107: 通过海康 ISAPI 官方方法启动摄像头录像。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    target_name = req.target_name if req else ""
    log_info("ptz", "record_start", {"device": mac, "target": target_name})
    # v10.52: 使用 asyncio.to_thread 避免阻塞事件循环
    import asyncio
    return await asyncio.to_thread(mgr.start_recording, target_ip, target_name=target_name)


@api_router.post("/ptz/{mac}/record/stop", summary="停止录像(ISAPI 官方)", tags=["PTZ"])
async def ptz_record_stop(mac: str) -> dict:
    """停止 FFmpeg 录制进程。如有 FTP 配置则自动上传。
    """
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")  # type: ignore[assignment]
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    log_info("ptz", "record_stop", {"device": mac})
    # v10.52: 使用 asyncio.to_thread 避免阻塞事件循环
    import asyncio
    return await asyncio.to_thread(mgr.stop_recording, target_ip)


# ================================================================ #
#  CLICK_POINT 点击指向 v9.75
# ================================================================ #


class ClickPointMoveRequest(BaseModel):
    """点击指向请求: 比例坐标 (0~1)。"""
    x: float  # 水平比例 0~1
    y: float  # 垂直比例 0~1


class ClickPointApplyRequest(BaseModel):
    """校准点应用请求。"""
    point_index: int  # 0~4


@api_router.post("/click_point/{mac}/move", summary="点击指向", tags=["ClickPoint"])
async def click_point_move(mac: str, req: ClickPointMoveRequest) -> dict:
    """点击屏幕点 → PTZ绝对值移动。v11.12: 业务逻辑迁入 click_point.py。"""
    try:
        mgr: PTZDeviceController | None = _managers.get("ptz_controller")
        if not mgr:
            return {"success": False, "message": "PTZDeviceController未初始化"}

        target_ip = _resolve_device_id_to_ip(mgr, mac)
        if not target_ip:
            return {"success": False, "message": f"无法解析设备标识: {mac}"}

        ctrl, err = mgr._get_controller(target_ip)
        if err:
            return {"success": False, "message": err}

        client = mgr._clients.get(target_ip)
        if not client:
            return {"success": False, "message": f"ISAPI client not found for {target_ip}"}

        mac_clean = normalize_mac(mac)
        session = get_session(mac_clean)
        # v11.30: 位置只从 DeviceReader 缓存读（设计约束：禁止端点直接 ISAPI 查位置）
        _reader = get_device_reader()
        _pos = _reader.get_position() if _reader else None
        return await asyncio.to_thread(session.execute_move, client, ctrl, mac, req.x, req.y, _pos)
    except Exception as e:
        import traceback
        error_msg = f"点击指向异常: {str(e)}"
        print(f"[ClickPoint] {error_msg}")
        traceback.print_exc()
        return {"success": False, "message": error_msg}


@api_router.post("/click_point/{mac}/apply", summary="应用校准点", tags=["ClickPoint"])
async def click_point_apply(mac: str, req: ClickPointApplyRequest) -> dict:
    """用户手动对准后点击"应用" → 记录校准点 + 重建模型。v11.12: 业务逻辑迁入 click_point.py。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}

    mac_clean = normalize_mac(mac)
    session = get_session(mac_clean)

    # v11.30: 位置只从 DeviceReader 缓存读
    _reader = get_device_reader()
    _pos = _reader.get_position() if _reader else None

    # v11.12: 业务逻辑迁入 execute_apply；v11.30: 延迟落盘，apply 只记内存，
    # 持久化由 /click_point/{mac}/finish 在用户点「完成」时执行
    result = await asyncio.to_thread(session.execute_apply, ctrl, req.point_index, _pos)

    return result


@api_router.get("/click_point/{mac}/status", summary="点击指向状态", tags=["ClickPoint"])
async def click_point_status(mac: str) -> dict:
    """返回当前校准模型状态、校准点列表、当前zoom。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}

    mac_clean = normalize_mac(mac)
    session = get_session(mac_clean)

    # v11.30: 位置只从 DeviceReader 缓存读（零 ISAPI）——本端点被前端
    # 1s/2s/10s 定时器高频命中，绝不能再打设备
    _reader = get_device_reader()
    real_pos = _reader.get_position() if _reader else None
    current_zoom = real_pos.get("zoom", 0) if real_pos else 0

    # 校准点列表 (含预期P/T)
    points_display = []
    from src.advanced.click_point import _get_fov
    for i, p in enumerate(session.points):
        fov_h, fov_v = _get_fov(p["img_w"], p["img_h"], zoom=int(current_zoom))
        exp_pan, exp_tilt = session.predict(
            p["px"], p["py"], p["img_w"], p["img_h"],
            current_pan=p["pan"],
            current_tilt=p["tilt"],
        )
        points_display.append({
            "index": i,
            "px": round(p["px"], 1),
            "py": round(p["py"], 1),
            "pan": round(p["pan"], 4),
            "tilt": round(p["tilt"], 4),
            "expected_pan": round(exp_pan, 4),
            "expected_tilt": round(exp_tilt, 4),
        })

    # 距离校验
    distance_to_last: float | None = None
    distance_ok: bool | None = None
    if session.points and real_pos:
        last_pt = session.points[-1]
        cur_pan_deg = (real_pos.get("pan", 0) or 0) / 10.0
        cur_tilt_deg = (real_pos.get("tilt", 0) or 0) / 10.0
        distance_to_last = round(
            ((cur_pan_deg - last_pt["pan"]) ** 2 + (cur_tilt_deg - last_pt["tilt"]) ** 2) ** 0.5,
            1,
        )
        distance_ok = distance_to_last >= MIN_CALIBRATION_DISTANCE_DEG

    return {
        "success": True,
        "data": {
            "current_zoom": round(current_zoom, 1),
            "point_count": len(session.points),
            "points": points_display,
            "model_type": session.model_type,
            "updated_at": session.updated_at,
            "distance_to_last_point": distance_to_last,
            "distance_ok": distance_ok,
        }
    }


@api_router.post("/click_point/{mac}/reset", summary="重置点击指向", tags=["ClickPoint"])
async def click_point_reset(mac: str) -> dict:
    """清空本轮校准数据（v11.30: 仅内存，不写盘——重来/取消不碰磁盘校准）。"""
    mac_clean = normalize_mac(mac)
    reset_session(mac_clean)
    return {"success": True, "message": "点击指向校准已重置"}


@api_router.post("/click_point/{mac}/finish", summary="完成点击指向校准(落盘)", tags=["ClickPoint"])
async def click_point_finish(mac: str) -> dict:
    """v11.30: 完成校准 — 将内存会话落盘并清理待应用数据。

    延迟落盘语义：校准期间零写盘，仅本端点在用户点「完成」时写文件。
    """
    mac_clean = normalize_mac(mac)
    session = get_session(mac_clean)
    if not session.points:
        return {"success": False, "message": "无校准点可保存"}

    await asyncio.to_thread(finish_session, mac_clean)
    log_info("click_point", "finish", {"device": mac, "points": len(session.points)})
    return {
        "success": True,
        "message": f"点击指向校准已保存（{len(session.points)} 个点）",
        "data": {"point_count": len(session.points)},
    }


@api_router.post("/click_point/{mac}/capture", summary="点击指向截图(不落盘)", tags=["ClickPoint"])
async def click_point_capture(
    mac: str,
    click_x: float | None = Query(None, description="点击位置x (0~1)"),
    click_y: float | None = Query(None, description="点击位置y (0~1)"),
    crop_size: int = Query(100, description="裁剪区域像素尺寸")
) -> dict:
    """点击指向专用截图 - 不落盘,仅返回base64。

    与 /ptz/{mac}/capture 区别:
    - 不保存全图到磁盘
    - 仅返回裁剪区域的base64编码
    - 用于校准流程中的实时预览
    """
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}

    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}

    client = mgr._clients.get(target_ip)
    if not client:
        return {"success": False, "message": f"ISAPI client not found for {target_ip}"}

    try:
        # v10.54: to_thread 避免阻塞事件循环
        jpeg_bytes = await asyncio.to_thread(client.capture_picture)
        if not jpeg_bytes:
            return {"success": False, "message": "ISAPI截图失败"}

        if jpeg_bytes[:3] != b"\xff\xd8\xff":
            return {"success": False, "message": "返回的数据不是有效的JPEG"}

        # v10.73: cv2 图像处理（解码+裁剪+编码+base64）是 CPU 密集操作，
        # 1080p JPEG 约 150-500ms，直接在事件循环中执行会冻结所有 API 请求。
        # 移到线程池执行，保持事件循环响应。
        def _process_image() -> dict:
            import base64

            import cv2
            import numpy as np

            bgr = cv2.imdecode(np.frombuffer(jpeg_bytes, np.uint8), cv2.IMREAD_COLOR)
            if bgr is None:
                return {"error": "图像解码失败"}
            img_h, img_w = bgr.shape[:2]
            px = int(click_x * img_w) if click_x is not None else img_w // 2
            py = int(click_y * img_h) if click_y is not None else img_h // 2
            half = crop_size // 2
            x1 = max(0, px - half)
            y1 = max(0, py - half)
            x2 = min(img_w, px + half)
            y2 = min(img_h, py + half)
            cropped = bgr[y1:y2, x1:x2]
            _, buf = cv2.imencode('.jpg', cropped)
            crop_b64 = base64.b64encode(buf.tobytes()).decode('utf-8')
            return {
                "img_w": img_w, "img_h": img_h,
                "px": px, "py": py,
                "x1": x1, "y1": y1,
                "x2": x2, "y2": y2,
                "crop_base64": crop_b64,
            }

        result = await asyncio.to_thread(_process_image)
        if "error" in result:
            return {"success": False, "message": result["error"]}

        img_w, img_h = result["img_w"], result["img_h"]
        px, py = result["px"], result["py"]
        x1, y1 = result["x1"], result["y1"]
        x2, y2 = result["x2"], result["y2"]
        crop_base64 = result["crop_base64"]

        # v11.30: 位置只从 DeviceReader 缓存读（截图时刻的位置由缓存提供，
        # 升频轮询保证缓存新鲜度）
        _reader = get_device_reader()
        pos = _reader.get_position() if _reader else None
        if pos:
            current_pan = pos.get("pan", 0) / 10  # 0.1度 → 度
            current_tilt = pos.get("tilt", 0) / 10
            current_zoom = pos.get("zoom", 1)
        else:
            current_pan = 0
            current_tilt = 0
            current_zoom = 1

        # 存储到 session._pending_click
        mac_clean = normalize_mac(mac)
        session = get_session(mac_clean)
        # 使用开始校准时记录的zoom,如果没有则用当前zoom
        start_zoom = getattr(session, '_start_zoom', None)
        record_zoom = start_zoom if start_zoom is not None else int(current_zoom)
        session._pending_click = {
            "px": px, "py": py,
            "img_w": img_w, "img_h": img_h,
            "current_pan": current_pan,
            "current_tilt": current_tilt,
            "target_pan": current_pan,
            "target_tilt": current_tilt,
            "zoom": record_zoom,
        }

        return {
            "success": True,
            "message": "截图成功",
            "data": {
                "img_w": img_w,
                "img_h": img_h,
                "click_x": click_x,
                "click_y": click_y,
                "crop_x": x1,
                "crop_y": y1,
                "crop_w": x2 - x1,
                "crop_h": y2 - y1,
                "crop_base64": crop_base64,
            }
        }
    except Exception as e:
        return {"success": False, "message": f"截图异常: {str(e)}"}


# ================================================================ #
#  ASCOM 端点
# ================================================================ #


@api_router.post("/ascom/{ascom_type}/connect", summary="连接 ASCOM 设备", tags=["ASCOM"])
async def ascom_connect(ascom_type: str) -> dict:
    """连接 ASCOM 设备。

    v10.74: ASCOM 功能暂未实现，返回占位响应。
    """
    return {
        "success": False,
        "message": f"ASCOM 功能暂未实现 (type={ascom_type})",
        "data": {"type": ascom_type, "connected": False},
    }


# ================================================================ #
#  ASCOM Telescope 控制端点 (P1 集成)
# ================================================================ #


@api_router.post("/ascom/telescope/slew", summary="望远镜 Slew 到目标坐标", tags=["ASCOM"])
async def ascom_telescope_slew(req: TelescopeSlewRequest | None = None,
                                ra: float = 0, dec: float = 0) -> dict:
    """控制望远镜 Slew 到目标赤经/赤纬。

    支持两种参数传入方式:
    1. JSON body: {"ra": 12.5, "dec": 45.0}
    2. Query params: ?ra=12.5&dec=45.0
    """
    ra_val = req.ra if req else ra
    dec_val = req.dec if req else dec

    from src.ascom.core.driver_manager import get_telescope

    try:
        scope = get_telescope()
        result = scope.slew_to_coordinates(ra_val, dec_val)
        if not result.get("success"):
            return {
                "success": False,
                "message": result.get("message", "Slew 失败"),
                "code": result.get("code", "ASCOM_TELESCOPE_SLEW_FAILED"),
            }
        log_info("ascom", "slew", {"ra": ra_val, "dec": dec_val})
        return {
            "success": True,
            "message": result.get("message", "Slew 成功"),
            "data": {"ra": ra_val, "dec": dec_val},
        }
    except Exception as e:
        return {"success": False, "message": f"Slew 异常: {e}"}


@api_router.post("/ascom/telescope/tracking", summary="设置望远镜跟踪模式", tags=["ASCOM"])
async def ascom_telescope_tracking(mode: str = "trackSidereal") -> dict:
    """设置望远镜跟踪模式。

    支持的模式:
    - trackSidereal: 恒星跟踪
    - trackLunar: 月球跟踪
    - trackSolar: 太阳跟踪
    - trackOff: 关闭跟踪
    """
    from src.ascom.constants import TrackingMode
    from src.ascom.core.driver_manager import get_telescope

    mode_map = {
        "trackSidereal": TrackingMode.SIDEREAL,
        "trackLunar": TrackingMode.LUNAR,
        "trackSolar": TrackingMode.SOLAR,
        "trackOff": TrackingMode.OFF,
    }

    tracking_mode = mode_map.get(mode)
    if tracking_mode is None:
        return {
            "success": False,
            "message": f"无效的跟踪模式: {mode} (支持: {list(mode_map.keys())})",
        }

    try:
        scope = get_telescope()
        result = scope.set_tracking_mode(tracking_mode)
        if not result.get("success"):
            return {
                "success": False,
                "message": result.get("message", "设置跟踪模式失败"),
            }
        log_info("ascom", "tracking", {"mode": mode})
        return {
            "success": True,
            "message": result.get("message", f"跟踪模式已设为 {mode}"),
            "data": {"mode": mode},
        }
    except Exception as e:
        return {"success": False, "message": f"设置跟踪模式异常: {e}"}


@api_router.post("/ascom/telescope/disconnect", summary="断开望远镜连接", tags=["ASCOM"])
async def ascom_telescope_disconnect() -> dict:
    """断开 ASCOM 望远镜连接。"""
    from src.ascom.core.driver_manager import get_telescope

    try:
        scope = get_telescope()
        result = scope.disconnect()
        return {
            "success": True,
            "message": result.get("message", "望远镜已断开"),
        }
    except Exception as e:
        return {"success": False, "message": f"断开连接异常: {e}"}


@api_router.get("/ascom/telescope/position", summary="查询望远镜位置", tags=["ASCOM"])
async def ascom_telescope_position() -> dict:
    """查询望远镜当前赤经/赤纬。"""
    from src.ascom.core.driver_manager import get_telescope

    try:
        scope = get_telescope()
        result = scope.get_position()
        if not result.get("success"):
            return {"success": False, "message": result.get("message", "获取位置失败")}
        return {"success": True, "data": result.get("data")}
    except Exception as e:
        return {"success": False, "message": f"获取位置异常: {e}"}


@api_router.post("/ascom/telescope/abort", summary="取消 Slew", tags=["ASCOM"])
async def ascom_telescope_abort() -> dict:
    """取消正在进行的 Slew 操作。"""
    from src.ascom.core.driver_manager import get_telescope

    try:
        scope = get_telescope()
        result = scope.abort_slew()
        if not result.get("success"):
            return {"success": False, "message": result.get("message", "取消 Slew 失败")}
        return {"success": True, "message": result.get("message", "Slew 已取消")}
    except Exception as e:
        return {"success": False, "message": f"取消 Slew 异常: {e}"}


@api_router.get("/settings", summary="获取系统设置", tags=["Settings"])
async def get_settings() -> dict:
    """获取当前系统配置。"""
    try:
        from src.config import HOST, PORT, WINDOW_HEIGHT, WINDOW_TITLE, WINDOW_WIDTH
        return {
            "success": True,
            "data": {
                "host": HOST,
                "port": PORT,
                "window_title": WINDOW_TITLE,
                "window_width": WINDOW_WIDTH,
                "window_height": WINDOW_HEIGHT,
            },
        }
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/settings", summary="保存系统设置", tags=["Settings"])
async def save_settings() -> dict:
    """保存系统设置。"""
    return {
        "data": {"saved": True},
        "message": "设置已保存",
    }


# === v8.69: 控制台状态持久化 ===
import json as _json

from src.config_paths import get_project_root as _get_project_root

_SNAPSHOT_FILE = _get_project_root() / "data" / "console_snapshot.json"


def _read_registry() -> dict:
    """读 registry（优先 DeviceManager 内存缓存，避免每次读盘）。"""
    dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
    if dm:
        return dm.get_registry()
    # 兑底：manager 未注入时直读磁盘（仅启动早期路径）
    from src.config_paths import REGISTRY_FILE
    if REGISTRY_FILE.exists():
        with open(REGISTRY_FILE, encoding="utf-8") as f:
            return _json.load(f)
    return {}


def _write_registry(data: dict):
    """写 registry（优先走 DeviceManager，同步刷新内存缓存）。"""
    dm: DeviceManager | None = _managers.get("device_manager")  # type: ignore[assignment]
    if dm:
        dm.update_registry(data)
        return
    from src.config_paths import REGISTRY_FILE
    REGISTRY_FILE.parent.mkdir(parents=True, exist_ok=True)
    with open(REGISTRY_FILE, "w", encoding="utf-8") as f:
        _json.dump(data, f, indent=2, ensure_ascii=False)


def _get_station_coordinates() -> tuple[float, float]:
    """获取观测站坐标(lat, lon)。

    唯一坐标来源:registry.json.geo_location,由高级功能页面唯一写入。

    Returns:
        (lat, lon): 浮点数纬度、经度
    Raises:
        ValueError: 未设置坐标时抛出
    """
    registry = _read_registry()
    geo = registry.get("geo_location", {})
    lat = geo.get("lat")
    lon = geo.get("lon")
    if lat is None or lon is None:
        raise ValueError("未设置观测站坐标,请先在高级功能页面设置")
    return float(lat), float(lon)


@api_router.get("/console/snapshot", summary="获取控制台快照", tags=["Console"])
async def get_console_snapshot() -> dict:
    """读取控制台快照。"""
    if _SNAPSHOT_FILE.exists():
        content = await asyncio.to_thread(_read_file, _SNAPSHOT_FILE)
        return {"success": True, "data": _json.loads(content)}
    return {"success": True, "data": {}}


@api_router.post("/console/snapshot", summary="保存控制台快照", tags=["Console"])
async def save_console_snapshot(data: dict) -> dict:
    """保存控制台快照。"""
    _SNAPSHOT_FILE.parent.mkdir(parents=True, exist_ok=True)
    await asyncio.to_thread(_write_file, _SNAPSHOT_FILE, _json.dumps(data, indent=2, ensure_ascii=False))
    return {"success": True}


@api_router.get("/geo", summary="获取地理坐标", tags=["Geo"])
async def get_geo_location() -> dict:
    """获取已保存的地理坐标。"""
    registry = _read_registry()
    return {"success": True, "data": registry.get("geo_location", {})}


@api_router.post("/geo", summary="保存地理坐标", tags=["Geo"])
async def save_geo_location(data: dict) -> dict:
    """保存地理坐标到 registry.json。"""
    import re
    lat_raw = str(data.get("lat", "")).strip()
    lon_raw = str(data.get("lon", "")).strip()
    ns = str(data.get("ns", "N")).strip().upper()
    ew = str(data.get("ew", "E")).strip().upper()
    if not lat_raw or not lon_raw:
        return {"success": False, "message": "经纬度不能为空"}

    def parse_dms(s):
        """解析度分秒或小数格式,返回十进制度。"""
        s = s.replace(" ", "").replace("\u3000", "")
        # 纯小数
        if re.match(r'^-?\d+\.?\d*$', s):
            return float(s)
        # 度分秒: 29°27'00" 或 29°27' 或 29°27.5'
        m = re.match(r"""^(\d+)\u00b0(\d+)['\u2019'](\d+(?:\.\d+)?)["\u201d\u2019']?$""", s)
        if m:
            d, mi, se = int(m.group(1)), int(m.group(2)), float(m.group(3) or 0)
            return d + mi / 60 + se / 3600
        # 度分: 29°27.5'
        m = re.match(r"""^(\d+)\u00b0(\d+)['\u2019'](\d+(?:\.\d+)?)["\u201d\u2019']?$""", s)
        if m:
            d, mi = int(m.group(1)), float(m.group(2))
            return d + mi / 60
        # 度: 29°
        m = re.match(r'^(\d+)°$', s)
        if m:
            return float(m.group(1))
        raise ValueError(f"无法解析: {s}")

    try:
        lat = parse_dms(lat_raw)
        lon = parse_dms(lon_raw)
        if ns == 'S':
            lat = -lat
        if ew == 'W':
            lon = -lon
    except ValueError as e:
        return {"success": False, "message": f"坐标格式错误: {e}"}

    registry = _read_registry()
    registry["geo_location"] = {"lat": lat, "lon": lon, "lat_raw": lat_raw, "lon_raw": lon_raw, "ns": ns, "ew": ew}
    _write_registry(registry)
    return {"success": True, "message": f"坐标已保存: {lat}°, {lon}°", "data": registry["geo_location"]}


@api_router.post("/ntp/sync", summary="NTP时间同步", tags=["NTP"])
async def ntp_sync(data: dict):
    """NTP时间同步 - SSE流式返回进度。"""
    import asyncio

    from src.advanced.ntp_sync import read_ntp_time, sync_device_time, sync_windows_time

    ntp_server = data.get("ntp_server", "ntp.aliyun.com")

    async def _generate():
        # 1. 读取NTP时间
        yield f"data: {_json.dumps({'step': 'reading', 'message': '正在读取NTP时间......'})}\n\n"
        ntp_r = await asyncio.to_thread(read_ntp_time, ntp_server, 5)
        if not ntp_r.get("success"):
            yield f"data: {_json.dumps({'step': 'error', 'message': 'NTP读取失败: ' + ntp_r.get('error', '未知错误')})}\n\n"
            return
        ntp_ts = ntp_r["ntp_timestamp"]
        ntp_time_cst = ntp_r["ntp_time_cst"]
        # NTP读取完成立刻测量本机误差
        from datetime import datetime, timedelta
        CST = timezone(timedelta(hours=8))
        now_pc = datetime.now(CST)
        ntp_dt = datetime.fromtimestamp(ntp_ts, tz=CST)
        pc_offset_ms = int((now_pc - ntp_dt).total_seconds() * 1000)
        yield f"data: {_json.dumps({'step': 'ntp_ok', 'message': f'当前NTP时间: {ntp_time_cst} | 校准误差 {pc_offset_ms:+d} ms'})}\n\n"

        # 2. 获取已连接设备信息
        mgr: PTZDeviceController | None = _managers.get("ptz_controller")
        if not mgr:
            yield f"data: {_json.dumps({'step': 'error', 'message': 'PTZDeviceController未初始化'})}\n\n"
            return
        device = mgr.get_connected_device()
        if not device:
            yield f"data: {_json.dumps({'step': 'error', 'message': '没有已连接的设备'})}\n\n"
            return
        ip = device.get("ip", "")
        if not ip:
            yield f"data: {_json.dumps({'step': 'error', 'message': '设备IP为空'})}\n\n"
            return
        yield f"data: {_json.dumps({'step': 'device_handshake', 'message': '完成PTZ设备握手......'})}\n\n"

        # 3. 写入设备时间(+3秒补偿,一次性同时写timeMode=manual + localTime)
        creds = mgr.get_credentials(ip)
        if not creds:
            yield f"data: {_json.dumps({'step': 'error', 'message': '无法获取设备凭据'})}\n\n"
            return
        dev_r = await asyncio.to_thread(
            sync_device_time,
            ip=ip,
            username=creds["username"],
            password=creds["password"],
            port=creds.get("port", 80),
            ntp_server=ntp_server,
            ntp_timestamp=ntp_ts,
        )
        dev_success = dev_r.get("success", False)
        dev_r.get("device_new_time", "unknown")
        yield f"data: {_json.dumps({'step': 'device_write', 'message': 'PTZ设备写入......' + ('成功' if dev_success else '失败: ' + dev_r.get('message', ''))})}\n\n"
        if not dev_success:
            return

        # 4. 写入本机时间
        yield f"data: {_json.dumps({'step': 'local_write', 'message': '本机写入......'})}\n\n"
        win_r = await asyncio.to_thread(sync_windows_time, ntp_ts)
        if not win_r.get("success"):
            yield f"data: {_json.dumps({'step': 'error', 'message': '本机写入失败: ' + win_r.get('error', '未知')})}\n\n"
            return

        yield f"data: {_json.dumps({'step': 'done', 'message': '✓ 同步成功'})}\n\n"

    return StreamingResponse(_generate(), media_type="text/event-stream")


class StorageSettingsRequest(BaseModel):
    record_path: str = ""       # 录像路径
    screenshot_path: str = ""   # v10.13: 截图路径（独立配置）


class PathSettingsRequest(BaseModel):
    category: str = ""  # "config" or "ext"
    local_config: str = ""
    device_config: str = ""
    test_data: str = ""
    sdk_path: str = ""
    astap_path: str = ""
    obs_path: str = ""
    obs_plugin_path: str = ""      # v10.02: OBS 插件目录
    gst_runtime_path: str = ""     # v10.02: GStreamer 运行时路径
    ffmpeg_path: str = ""          # v1.03: FFmpeg 路径


@api_router.get("/settings/storage", summary="获取存储设置", tags=["Settings"])
async def get_storage_settings() -> dict:
    """v7.105: 获取统一存储路径。v10.13: 截图/录像路径独立。"""
    record_path = _resolve_storage_path()
    screenshot_path = _resolve_screenshot_path()
    return {"success": True, "data": {"record_path": record_path, "screenshot_path": screenshot_path}}


@api_router.get("/files/list", summary="获取文件列表(按日期分组,内嵌元数据)", tags=["Files"])
async def list_files() -> dict:
    """v8.57: 列出存储目录下的文件,按日期分组返回。

    排除 .thumb.jpg 和 .meta.json 文件。
    视频文件内嵌元数据(duration/width/height),无需前端再请求 /files/info。
    """
    import json
    from datetime import datetime
    from pathlib import Path

    try:
        record_path = _resolve_storage_path()
        rdir = Path(record_path)

        if not rdir.exists():
            return {"success": True, "data": {"groups": []}}

        groups = {}
        media_exts = ["*.jpg", "*.jpeg", "*.mp4", "*.avi", "*.mov", "*.mkv"]
        skip_suffixes = ('.thumb.jpg', '.meta.json')

        for d in rdir.iterdir():
            if d.is_dir() and len(d.name) == 10 and d.name[4] == '-' and d.name[7] == '-':
                for pattern in media_exts:
                    for f in d.glob(pattern):
                        # 排除缩略图和元数据文件
                        if f.name.endswith(skip_suffixes):
                            continue
                        try:
                            mtime = f.stat().st_mtime
                            ext = f.suffix.lower()
                            file_type = "video" if ext in [".mp4", ".avi", ".mov", ".mkv"] else "image"
                            item = {
                                "file": d.name + "/" + f.name,
                                "date": datetime.fromtimestamp(mtime).strftime("%Y-%m-%d %H:%M:%S"),
                                "type": file_type
                            }
                            # 视频文件:内嵌元数据
                            if file_type == "video":
                                meta_file = f.with_suffix('.meta.json')
                                if meta_file.exists():
                                    try:
                                        meta = json.loads(meta_file.read_text(encoding='utf-8'))
                                        item['duration'] = meta.get('duration', '')
                                        item['width'] = meta.get('width', 0)
                                        item['height'] = meta.get('height', 0)
                                    except Exception:
                                        pass
                            groups.setdefault(d.name, []).append(item)
                        except Exception:
                            pass

        sorted_groups = []
        for date_str in sorted(groups.keys(), reverse=True):
            sorted_groups.append({"date": date_str, "items": sorted(groups[date_str], key=lambda x: x["date"], reverse=True)})

        return {"success": True, "data": {"groups": sorted_groups}}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/files/serve/{filename:path}", summary="提供文件访问", tags=["Files"])
async def serve_file(filename: str):
    """v7.105: 提供文件直接访问(流式传输,零内存占用)."""
    from pathlib import Path

    from fastapi.responses import FileResponse

    record_path = _resolve_storage_path()
    file_path = Path(record_path) / filename

    if not file_path.exists():
        return JSONResponse(status_code=404, content={"success": False, "message": "文件不存在"})

    return FileResponse(
        path=str(file_path),
        filename=file_path.name,
        media_type=_guess_mime(file_path.suffix.lower()),
    )


def _guess_mime(ext: str) -> str:
    """v7.105: 根据扩展名猜 MIME 类型."""
    mime_map = {
        '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg',
        '.png': 'image/png', '.gif': 'image/gif',
        '.mp4': 'video/mp4', '.avi': 'video/x-msvideo',
        '.mov': 'video/quicktime', '.mkv': 'video/x-matroska',
    }
    return mime_map.get(ext, 'application/octet-stream')


@api_router.post("/files/open-folder", summary="打开存储目录", tags=["Files"])
async def open_record_folder() -> dict:
    """v7.105: 打开存储目录。"""
    import platform
    import subprocess
    from pathlib import Path

    record_path = _resolve_storage_path()
    folder = Path(record_path)
    if not folder.exists():
        folder.mkdir(parents=True, exist_ok=True)

    try:
        if platform.system() == 'Windows':
            await asyncio.to_thread(subprocess.Popen,['explorer', str(folder.resolve())])
        elif platform.system() == 'Darwin':
            await asyncio.to_thread(subprocess.Popen, ['open', str(folder)])
        else:
            await asyncio.to_thread(subprocess.Popen, ['xdg-open', str(folder)])
        return {"success": True, "path": str(folder.resolve())}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/files/thumb/{filename:path}", summary="返回缩略图(仅返回已生成的)", tags=["Files"])
async def get_video_thumb(filename: str):
    """v8.57: 返回视频缩略图(仅返回已生成的.thumb.jpg),不再动态生成。"""
    from pathlib import Path

    from fastapi.responses import FileResponse

    record_path = _resolve_storage_path()
    file_path = Path(record_path) / filename
    if not file_path.exists():
        return JSONResponse(status_code=404, content={"success": False, "message": "文件不存在"})

    # 非视频文件直接返回原文件
    ext = file_path.suffix.lower()
    if ext not in ['.mp4', '.avi', '.mov', '.mkv']:
        return FileResponse(path=str(file_path), media_type=_guess_mime(ext))

    # 返回已生成的缩略图
    thumb_path = file_path.with_suffix('.thumb.jpg')
    if thumb_path.exists():
        return FileResponse(path=str(thumb_path), media_type='image/jpeg')

    # 缩略图未生成,返回原文件
    return FileResponse(path=str(file_path), media_type=_guess_mime(ext))


class RegenerateThumbRequest(BaseModel):
    """v1.1.16: 批量再生成缩略图请求。"""
    files: list[str] = []


@api_router.post("/files/regenerate-thumb", summary="批量再生成缺失的缩略图", tags=["Files"])
async def regenerate_thumbs(req: RegenerateThumbRequest) -> dict:
    """v1.1.16: 为缺少缩略图的视频文件重新生成 .thumb.jpg。"""
    import asyncio

    record_path = _resolve_storage_path()
    results = {"success": True, "generated": 0, "skipped": 0, "failed": 0}

    for filename in req.files:
        try:
            file_path = Path(record_path) / filename
            if not file_path.exists():
                results["skipped"] += 1
                continue
            ext = file_path.suffix.lower()
            if ext not in ['.mp4', '.avi', '.mov', '.mkv']:
                results["skipped"] += 1
                continue
            thumb_path = file_path.with_suffix('.thumb.jpg')
            if thumb_path.exists():
                results["skipped"] += 1
                continue
            # 异步调用生成函数
            await asyncio.to_thread(_generate_thumb_and_meta, file_path)
            if thumb_path.exists():
                results["generated"] += 1
            else:
                results["failed"] += 1
        except Exception:
            results["failed"] += 1

    return results


class DeleteFilesRequest(BaseModel):
    files: list[str] = []


@api_router.post("/files/delete", summary="批量删除文件", tags=["Files"])
async def delete_files(req: DeleteFilesRequest) -> dict:
    """v7.105: 批量删除文件."""
    from pathlib import Path

    record_path = _resolve_storage_path()
    deleted = []
    failed = []

    for f in req.files:
        file_path = Path(record_path) / f
        if file_path.exists():
            try:
                file_path.unlink()
                deleted.append(f)
            except Exception:
                failed.append(f)
        else:
            failed.append(f)

    return {"success": True, "deleted": deleted, "failed": failed}


@api_router.post("/files/upload-record", summary="上传录像文件,保存后自动生成缩略图和元数据", tags=["Files"])
async def upload_record(request: Request) -> dict:
    """v8.57: 上传录像文件,保存后自动生成缩略图(.thumb.jpg)和元数据(.meta.json)。"""
    from datetime import datetime
    from pathlib import Path

    from fastapi import UploadFile

    try:
        form = await request.form()
        file: UploadFile = form.get("file")
        if not file:
            return {"success": False, "message": "未收到文件"}

        record_path = _resolve_storage_path()
        date_str = datetime.now().strftime("%Y-%m-%d")
        date_dir = Path(record_path) / date_str
        date_dir.mkdir(parents=True, exist_ok=True)

        filepath = date_dir / file.filename
        file_data = await file.read()
        await asyncio.to_thread(_write_binary, filepath, file_data)

        # 录像文件保存后自动生成缩略图和元数据
        ext = filepath.suffix.lower()
        if ext in ['.mp4', '.avi', '.mov', '.mkv']:
            await asyncio.to_thread(_generate_thumb_and_meta, filepath)

        return {"success": True, "filepath": str(filepath), "filename": file.filename}
    except Exception as e:
        return {"success": False, "message": str(e)}


class MoveRecordingRequest(BaseModel):
    filename: str
    device_ip: str = ""


@api_router.post("/files/move-recording", summary="静默移动录像到存储路径", tags=["Files"])
async def move_recording(req: MoveRecordingRequest) -> dict:
    """v7.113: WASM 录制存到浏览器下载目录后,服务端静默移动到配置路径。"""
    import shutil
    from datetime import datetime
    from pathlib import Path

    try:
        # 目标路径(复用 _resolve_storage_path)
        record_path = _resolve_storage_path()
        date_str = datetime.now().strftime("%Y-%m-%d")
        date_str_alt = datetime.now().strftime("%Y%m%d")
        target_dir = Path(record_path) / date_str
        target_dir.mkdir(parents=True, exist_ok=True)

        target_file = target_dir / req.filename

        # 源目录:浏览器默认下载目录(bDateDir 创建的目录可能是 YYYY-MM-DD 或 YYYYMMDD)
        downloads = Path.home() / "Downloads"
        src_candidates = [
            downloads / date_str / req.filename,
            downloads / date_str_alt / req.filename,
            downloads / req.filename,
        ]

        src_file = None
        # 重试等待文件落盘(I_StopRecord 回调成功不代表文件已写完)
        for _ in range(6):
            for candidate in src_candidates:
                if candidate.exists():
                    src_file = candidate
                    break
            if src_file:
                break
            await asyncio.sleep(0.5)

        if src_file is None:
            return {"success": False, "message": f"源文件未找到: {req.filename}"}

        # 移动文件
        shutil.move(str(src_file), str(target_file))
        # 生成缩略图和元数据
        await asyncio.to_thread(_generate_thumb_and_meta, target_file)
        return {"success": True, "filepath": str(target_file), "filename": req.filename}
    except Exception as e:
        return {"success": False, "message": str(e)}


def _generate_thumb_and_meta(filepath: Path) -> None:
    """v8.57: 为视频生成缩略图(.thumb.jpg)和元数据(.meta.json)。

    v1.1.16: 添加 logging 记录生成失败原因（不再静默 pass）。
    """
    import json
    import subprocess

    from src.logger import get_logger
    _logger = get_logger(__name__)

    try:
        # v1.1.1: 统一探测 FFmpeg/FFprobe 绝对路径（含随包安装落点），不再仅依赖 PATH
        from src.updater.env_checker import find_ffmpeg_exe

        ffmpeg_exe = find_ffmpeg_exe()
        ffprobe_exe: str | None = None
        if ffmpeg_exe:
            sibling = Path(ffmpeg_exe).with_name("ffprobe.exe")
            ffprobe_exe = str(sibling) if sibling.exists() else None
        if not ffprobe_exe:
            import shutil
            ffprobe_exe = shutil.which("ffprobe")

        # 生成缩略图
        thumb_file = filepath.with_suffix('.thumb.jpg')
        if not thumb_file.exists():
            if not ffmpeg_exe:
                _logger.warning("缩略图生成跳过: FFmpeg 未找到, 文件=%s", filepath.name)
            else:
                try:
                    subprocess.run([
                        ffmpeg_exe, '-y', '-ss', '1', '-i', str(filepath),
                        '-vframes', '1', '-q:v', '5', '-s', '320x180', str(thumb_file)
                    ], capture_output=True, timeout=10,
                    creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, 'CREATE_NO_WINDOW') else 0)
                    if not thumb_file.exists():
                        _logger.warning("缩略图生成失败: FFmpeg 未报错但未生成文件, 文件=%s", filepath.name)
                except subprocess.TimeoutExpired:
                    _logger.warning("缩略图生成超时: 文件=%s", filepath.name)
                except Exception as e:
                    _logger.warning("缩略图生成异常: 文件=%s, 错误=%s", filepath.name, e)

        # 生成元数据
        meta_file = filepath.with_suffix('.meta.json')
        if not meta_file.exists() and ffprobe_exe:
            try:
                result = subprocess.run([
                    ffprobe_exe, '-v', 'quiet', '-print_format', 'json',
                    '-show_format', '-show_streams', str(filepath)
                ], capture_output=True, timeout=10,
                creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, 'CREATE_NO_WINDOW') else 0)
                if result.returncode == 0:
                    info = json.loads(result.stdout)
                    meta = {"type": "video"}
                    for s in info.get('streams', []):
                        if s.get('codec_type') == 'video':
                            meta['width'] = s.get('width', 0)
                            meta['height'] = s.get('height', 0)
                            break
                    dur = float(info.get('format', {}).get('duration', 0))
                    h, m = divmod(int(dur), 3600)
                    m, s = divmod(m, 60)
                    meta['duration'] = f"{h:02d}:{m:02d}:{s:02d}"
                    meta['duration_sec'] = int(dur)
                    meta_file.write_text(json.dumps(meta, ensure_ascii=False), encoding='utf-8')
            except Exception as e:
                _logger.warning("元数据生成失败: 文件=%s, 错误=%s", filepath.name, e)
    except Exception as e:
        _logger.warning("_generate_thumb_and_meta 外层异常: 文件=%s, 错误=%s", filepath.name, e)


_storage_config_cache: tuple[float, dict] | None = None


def _read_storage_config() -> dict:
    """v11.20: 读取 storage.json（内存缓存 + mtime 校验，避免每次截图/录像读盘）。"""
    global _storage_config_cache

    from src.config_paths import DATA_DIR

    config_path = DATA_DIR / "config" / "storage.json"
    try:
        mtime = config_path.stat().st_mtime
    except OSError:
        return {}
    if _storage_config_cache is not None and _storage_config_cache[0] == mtime:
        return _storage_config_cache[1]
    try:
        config = _json.loads(config_path.read_text(encoding="utf-8"))
    except Exception:
        config = {}
    _storage_config_cache = (mtime, config)
    return config


def _resolve_storage_path() -> str:
    """v7.105: 获取录像存储路径(优先从配置文件读取)。"""
    from src.config_paths import RECORD_DIR

    default_path = str(RECORD_DIR)
    config = _read_storage_config()
    return config.get("record_path") or default_path


def _resolve_screenshot_path() -> str:
    """v10.13: 获取截图存储路径(优先从配置文件读取,回退到录像路径)。"""
    from src.config_paths import RECORD_DIR

    default_path = str(RECORD_DIR / "screenshots")
    config = _read_storage_config()
    # 优先使用独立截图路径，回退到录像路径（兼容旧配置）
    return config.get("screenshot_path") or config.get("record_path") or default_path


@api_router.post("/settings/storage", summary="保存存储设置", tags=["Settings"])
async def save_storage_settings(req: StorageSettingsRequest) -> dict:
    """v7.105: 保存统一存储路径。v10.13: 截图/录像路径独立。"""
    import json

    from src.config_paths import DATA_DIR
    config_path = DATA_DIR / "config" / "storage.json"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config = {
        "record_path": req.record_path,
        "screenshot_path": req.screenshot_path,
    }
    await asyncio.to_thread(_write_file, config_path, json.dumps(config, indent=2, ensure_ascii=False))
    return {"success": True, "message": "存储设置已保存"}


@api_router.get("/settings/folder-picker", summary="打开文件夹选择对话框", tags=["Settings"])
async def folder_picker() -> dict:
    """打开系统原生文件夹选择对话框,返回用户选择的路径。"""
    import asyncio

    result = {"path": ""}

    def _pick():
        try:
            import tkinter as tk
            from tkinter import filedialog
            root = tk.Tk()
            root.withdraw()
            root.attributes("-topmost", True)
            path = filedialog.askdirectory(title="选择文件夹")
            root.destroy()
            result["path"] = path or ""
        except Exception as e:
            result["error"] = str(e)

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, lambda: _pick())

    if result.get("error"):
        return {"success": False, "message": result["error"]}
    if result["path"]:
        return {"success": True, "path": result["path"]}
    return {"success": False, "message": "未选择文件夹"}


@api_router.get("/settings/file-picker", summary="打开文件选择对话框", tags=["Settings"])
async def file_picker() -> dict:
    """打开系统原生文件选择对话框,返回用户选择的文件路径。"""
    import asyncio

    result = {"path": ""}

    def _pick():
        try:
            import tkinter as tk
            from tkinter import filedialog
            root = tk.Tk()
            root.withdraw()
            root.attributes("-topmost", True)
            path = filedialog.askopenfilename(title="选择文件")
            root.destroy()
            result["path"] = path or ""
        except Exception as e:
            result["error"] = str(e)

    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, lambda: _pick())

    if result.get("error"):
        return {"success": False, "message": result["error"]}
    if result["path"]:
        return {"success": True, "path": result["path"]}
    return {"success": False, "message": "未选择文件"}


@api_router.get("/settings/paths", summary="获取所有路径配置", tags=["Settings"])
async def get_all_paths() -> dict:
    """获取配置路径和扩展路径的当前设置。"""
    import json

    from src.config_paths import CONFIG_DIR, DATA_DIR, DEVICES_DIR, SDK_LIBS_DIR

    # 默认值来自 config_paths (实际路径)
    defaults = {
        "local_config": str(CONFIG_DIR),
        "device_config": str(DEVICES_DIR),
        "test_data": str(DEVICES_DIR),
        "sdk_path": str(SDK_LIBS_DIR),
        "astap_path": r"C:\Program Files\astap\astap.exe",
        "obs_path": r"C:\Program Files\obs-studio\bin\64bit\obs64.exe",
        "obs_plugin_path": r"C:\Program Files\obs-studio\obs-plugins\64bit",
        "gst_runtime_path": r"C:\gstreamer\1.0\msvc_x86_64\bin",
    }

    # 从存储文件读取已保存的配置 (覆盖默认值)
    config_path = DATA_DIR / "config" / "paths.json"
    if config_path.exists():
        try:
            content = await asyncio.to_thread(_read_file, config_path)
            saved = json.loads(content)
            defaults.update(saved)
        except Exception:
            pass

    return {"success": True, "data": defaults}


@api_router.post("/settings/paths", summary="保存路径配置", tags=["Settings"])
async def save_paths(req: PathSettingsRequest) -> dict:
    """保存配置路径和扩展路径。动态更新 astap_solve 等模块的路径引用。"""
    import json

    from src.config_paths import DATA_DIR

    config_path = DATA_DIR / "config" / "paths.json"
    config_path.parent.mkdir(parents=True, exist_ok=True)

    # 读取现有配置
    existing = {}
    if config_path.exists():
        try:
            content = await asyncio.to_thread(_read_file, config_path)
            existing = json.loads(content)
        except Exception:
            pass

    # 更新配置
    if req.category == "config":
        if req.local_config:
            existing["local_config"] = req.local_config
        if req.device_config:
            existing["device_config"] = req.device_config
        if req.test_data:
            existing["test_data"] = req.test_data
    elif req.category == "ext":
        if req.sdk_path:
            existing["sdk_path"] = req.sdk_path
        if req.astap_path:
            existing["astap_path"] = req.astap_path
        if req.obs_path:
            existing["obs_path"] = req.obs_path
        if req.obs_plugin_path:
            existing["obs_plugin_path"] = req.obs_plugin_path
        if req.gst_runtime_path:
            existing["gst_runtime_path"] = req.gst_runtime_path
        if req.ffmpeg_path:
            existing["ffmpeg_path"] = req.ffmpeg_path

    await asyncio.to_thread(_write_file, config_path, json.dumps(existing, indent=2, ensure_ascii=False))

    # v10.02: OBS 插件/GStreamer 路径同步到 local.json 的 obs 段（能力检测单一数据源）
    if req.category == "ext" and (req.obs_plugin_path or req.gst_runtime_path):
        try:
            local_path = DATA_DIR / "config" / "local.json"
            local_cfg = {}
            if local_path.exists():
                local_cfg = json.loads(await asyncio.to_thread(_read_file, local_path))
            obs = local_cfg.setdefault("obs", {})
            if req.obs_plugin_path:
                obs["plugin_path"] = req.obs_plugin_path
            if req.gst_runtime_path:
                obs["gst_runtime_path"] = req.gst_runtime_path
            await asyncio.to_thread(_write_file, local_path, json.dumps(local_cfg, indent=2, ensure_ascii=False))
        except Exception as e:
            return {"success": False, "message": f"同步 local.json 失败: {e}"}

    # 动态更新 ASTAP 路径
    if req.category == "ext" and req.astap_path:
        try:
            from src.api import astap_solve
            astap_solve.ASTAP_EXE = req.astap_path
        except Exception:
            pass

    return {"success": True, "message": "路径配置已保存"}


# === v1.1.27: 开发者模式 API ===


class DebugSettingsRequest(BaseModel):
    """开发者模式设置请求。"""
    debug_mode: bool = False
    log_level: str = "INFO"


@api_router.get("/settings/debug", summary="获取开发者模式状态", tags=["Settings"])
async def get_debug_settings() -> dict:
    """v1.1.27: 获取开发者模式状态和日志级别。"""
    from src.debug_config import get_debug_config
    return get_debug_config()


@api_router.post("/settings/debug", summary="设置开发者模式", tags=["Settings"])
async def set_debug_settings(req: DebugSettingsRequest) -> dict:
    """v1.1.27: 设置开发者模式和日志级别。"""
    from src.debug_config import set_debug_config
    return set_debug_config(req.debug_mode, req.log_level)


@api_router.post("/debug/client-log", summary="接收前端调试日志", tags=["Debug"])
async def receive_client_log(request: Request) -> dict:
    """v1.1.27: 接收前端发送的调试日志并写入 debug 目录。"""
    from datetime import datetime

    from src.logger import get_debug_dir, is_debug_enabled

    if not is_debug_enabled():
        return {"success": False, "message": "开发者模式未启用"}

    try:
        body = await request.json()
    except Exception:
        return {"success": False, "message": "无效的 JSON"}

    entries = body.get("entries", [])
    if not entries:
        return {"success": True, "message": "无日志条目"}

    # 写入前端日志文件
    debug_dir = get_debug_dir()
    debug_dir.mkdir(parents=True, exist_ok=True)
    today = datetime.now().strftime("%Y%m%d")
    fe_log = debug_dir / f"frontend_{today}.log"

    try:
        with open(fe_log, "a", encoding="utf-8") as f:
            for entry in entries:
                ts = entry.get("timestamp", "")
                method = entry.get("method", "")
                url = entry.get("url", "")
                status = entry.get("status", "")
                elapsed = entry.get("elapsed", "")
                req_body = entry.get("request_body", "")
                resp_body = entry.get("response_body", "")

                line = f"[{ts}] {method} {url} STATUS={status} TIME={elapsed}ms"
                if req_body:
                    line += f" | REQ={req_body[:512]}"
                if resp_body:
                    line += f" | RESP={resp_body[:512]}"
                f.write(line + "\n")
    except OSError as e:
        return {"success": False, "message": f"写入日志失败: {e}"}

    return {"success": True, "message": f"已记录 {len(entries)} 条日志"}


class ClientErrorRequest(BaseModel):
    """v1.1.31: 前端 JS 错误上报请求。"""
    msg: str = ""
    src: str = ""
    line: int = 0
    col: int = 0


@api_router.post("/client-error", summary="接收前端 JS 错误", tags=["Debug"])
async def receive_client_error(req: ClientErrorRequest) -> dict:
    """v1.1.31: 接收前端未捕获 JS 错误并记入后端日志（不受开发者模式门控）。"""
    from src.logger import get_logger

    get_logger(__name__).error(
        f"[前端JS错误] {req.msg} @ {req.src}:{req.line}:{req.col}"
    )
    return {"success": True}


@api_router.post("/obs/reconfigure", summary="重新配置 OBS 全链路", tags=["OBS"])
async def obs_reconfigure() -> dict:
    """v1.1.31: 重新配置 OBS 全链路（profile 编码器修补 + 重启 + 场景/推流目标重建）。

    供前端依赖弹窗"重新配置OBS"按钮调用，返回 health 快照。
    """
    from src.logger import get_logger
    from src.obs import obs_manager as _obs_manager_mod
    from src.obs.obs_manager import get_obs_manager

    log = get_logger(__name__)
    # manager 未创建时不隐式实例化，与现有端点风格一致返回 200 + success=False
    if getattr(_obs_manager_mod, "_obs_manager", None) is None:
        return {"success": False, "message": "ObsManager 未初始化"}
    try:
        mgr = get_obs_manager()
        health = await mgr.reconfigure()
        return {"success": True, "data": health}
    except Exception as e:
        log.error(f"OBS 重新配置失败: {e}")
        return {"success": False, "message": str(e)}


# ================================================================ #
#  流服务器扩展端点 (P0: 启动 / Snapshot)
# ================================================================ #






# ================================================================ #
#  高级功能端点 (Advanced: Function/Limit/Speed/Config/Onboarding)
# ================================================================ #

def _resolve_mac_from_ip(device_ip: str, client: ISAPIClient | None = None,
                          ptz_mgr: PTZDeviceController = None) -> str:

    """通过 PTZDeviceController 缓存或 ISAPI 获取设备的 MAC 地址。"""
    # Try PTZDeviceController cache first
    if ptz_mgr:
        discovered = ptz_mgr.get_discovered_devices()
        for dev in discovered:
            if dev.get("ip") == device_ip:
                mac = dev.get("mac", "")
                if mac:
                    # Normalize to colon-format uppercase
                    return mac.replace("-", ":").upper()
        stored = ptz_mgr.list_stored_devices()
        for dev in stored:
            if dev.get("ip") == device_ip:
                mac = dev.get("mac", "")
                if mac:
                    return mac.replace("-", ":").upper()
    # Fallback: use IP as identifier
    return device_ip.replace(".", "_").replace("/", "_").upper()

# Pydantic 请求模型
class AdvancedConfigWriteRequest(BaseModel):
    """配置写入请求。"""
    mac: str
    ip: str = ""
    model: str = ""
    capabilities: dict | None = None
    limits: dict | None = None
    speed: dict | None = None


class AdvancedOnboardingStartRequest(BaseModel):
    """引导开始请求。"""
    mac: str







# --- Individual Detect Endpoints (单项检测) ---

class _DetectRequest(BaseModel):
    """单项检测请求。"""
    device_ip: str = ""
    username: str = "admin"
    password: str = ""
    port: int = 80
    speed_profile: str = "lite"


async def _setup_client(device_ip: str, username: str, password: str, port: int):
    """通用设备连接: 返回 (mgr, ctrl, client, mac, err)。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return None, None, None, None, "PTZDeviceController not initialized"

    if not password:
        creds = mgr.get_credentials(device_ip)
        if creds:
            username = creds.get("username", username)
            password = creds.get("password", "")
            port = creds.get("port", port)

    ctrl, err = mgr._get_controller(device_ip)
    if err:
        return None, None, None, None, f"Controller error: {err}"

    client = ISAPIClient(ip=device_ip, username=username, password=password, port=port)
    if not client.verify_credentials():
        return None, None, None, None, "Authentication failed"

    mac = _resolve_mac_from_ip(device_ip, ptz_mgr=mgr)
    return mgr, ctrl, client, mac, None


@api_router.post("/advanced/detect/function", summary="功能检测", tags=["Advanced"])
async def detect_function(req: _DetectRequest) -> dict:
    """同步执行功能检测并返回结果。v11.12: 业务逻辑迁入 function.py。"""
    client = None
    try:
        mgr, ctrl, client, mac, err = await _setup_client(req.device_ip, req.username, req.password, req.port)
        if err:
            return {"success": False, "message": err}

        from src.advanced.function import FunctionDetector

        detector = FunctionDetector(client)
        result = await asyncio.to_thread(detector.run_full_detection)

        if not result.get("success"):
            return {"success": False, "message": result.get("error", "检测失败")}

        # 写入设备配置
        from src.advanced.config_writer import write_device_config
        write_device_config(mac=mac, capabilities=result["result"], ip=req.device_ip)

        return {"success": True, "result": result["result"], "supported_count": result["supported_count"], "total": result["total"]}
    except Exception as e:
        return {"success": False, "message": str(e)}
    finally:
        if client:
            client.close()


@api_router.get("/advanced/detect/function/status", summary="获取功能测试进度", tags=["Advanced"])
async def get_function_detect_status() -> dict:
    """v10.70: 获取功能测试实时进度。"""
    from src.advanced.function import get_function_test_progress
    return {"success": True, "progress": get_function_test_progress()}


@api_router.post("/advanced/detect/function/cancel", summary="取消功能测试", tags=["Advanced"])
async def cancel_function_detect() -> dict:
    """v10.70: 取消正在执行的功能测试。"""
    from src.advanced.function import cancel_active_function
    cancelled = cancel_active_function("")
    return {"success": True, "cancelled": cancelled}


@api_router.get("/advanced/detect/limit/status", summary="获取限位测试进度", tags=["Advanced"])
async def get_limit_detect_status() -> dict:
    """v10.70: 获取限位测试实时进度。"""
    from src.advanced.limit import get_limit_test_progress
    return {"success": True, "progress": get_limit_test_progress()}


@api_router.post("/advanced/detect/limit/cancel", summary="取消限位测试", tags=["Advanced"])
async def cancel_limit_detect() -> dict:
    """v10.70: 取消正在执行的限位测试。"""
    from src.advanced.limit import cancel_active_limit
    cancelled = cancel_active_limit("")
    return {"success": True, "cancelled": cancelled}


@api_router.get("/advanced/detect/speed/status", summary="获取速度测试进度", tags=["Advanced"])
async def get_speed_detect_status() -> dict:
    """v10.70: 获取速度测试实时进度。"""
    from src.advanced.speed import get_speed_test_progress
    return {"success": True, "progress": get_speed_test_progress()}


@api_router.post("/advanced/detect/speed/cancel", summary="取消速度测试", tags=["Advanced"])
async def cancel_speed_detect() -> dict:
    """v10.70: 取消正在执行的速度测试。"""
    from src.advanced.speed import cancel_active_speed
    cancelled = cancel_active_speed("")
    return {"success": True, "cancelled": cancelled}


@api_router.post("/advanced/detect/limit", summary="限位测试", tags=["Advanced"])
async def detect_limit(req: _DetectRequest) -> dict:
    """同步执行限位测试并返回结果。"""
    client = None
    try:
        mgr, ctrl, client, mac, err = await _setup_client(req.device_ip, req.username, req.password, req.port)
        if err:
            return {"success": False, "message": err}

        device_id = mac.upper().replace(":", "-") if mac else "onboarding"

        from src.advanced.config_writer import write_device_config
        from src.advanced.limit import LimitTester, set_active_limit_tester

        def _run():
            if not ISAPIPTZController(client).setup_home_preset():
                return None, "HOME预设10设置失败"
            limit_tester = LimitTester(client, device_id=device_id)
            set_active_limit_tester(limit_tester)  # v10.70: 注册活跃测试器
            result = limit_tester.run_all_tests()
            set_active_limit_tester(None)  # v10.70: 注销活跃测试器
            return result, None

        limit_result, test_err = await asyncio.to_thread(_run)
        if test_err:
            return {"success": False, "message": test_err}
        if not limit_result.get("success"):
            return {"success": False, "message": limit_result.get("message", limit_result.get("error", "未知错误"))}

        write_device_config(mac=mac, limits=limit_result, ip=req.device_ip)
        log_info("limit", "限位测试完成", {"mac": mac})
        return {"success": True, "result": limit_result}
    except Exception as e:
        log_error("limit", "限位测试异常", {"error": str(e)})
        return {"success": False, "message": str(e)}
    finally:
        if client:
            client.close()


@api_router.post("/advanced/detect/speed", summary="速度测试", tags=["Advanced"])
async def detect_speed(req: _DetectRequest) -> dict:
    """同步执行速度测试并返回结果。"""
    client = None
    try:
        mgr, ctrl, client, mac, err = await _setup_client(req.device_ip, req.username, req.password, req.port)
        if err:
            return {"success": False, "message": err}

        speed_profile: str = req.speed_profile or 'lite'

        from src.advanced.config_writer import write_device_config
        from src.advanced.speed import SpeedTester, set_active_speed_tester

        def _run():
            if not ISAPIPTZController(client).setup_home_preset():
                return None, "HOME预设10设置失败"
            speed_tester = SpeedTester(ISAPIPTZController(client))
            set_active_speed_tester(speed_tester)  # v10.70: 注册活跃测试器
            result = speed_tester.run_all_tests(None, speed_profile)
            set_active_speed_tester(None)  # v10.70: 注销活跃测试器
            return result, None

        speed_result, test_err = await asyncio.to_thread(_run)
        if test_err:
            return {"success": False, "message": test_err}
        if not speed_result:
            return {"success": False, "message": "速度测试无结果"}

        write_device_config(mac=mac, speed=speed_result, ip=req.device_ip)
        log_info("speed", "速度测试完成", {"mac": mac, "profile": speed_profile})
        return {"success": True, "result": speed_result}
    except Exception as e:
        log_error("speed", "速度测试异常", {"error": str(e)})
        return {"success": False, "message": str(e)}
    finally:
        if client:
            client.close()


def _scan_test_times(mac: str) -> dict:
    """同步扫描设备目录下 function/limit/speed.json 的修改时间（供 to_thread 调用）。

    v1.1.49: 从 async 端点抽出——目录 iterdir/stat 属同步文件 IO，直接在事件循环内执行
    会阻塞并发请求（AGENTS.md §5.1）。
    """
    import datetime

    from src.advanced.device_path import get_devices_dir as _get_devices_dir
    devices_dir = _get_devices_dir()
    result = {"function": None, "limit": None, "speed": None}

    if devices_dir.exists():
        target_dir = None
        if mac:
            # 按MAC精确匹配(目录名可能是MAC或IP,不区分大小写)
            mac_norm = mac.replace("-", ":").upper()
            for d in devices_dir.iterdir():
                if d.is_dir() and d.name.replace("-", ":").upper() == mac_norm:
                    target_dir = d
                    break
        if not target_dir:
            # 无mac或未匹配到,回退到最新修改的目录
            device_dirs = [d for d in devices_dir.iterdir() if d.is_dir()]
            if device_dirs:
                target_dir = max(device_dirs, key=lambda d: d.stat().st_mtime)

        if target_dir:
            for test_type in ["function", "limit", "speed"]:
                json_file = target_dir / f"{test_type}.json"
                if json_file.exists():
                    mtime = json_file.stat().st_mtime
                    dt = datetime.datetime.fromtimestamp(mtime)
                    result[test_type] = dt.strftime("%Y年%m月%d日 %H时%M分%S秒")

    return result


@api_router.get("/advanced/test-times", summary="获取测试最后完成时间", tags=["Advanced"])
async def get_test_times(mac: str = "") -> dict:
    """获取三个测试的最后完成时间(从JSON文件修改时间读取)。
    通过mac参数精确定位设备目录,无mac时取最新修改的目录。"""
    # v1.1.49: 文件扫描移入线程池，避免同步 IO 阻塞事件循环
    result = await asyncio.to_thread(_scan_test_times, mac)
    return {"success": True, "data": result}


# --- Config 配置 ---

@api_router.get("/advanced/config/{mac}", summary="获取设备配置", tags=["Advanced"])
async def advanced_get_config(mac: str) -> dict:
    """获取设备配置文件。"""
    try:
        from src.advanced.config_writer import list_device_configs, load_device_config

        if mac == "all":
            configs = list_device_configs()
            return {
                "success": True,
                "data": configs,
                "total": len(configs),
            }

        config = load_device_config(mac)
        if config:
            return {
                "success": True,
                "data": config.to_dict(),
            }
        return {
            "success": False,
            "message": f"未找到设备配置: {mac}",
        }
    except Exception as e:
        return {"success": False, "message": f"获取配置异常: {e}"}


@api_router.post("/advanced/config/write", summary="写入设备配置", tags=["Advanced"])
async def advanced_write_config(req: AdvancedConfigWriteRequest) -> dict:
    """写入设备配置文件。"""
    try:
        from src.advanced.config_writer import write_device_config

        result = write_device_config(
            mac=req.mac,
            capabilities=req.capabilities,
            limits=req.limits,
            speed=req.speed,
            ip=req.ip,
            model=req.model,
        )

        return result
    except Exception as e:
        return {"success": False, "message": f"写入配置异常: {e}"}


# --- Onboarding 引导 ---

@api_router.get("/advanced/onboarding/status", summary="获取引导状态", tags=["Advanced"])
async def advanced_onboarding_status(mac: str = Query(default="")) -> dict:
    """获取设备引导状态。"""
    if not mac:
        return {"success": True, "is_new_device": False, "progress": {"total_steps": 0, "completed": False}}
    try:
        from src.advanced.onboarding import OnboardingManager

        manager = OnboardingManager()
        is_new = manager.is_new_device(mac)
        progress = manager.get_progress(mac)

        return {
            "success": True,
            "is_new_device": is_new,
            "progress": progress,
        }
    except Exception as e:
        return {"success": False, "message": f"获取引导状态异常: {e}"}


@api_router.post("/advanced/onboarding/start", summary="开始引导流程", tags=["Advanced"])
async def advanced_onboarding_start(req: AdvancedOnboardingStartRequest) -> dict:
    """开始设备首次连接引导流程。"""
    try:
        from src.advanced.onboarding import OnboardingManager

        manager = OnboardingManager()
        result = manager.start_onboarding(req.mac)

        return result
    except Exception as e:
        return {"success": False, "message": f"开始引导异常: {e}"}


@api_router.post("/advanced/onboarding/complete", summary="完成引导流程", tags=["Advanced"])
async def advanced_onboarding_complete(mac: str) -> dict:
    """完成设备引导流程。"""
    try:
        from src.advanced.onboarding import OnboardingManager

        manager = OnboardingManager()
        result = manager.complete_onboarding(mac)

        return result
    except Exception as e:
        return {"success": False, "message": f"完成引导异常: {e}"}


@api_router.post("/advanced/onboarding/reset", summary="重置引导流程", tags=["Advanced"])
async def advanced_onboarding_reset(mac: str) -> dict:
    """重置设备引导流程 (允许重新执行)。"""
    try:
        from src.advanced.onboarding import OnboardingManager

        manager = OnboardingManager()
        result = manager.reset_onboarding(mac)

        return result
    except Exception as e:
        return {"success": False, "message": f"重置引导异常: {e}"}


@api_router.post("/advanced/onboarding/run", summary="完整执行设备引导流程", tags=["Advanced"])
async def advanced_onboarding_run(req: AdvancedOnboardingStartRequest) -> dict:
    """自动执行完整的设备首次连接引导流程。

    依次执行: 设备信息 → 功能探测 → HOME验证 → 限位测试 → 速度校准 → 恢复默认 → 保存配置 → 完成
    """
    from src.advanced.onboarding import OnboardingManager

    mgr = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}

    # Find device IP from PTZDeviceController
    device_ip = None
    for dev in mgr.list_stored_devices():
        if dev.get("mac", "").upper().replace("-", ":") == req.mac.upper().replace("-", ":"):
            device_ip = dev.get("ip")
            break
    if not device_ip:
        for dev in mgr.get_discovered_devices():
            if dev.get("mac", "").upper().replace("-", ":") == req.mac.upper().replace("-", ":"):
                device_ip = dev.get("ip")
                break
    if not device_ip:
        return {"success": False, "message": f"未找到设备 {req.mac} 的 IP 地址"}

    creds = mgr.get_credentials(device_ip)
    if not creds:
        return {"success": False, "message": "设备未保存凭据,请先连接设备"}

    manager = OnboardingManager()

    import asyncio
    result = await asyncio.to_thread(
        manager.execute_full_onboarding,
        mac=req.mac,
        ip=device_ip,
        username=str(creds.get("username", "admin")),
        password=str(creds.get("password", "")),
        port=int(creds.get("port", 80)),
    )
    return result


@api_router.get("/ws/stats", summary="WebSocket连接统计", tags=["WebSocket"])
async def get_ws_stats() -> dict:
    """获取当前 WebSocket 连接统计。"""
    try:
        from src.websocket.core.ws_manager import get_ws_manager
        ws_mgr = get_ws_manager()
        if ws_mgr:
            return {"success": True, "data": ws_mgr.get_stats()}
        return {"success": False, "error": "WebSocket not initialized"}
    except Exception as e:
        return {"success": False, "error": str(e)}


@api_router.get("/ws/connections", summary="WebSocket详细连接列表", tags=["WebSocket"])
async def get_ws_connections() -> dict:
    """获取当前 WebSocket 连接详细信息。"""
    try:
        from src.websocket.core.ws_manager import get_ws_manager
        ws_mgr = get_ws_manager()
        if ws_mgr:
            return {"success": True, "data": ws_mgr.list_all_connections() if hasattr(ws_mgr, "list_all_connections") else []}
        return {"success": False, "error": "WebSocket not initialized"}
    except Exception as e:
        return {"success": False, "error": str(e)}


# ASTAP 截图解析
from src.api.astap_solve import router as astap_router

api_router.include_router(astap_router)

@api_router.get("/ptz/devices/{ip}/credentials")
async def get_ptz_device_credentials(ip: str):
    """获取设备凭据。
    先查内存,再回退读 data/devices/{MAC}.json
    """
    ptz_mgr = _managers.get("ptz_controller")
    if ptz_mgr:
        creds = ptz_mgr._credentials.get(ip)
        if creds:
            return {
                "success": True,
                "ip": ip,
                "username": creds.get("username", "admin"),
                "password": creds.get("password", ""),
                "port": creds.get("port", 80),
            }
    # 回退:从 data/devices/{mac}/info.json 子目录读取
    try:
        import json as _json

        from src.advanced.device_path import get_devices_dir as _get_devices_dir
        devices_dir = _get_devices_dir()
        if devices_dir.exists():
            for device_dir in devices_dir.iterdir():
                if not device_dir.is_dir():
                    continue
                info_file = device_dir / "info.json"
                if not info_file.exists():
                    continue
                try:
                    content = await asyncio.to_thread(_read_file, info_file)
                    dev = _json.loads(content)
                    if dev.get("ip") == ip:
                        return {
                            "success": True,
                            "ip": ip,
                            "username": dev.get("username", "admin"),
                            "password": dev.get("password", ""),
                            "port": dev.get("port", 80),
                        }
                except Exception:
                    continue
    except Exception:
        pass
    return {"success": False, "message": "凭据不存在"}


# ================================================================ #
#  图像参数 API 端点 (白平衡/快门/光圈/降噪/锐度/OSD)
# ================================================================ #

@api_router.get("/ptz/{mac}/function", summary="获取设备功能探测结果", tags=["Image"])
async def get_device_function(mac: str) -> dict:
    """获取设备功能探测结果。

    v11.20: 优先读 DeviceReader 内存缓存，避免每请求目录扫描；
    缓存缺失时回退读文件（线程内执行，不阻塞事件循环）。
    回退路径保留时间戳文件优先逻辑（function_*.json）。
    """
    import json

    reader = get_device_reader()
    if reader:
        functions = reader.get_device_functions()
        if functions:
            return {"success": True, "data": {"functions": functions}}

    from src.advanced.device_path import get_data_path_read

    def _read_func_file() -> dict | None:
        func_file = get_data_path_read(None, mac, "function")
        if func_file and func_file.exists():
            return json.loads(func_file.read_text(encoding="utf-8"))
        return None

    try:
        func_data = await asyncio.to_thread(_read_func_file)
    except Exception as e:
        return {"success": False, "message": f"读取功能探测数据失败: {e}", "data": None}
    if func_data is not None:
        return {"success": True, "data": func_data}
    return {"success": False, "message": "未找到功能探测数据,请先完成功能测试", "data": None}


@api_router.get("/ptz/{mac}/test-status", summary="获取设备测试完成状态", tags=["Image"])
async def get_device_test_status(mac: str) -> dict:
    """一次性返回设备 5 项测试/校准的完成状态及文件修改时间。"""
    import os
    from datetime import datetime

    from src.advanced.device_path import (
        get_calibration_path,
        get_click_point_path,
        get_data_path_read,
    )

    def _fmt_mtime(path) -> str | None:
        """将文件 mtime 格式化为 YYYY-MM-DD HH:MM，失败返回 None。"""
        try:
            if path and os.path.isfile(path):
                return datetime.fromtimestamp(os.path.getmtime(path)).strftime("%Y-%m-%d %H:%M")
        except Exception:
            pass
        return None

    result = {
        "function": {"done": False, "time": None},
        "limit": {"done": False, "time": None},
        "speed": {"done": False, "time": None},
        "calibration": {"done": False, "time": None},
        "clickpoint": {"done": False, "time": None},
    }

    # function.json: 从 DeviceReader 缓存检查 + 文件 mtime
    reader = get_device_reader()
    func_path = None
    if reader:
        functions = reader.get_device_functions()
        if functions:
            result["function"]["done"] = True
            func_path = get_data_path_read(None, mac, "function")
    result["function"]["time"] = _fmt_mtime(func_path)

    # 其他 4 个文件: 检查路径是否存在 + mtime
    try:
        limit_path = get_data_path_read(None, mac, "limit")
        if limit_path and os.path.isfile(limit_path):
            result["limit"]["done"] = True
            result["limit"]["time"] = _fmt_mtime(limit_path)
    except Exception:
        pass

    try:
        speed_path = get_data_path_read(None, mac, "speed")
        if speed_path and os.path.isfile(speed_path):
            result["speed"]["done"] = True
            result["speed"]["time"] = _fmt_mtime(speed_path)
    except Exception:
        pass

    try:
        cal_path = get_calibration_path(mac)
        if cal_path and cal_path.exists():
            result["calibration"]["done"] = True
            result["calibration"]["time"] = _fmt_mtime(str(cal_path))
    except Exception:
        pass

    try:
        cp_path = get_click_point_path(mac)
        if cp_path and cp_path.exists():
            result["clickpoint"]["done"] = True
            result["clickpoint"]["time"] = _fmt_mtime(str(cp_path))
    except Exception:
        pass

    return {"success": True, "data": result}


async def _read_device_data_file(resolve) -> dict | None:
    """线程内读取设备数据 JSON，避免阻塞事件循环。

    Args:
        resolve: 无参函数，返回目标文件 Path（含目录扫描逻辑）。

    Returns:
        数据字典；文件不存在时返回 None。
    """
    import json

    def _do_read():
        f = resolve()
        if f and f.exists():
            return json.loads(f.read_text(encoding="utf-8"))
        return None

    return await asyncio.to_thread(_do_read)


@api_router.get("/advanced/data/{mac}/limit", summary="获取限位测试数据", tags=["Advanced"])
async def get_limit_data(mac: str) -> dict:
    """读取设备的限位测试数据 (limit.json)。"""
    from src.advanced.device_path import get_data_path_read

    try:
        data = await _read_device_data_file(lambda: get_data_path_read(None, mac, "limit"))
    except Exception as e:
        return {"success": False, "message": f"读取限位数据失败: {e}", "data": None}
    if data is not None:
        return {"success": True, "data": data}
    return {"success": False, "message": "未找到限位测试数据,请先完成限位测试", "data": None}


@api_router.get("/advanced/data/{mac}/speed", summary="获取速度测试数据", tags=["Advanced"])
async def get_speed_data(mac: str) -> dict:
    """读取设备的速度测试数据 (speed.json)。"""
    from src.advanced.device_path import get_data_path_read

    try:
        data = await _read_device_data_file(lambda: get_data_path_read(None, mac, "speed"))
    except Exception as e:
        return {"success": False, "message": f"读取速度数据失败: {e}", "data": None}
    if data is not None:
        return {"success": True, "data": data}
    return {"success": False, "message": "未找到速度测试数据,请先完成速度测试", "data": None}


@api_router.get("/advanced/data/{mac}/calibration", summary="获取三星校准数据", tags=["Advanced"])
async def get_calibration_data(mac: str) -> dict:
    """读取设备的三星校准数据 (calibration.json)。"""
    from src.advanced.device_path import get_calibration_path

    try:
        data = await _read_device_data_file(lambda: get_calibration_path(mac))
    except Exception as e:
        return {"success": False, "message": f"读取校准数据失败: {e}", "data": None}
    if data is not None:
        return {"success": True, "data": data}
    return {"success": False, "message": "未找到三星校准数据,请先完成三星校准", "data": None}


@api_router.get("/advanced/data/{mac}/clickpoint", summary="获取点击指向校准数据", tags=["Advanced"])
async def get_clickpoint_data(mac: str) -> dict:
    """读取设备的点击指向校准数据 (CLICK_POINT.JSON)。"""
    from src.advanced.device_path import get_click_point_path

    try:
        data = await _read_device_data_file(lambda: get_click_point_path(mac))
    except Exception as e:
        return {"success": False, "message": f"读取校准数据失败: {e}", "data": None}
    if data is not None:
        return {"success": True, "data": data}
    return {"success": False, "message": "未找到点击指向校准数据", "data": None}




@api_router.get("/ptz/{mac}/image/whitebalance", summary="获取白平衡设置", tags=["Image"])
async def get_whitebalance(mac: str) -> dict:
    """获取设备白平衡当前设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    wb_data = image_data.get("whitebalance", {})
    return {"success": True, "data": wb_data}


@api_router.post("/ptz/{mac}/image/whitebalance", summary="设置白平衡", tags=["Image"])
async def set_whitebalance(mac: str, data: dict) -> dict:
    """设置白平衡参数。v11.12: 委托给 ISAPIPTZController.set_whitebalance。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        mode_val = data.get("mode", "manual")
        red_val = data.get("red_gain", 100)
        blue_val = data.get("blue_gain", 80)
        result = await asyncio.to_thread(ctrl.set_whitebalance, mode_val, red_val, blue_val)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("whitebalance", {"mode": mode_val, "red_gain": red_val, "blue_gain": blue_val})
            log_info("image", "whitebalance", {"device": mac, "mode": mode_val, "red": red_val, "blue": blue_val})
            return {"success": True, "message": "白平衡已更新", "data": result["data"]}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置白平衡异常: {e}"}


@api_router.get("/ptz/{mac}/image/noisereduce", summary="获取降噪设置", tags=["Image"])
async def get_noisereduce(mac: str) -> dict:
    """获取设备降噪当前设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    nr_data = image_data.get("noisereduce", {})
    return {"success": True, "data": nr_data}


@api_router.post("/ptz/{mac}/image/noisereduce", summary="设置降噪", tags=["Image"])
async def set_noisereduce(mac: str, data: dict) -> dict:
    """设置降噪参数。v11.12: 委托给 ISAPIPTZController.set_noisereduce。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        spatial = data.get("spatial_level", 50)
        temporal = data.get("temporal_level", 50)
        result = await asyncio.to_thread(ctrl.set_noisereduce, spatial, temporal)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("noisereduce", {"spatial_level": spatial, "temporal_level": temporal})
            log_info("image", "noisereduce", {"device": mac, "spatial": spatial, "temporal": temporal})
            return {"success": True, "message": "降噪已更新", "data": result["data"]}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置降噪异常: {e}"}


@api_router.post("/ptz/{mac}/image/reset", summary="重置画面控制到默认值", tags=["Image"])
async def reset_image_controls(mac: str) -> dict:
    """重置画面控制到默认值。v11.12: 委托给 ISAPIPTZController.reset_image_defaults。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        result = await asyncio.to_thread(ctrl.reset_image_defaults)
        if result["success"]:
            log_info("image", "reset", {"device": mac})
        return {
            "success": result["success"],
            "message": "画面控制已重置到默认值" if result["success"] else "部分重置失败",
            "details": result["details"]
        }
    except Exception as e:
        return {"success": False, "message": f"重置异常: {e}"}


@api_router.get("/ptz/{mac}/image/exposure", summary="获取曝光模式", tags=["Image"])
async def get_exposure(mac: str) -> dict:
    """获取设备曝光当前模式(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    exp_data = image_data.get("exposure", {})
    return {"success": True, "data": exp_data}


@api_router.post("/ptz/{mac}/image/exposure", summary="设置曝光模式", tags=["Image"])
async def set_exposure(mac: str, data: dict) -> dict:
    """设置曝光模式。v11.12: 委托给 ISAPIPTZController.set_exposure。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        mode = data.get("mode", "manual")
        result = await asyncio.to_thread(ctrl.set_exposure, mode)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("exposure", {"mode": mode})
            log_info("image", "exposure", {"device": mac, "mode": mode})
            return {"success": True, "message": f"曝光模式已设置为 {mode}"}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置曝光模式异常: {e}"}


@api_router.get("/ptz/{mac}/image/shutter", summary="获取快门设置", tags=["Image"])
async def get_shutter(mac: str) -> dict:
    """获取设备快门当前设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    shutter_data = image_data.get("shutter", {})
    return {"success": True, "data": shutter_data}


@api_router.post("/ptz/{mac}/image/shutter", summary="设置快门", tags=["Image"])
async def set_shutter(mac: str, data: dict) -> dict:
    """设置快门参数。v11.12: 委托给 ISAPIPTZController.set_shutter。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        # 从 DeviceReader 内存缓存获取默认值（function.json 的缓存来源）
        default_level = "1/30000"
        cached_reader = get_device_reader()
        if cached_reader:
            functions = cached_reader.get_device_functions() or {}
            opt_values = (functions.get("shutter") or {}).get("opt_values", [])
            if opt_values:
                default_level = opt_values[0]
        level = data.get("level", default_level)
        result = await asyncio.to_thread(ctrl.set_shutter, level)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("shutter", {"current_level": level})
            log_info("image", "shutter", {"device": mac, "level": level})
            return {"success": True, "message": f"快门已设置为 {level}"}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置快门异常: {e}"}


@api_router.get("/ptz/{mac}/image/iris", summary="获取光圈设置", tags=["Image"])
async def get_iris(mac: str) -> dict:
    """获取设备光圈当前设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    iris_data = image_data.get("iris", {})
    return {"success": True, "data": iris_data}


@api_router.post("/ptz/{mac}/image/iris", summary="设置光圈", tags=["Image"])
async def set_iris(mac: str, data: dict) -> dict:
    """设置光圈参数。v11.12: 委托给 ISAPIPTZController.set_iris。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        level = data.get("level", 160)
        result = await asyncio.to_thread(ctrl.set_iris, level)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("iris", {"current_level": level})
            log_info("image", "iris", {"device": mac, "level": level})
            return {"success": True, "message": "光圈已设置"}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置光圈异常: {e}"}


# v7.03: 增益 (Gain) API
@api_router.get("/ptz/{mac}/image/gain", summary="获取增益设置", tags=["Image"])
async def get_gain(mac: str) -> dict:
    """获取设备增益当前设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    gain_data = image_data.get("gain", {})
    return {"success": True, "data": gain_data}


@api_router.post("/ptz/{mac}/image/gain", summary="设置增益", tags=["Image"])
async def set_gain(mac: str, data: dict) -> dict:
    """设置增益参数。v11.12: 委托给 ISAPIPTZController.set_gain。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        level = data.get("level")
        gain_limit = data.get("gain_limit")
        result = await asyncio.to_thread(ctrl.set_gain, level, gain_limit)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                gain_cache = {}
                if level is not None:
                    gain_cache["current_level"] = int(level)
                if gain_limit is not None:
                    gain_cache["gain_limit"] = int(gain_limit)
                reader.update_image_param("gain", gain_cache)
            log_info("image", "gain", {"device": mac, "level": level, "gain_limit": gain_limit})
            return {"success": True, "message": "增益已更新"}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置增益异常: {e}"}


@api_router.get("/ptz/{mac}/image/sharpness", summary="获取锐度设置", tags=["Image"])
async def get_sharpness(mac: str) -> dict:
    """获取设备锐度当前设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    sharpness_data = image_data.get("sharpness", {})
    return {"success": True, "data": sharpness_data}


@api_router.post("/ptz/{mac}/image/sharpness", summary="设置锐度", tags=["Image"])
async def set_sharpness(mac: str, data: dict) -> dict:
    """设置锐度参数。v11.12: 委托给 ISAPIPTZController.set_sharpness。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    try:
        level = data.get("level", 50)
        result = await asyncio.to_thread(ctrl.set_sharpness, level)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                reader.update_image_param("sharpness", {"current_level": level})
            log_info("image", "sharpness", {"device": mac, "level": level})
            return {"success": True, "message": f"锐度已设置为 {level}"}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置锐度异常: {e}"}


@api_router.get("/ptz/{mac}/image/color", summary="获取颜色设置", tags=["Image"])
async def get_color(mac: str) -> dict:
    """获取设备颜色(亮度/对比度/饱和度)当前设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    color_data = image_data.get("color", {})
    return {"success": True, "data": color_data}


# === 滤镜与日夜模式 (v8.43) ===

@api_router.get("/ptz/{mac}/image/filter", summary="获取滤镜与日夜模式", tags=["Image"])
async def get_filter_settings(mac: str) -> dict:
    """获取当前 dayNightMode 和 IrcutFilterType(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    filter_data = image_data.get("filter", {})
    return {"success": True, "data": filter_data}


@api_router.post("/ptz/{mac}/image/filter", summary="设置滤镜与日夜模式", tags=["Image"])
async def set_filter_settings(mac: str, body: dict) -> dict:
    """设置滤镜与日夜模式。v11.12: 委托给 ISAPIPTZController.set_filter。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    day_night = body.get("dayNightMode", "")
    ir_cut = body.get("IrcutFilterType", "")
    try:
        result = await asyncio.to_thread(ctrl.set_filter, day_night, ir_cut)
        if result.get("success"):
            log_info("image", "filter", {"device": mac, "dayNight": day_night, "irCut": ir_cut})
            reader = get_device_reader()
            if reader:
                reader.update_image_param("filter", {"dayNightMode": day_night or "", "IrcutFilterType": ir_cut or ""})
            return {"success": True, "message": "滤镜设置成功"}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置滤镜异常: {e}"}


# === 慢快门 DSS (v8.45) ===

@api_router.get("/ptz/{mac}/image/slow-shutter", summary="获取慢快门设置", tags=["Image"])
async def get_slow_shutter(mac: str) -> dict:
    """获取设备慢快门设置(从 DeviceReader 缓存读取)。"""
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings()
    if not image_data:
        return {"success": False, "message": "未获取到图像参数"}

    slow_shutter_data = image_data.get("slow_shutter", {})
    return {"success": True, "data": slow_shutter_data}

@api_router.post("/ptz/{mac}/image/slow-shutter", summary="设置慢快门", tags=["Image"])
async def set_slow_shutter(mac: str, body: dict) -> dict:
    """设置慢快门。v11.12: 委托给 ISAPIPTZController.set_slow_shutter。"""
    mgr = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZDeviceController未初始化"}
    target_ip = _resolve_device_id_to_ip(mgr, mac)
    if not target_ip:
        return {"success": False, "message": f"无法解析设备标识: {mac}"}
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return {"success": False, "message": err}
    dss_level = body.get("dss_level", "")
    try:
        result = await asyncio.to_thread(ctrl.set_slow_shutter, dss_level)
        if result.get("success"):
            reader = get_device_reader()
            if reader:
                enabled = dss_level != "off"
                reader.update_image_param("slow_shutter", {"enabled": enabled, "dss_level": dss_level})
            log_info("image", "dss", {"device": mac, "level": dss_level})
            return {"success": True, "message": "慢快门: " + ("关闭" if dss_level == "off" else dss_level)}
        return result
    except Exception as e:
        return {"success": False, "message": f"设置DSS异常: {e}"}

@api_router.get("/ptz/{mac}/osd/info", summary="获取Info OSD状态", tags=["OSD"])
async def get_osd_info(mac: str) -> dict:
    """获取Info OSD显示状态。v1.1.46: 实时查设备（单一事实源），
    成功回写 DeviceReader 缓存；ISAPI 不可达才回退缓存（旧版只读缓存，
    缓存脏/空时开关与画面叠加长期不一致）。"""
    from src.logger import get_logger as _get_logger

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if mgr:
        target_ip = _resolve_device_id_to_ip(mgr, mac)
        if target_ip:
            ctrl, err = mgr._get_controller(target_ip)
            if not err:
                try:
                    result = await asyncio.to_thread(ctrl.get_osd_info_status)
                    if result.get("success"):
                        reader = get_device_reader()
                        if reader:
                            reader.update_image_param(
                                "osd_info", {"enabled": result["enabled"]}
                            )
                        return {"success": True, "enabled": result["enabled"]}
                except Exception as e:
                    _get_logger(__name__).warning(
                        f"Info OSD 实时查询失败，回退缓存: {e}"
                    )

    # 回退：DeviceReader 缓存（连接时首次加载写入）
    reader = get_device_reader()
    if not reader:
        return {"success": False, "message": "DeviceReader 未初始化"}

    image_data = reader.get_image_settings() or {}
    info_osd = image_data.get("osd_info", {})
    if info_osd:
        return {"success": True, "enabled": info_osd.get("enabled", False)}
    return {"success": True, "enabled": False}


# ================================================================ #
#  系统管理端点
# ================================================================ #

@api_router.get("/system/token", summary="获取设备Token", tags=["System"])
async def get_device_token() -> dict:
    """v7.52: 后端获取设备Token(使用Digest认证)."""
    import requests
    from requests.auth import HTTPDigestAuth

    # v7.98: 从活跃设备配置读取,不再硬编码
    device_result = await get_active_device()
    if not device_result.get("success"):
        return {"success": False, "message": "无活跃设备,请先连接设备"}

    device = device_result["device"]
    camera_ip = device.get("ip", "")
    username = device.get("username", "admin")
    password = device.get("password", "")

    if not camera_ip:
        return {"success": False, "message": "活跃设备无IP地址"}

    try:
        resp = await asyncio.to_thread(requests.get,
            f'http://{camera_ip}:80/ISAPI/Security/token?format=json',
            auth=HTTPDigestAuth(username, password),
            timeout=10
        )
        if resp.status_code == 200:
            data = resp.json()
            token = data.get('Token', {}).get('value', '')
            return {"success": True, "token": token}
        else:
            return {"success": False, "message": f"Token获取失败: {resp.status_code}"}
    except Exception as e:
        return {"success": False, "message": f"Token获取异常: {e}"}

@api_router.post("/system/restart", summary="重启 SA.PTZhub", tags=["System"])
async def restart_astrohub() -> dict:
    """v1.1.0: 独立 ps1 重启（修复打包环境"重启变关闭"）。

    根因: frozen 下 sys.executable 即主 exe、restart_helper.py 未随包，
    旧逻辑等于启动第二个主程序实例，被 argparse/单实例互斥锁秒杀。
    新逻辑: 生成临时 ps1（等端口释放 → 拉起目标 → 自删除），
    CREATE_NO_WINDOW 启动后主进程延迟退出；dev/frozen 双分支。
    v1.1.1: PowerShell 绝对路径候选链（修复裸名启动 WinError 2）。
    """
    import shutil
    import subprocess
    import sys
    import tempfile

    from src.logger import get_logger
    logger = get_logger(__name__)

    frozen = getattr(sys, "frozen", False)
    if frozen:
        cwd = os.path.dirname(sys.executable)
        # 打包环境: 等待超时后兜底强杀旧进程再拉起主 exe
        kill_line = "Stop-Process -Name 'SA.PTZhub' -Force -ErrorAction SilentlyContinue"
        start_line = f"Start-Process -FilePath '{sys.executable}' -WorkingDirectory '{cwd}' -WindowStyle Hidden"
    else:
        cwd = os.getcwd()
        kill_line = "# dev 模式依赖主进程自退出"
        start_line = (
            f"Start-Process -FilePath '{sys.executable}' "
            f"-ArgumentList '-m src.main.main --headless' -WorkingDirectory '{cwd}' -WindowStyle Hidden"
        )

    # v1.1.0: 用 PowerShell 脚本承载重启（cmd bat 在 detached 无控制台环境
    # 下 timeout/start/管道 均不可靠，实测挂死）
    ps_dir = Path(tempfile.gettempdir()) / "astrohub_restart"
    ps_dir.mkdir(parents=True, exist_ok=True)
    ps_path = ps_dir / "restart.ps1"
    ps_content = f"""# SA.PTZhub 重启脚本 (v1.1.0)
$ErrorActionPreference = 'SilentlyContinue'
Set-Location -LiteralPath '{cwd}'
Start-Sleep -Seconds 2
$deadline = (Get-Date).AddSeconds(30)
while ((Get-Date) -lt $deadline) {{
    $conn = Get-NetTCPConnection -LocalPort 10280 -State Listen -ErrorAction SilentlyContinue
    if (-not $conn) {{ break }}
    Start-Sleep -Seconds 1
}}
{kill_line}
Start-Sleep -Seconds 1
{start_line}
Start-Sleep -Seconds 2
Remove-Item -LiteralPath $PSCommandPath -Force
"""
    ps_path.write_text(ps_content, encoding="utf-8")
    logger.info(f"[重启] 生成重启脚本: {ps_path} (frozen={frozen})")

    def _resolve_powershell() -> list[str]:
        """v1.1.1: PowerShell 绝对路径候选链（which → SystemRoot 内置 → pwsh）。"""
        candidates: list[str] = []
        found = shutil.which("powershell.exe")
        if found:
            candidates.append(found)
        sysroot = Path(os.environ.get("SYSTEMROOT", r"C:\Windows"))
        builtin = sysroot / "System32" / "WindowsPowerShell" / "v1.0" / "powershell.exe"
        if str(builtin) not in candidates and builtin.exists():
            candidates.append(str(builtin))
        pwsh = shutil.which("pwsh.exe")
        if pwsh:
            candidates.append(pwsh)
        return candidates

    # v1.1.0: 实测 DETACHED_PROCESS 下 powershell/cmd 子进程异常（不执行即退出/挂死），
    # 改用 CREATE_NO_WINDOW（隐藏控制台）保证脚本可靠执行
    creation_flags = subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP
    candidates = _resolve_powershell()
    launched = False
    last_err: Exception | None = None
    for exe in candidates:
        try:
            logger.info(f"[重启] 启动重启脚本: {exe} -File {ps_path}")
            await asyncio.to_thread(
                subprocess.Popen,
                [exe, "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", str(ps_path)],
                cwd=str(ps_dir),
                creationflags=creation_flags,
                close_fds=True,
            )
            logger.info("[重启] 重启脚本已启动")
            launched = True
            break
        except FileNotFoundError as e:
            last_err = e
            logger.warning(f"[重启] 候选不可用 {exe}: {e}")
    if not launched:
        msg = f"未找到可用 PowerShell (候选: {candidates or '无'}): {last_err}"
        logger.error(f"[重启] 启动重启脚本失败: {msg}")
        return {"success": False, "message": f"启动重启脚本失败: {msg}"}

    # 延迟退出,让 HTTP 响应先返回给客户端
    async def delayed_exit():
        await asyncio.sleep(0.5)  # 等待响应发送完成
        logger.info("[重启] 执行 sys.exit(0)")
        sys.exit(0)

    asyncio.create_task(delayed_exit())

    return {"success": True, "message": "重启中..."}


# ================================================================ #
#  v8.10: Vision 端点 - 框选分析 + 星点叠加
# ================================================================ #

import os

from pydantic import BaseModel as PydanticBase


class RegionAnalyzeRequest(PydanticBase):
    """框选区域分析请求。"""
    image_base64: str = ""  # 裁剪区域的 base64 图像
    image_path: str = ""    # 或:截图文件路径 + 裁剪坐标
    x: float = 0
    y: float = 0
    w: float = 0
    h: float = 0
    action: str = "both"  # whitebalance / focus / both


class StackStartRequest(PydanticBase):
    """叠加开始请求。"""
    total_exposure: float
    frame_exposure_ms: int = 100


# v11.12: 星点叠加管理已迁移到 src/vision/stack_engine.py
# v11.12: Vision 辅助函数已迁移到 src/controlpanel/region_base.py
from src.controlpanel.region_base import (
    capture_and_crop,
    cleanup_temp,
    get_search_state,
    guarded_search,
)
from src.vision.stack_engine import cancel_stacking, start_stacking, stop_stacking


def _get_client_for_device(mgr, device_id: str):
    """从 device_id (MAC或IP) 获取 ISAPIClient。"""
    target_ip = _resolve_device_id_to_ip(mgr, device_id)
    if not target_ip:
        return None, f"无法解析设备: {device_id}"
    ctrl, err = mgr._get_controller(target_ip)
    if err:
        return None, err
    client = mgr._clients.get(target_ip)
    if not client:
        return None, f"ISAPI client not found for {target_ip}"
    return client, ""


def _get_active_client(mgr):
    """获取当前活跃设备的 ISAPIClient。"""
    connected = mgr.get_connected_device()
    if not connected:
        return None, "无活跃设备"
    target_ip = connected.get("ip", "")
    if not target_ip:
        return None, "活跃设备无IP"
    return _get_client_for_device(mgr, target_ip)


@api_router.post("/vision/region-analyze", summary="框选区域分析(单次)", tags=["Vision"])
async def region_analyze(req: RegionAnalyzeRequest) -> dict:
    """单次框选分析:裁剪 + RGB统计 + 反差。用于预览。"""
    from src.controlpanel.region_base import _rgb_stats, calc_contrast

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZ控制器未初始化"}

    connected = mgr.get_connected_device()
    if not connected:
        return {"success": False, "message": "无活跃设备"}

    device_ip = connected["ip"]
    client, err = _get_client_for_device(mgr, device_ip)
    if err:
        return {"success": False, "message": err}

    bgr, crop_path, info = await asyncio.to_thread(
        capture_and_crop, client, mgr, device_ip, "WB",
        req.x, req.y, req.w, req.h
    )
    if bgr is None:
        return {"success": False, "message": info.get("error", "截图失败")}

    stats = _rgb_stats(bgr)
    contrast = calc_contrast(bgr)
    return {"success": True, "crop": info, "rgb": stats, "contrast": round(contrast, 2)}



# v8.72: 搜索器全局变量（v11.12: 互锁状态已迁移到 region_base.py）
_wb_searcher = None
_focus_searcher = None
_brightness_searcher = None


@api_router.get("/vision/search-status", summary="获取当前活跃搜索状态", tags=["Vision"])
async def search_status() -> dict:
    """获取当前活跃的搜索器类型,用于前端互锁。"""
    state = get_search_state()
    return {"success": True, "active": state['active']}


@api_router.post("/vision/check-baseline", summary="检查基线差异", tags=["Vision"])
async def check_baseline_api(data: dict) -> dict:
    """检查当前截图与基线的差异,判断是否需要执行搜索。"""
    from src.controlpanel.region_base import check_baseline, read_search_baseline

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZ控制器未初始化"}

    connected = mgr.get_connected_device()
    if not connected:
        return {"success": False, "message": "无活跃设备"}

    device_ip = connected["ip"]
    mac_clean = normalize_mac(connected.get("mac", ""))
    client, err = _get_client_for_device(mgr, device_ip)
    if err:
        return {"success": False, "message": err}

    mode = data.get("mode")  # 'focus' | 'whitebalance' | 'brightness'
    x = float(data.get("x", 0))
    y = float(data.get("y", 0))
    w = float(data.get("w", 0))
    h = float(data.get("h", 0))

    if not mode or w <= 0 or h <= 0:
        return {"success": False, "message": "参数无效"}

    # 截图
    bgr, _, info = await asyncio.to_thread(
        capture_and_crop, client, mgr, device_ip, mode.upper(), x, y, w, h,
        True  # skip_delay
    )
    if bgr is None:
        return {"success": False, "message": info.get("error", "截图失败")}

    # 读取基线(使用MAC作为设备标识)
    baseline = read_search_baseline(mac_clean, mode)
    if not baseline:
        return {"success": True, "should_search": True, "message": "无基线,执行搜索"}

    # 检查差异
    should_search, diff, message = check_baseline(bgr, mode, baseline)
    return {"success": True, "should_search": should_search, "diff_percent": round(diff, 1), "message": message}


@api_router.post("/vision/whitebalance-search", summary="白平衡迭代搜索 (SSE)", tags=["Vision"])
async def whitebalance_search(data: dict):
    """白平衡迭代搜索 - SSE流式。路由层只验证输入和调用白平衡模块。"""
    import json

    from src.advanced.scheduler import is_observation_blocking
    from src.controlpanel.whitebalance import WhiteBalanceSearcher

    if is_observation_blocking():
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '观测计划执行中，搜索功能已暂停'})}\n\n"]),
            media_type="text/event-stream"
        )

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': 'PTZ控制器未初始化'})}\n\n"]),
            media_type="text/event-stream"
        )

    connected = mgr.get_connected_device()
    if not connected:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '无活跃设备'})}\n\n"]),
            media_type="text/event-stream"
        )

    device_ip = connected["ip"]
    client, err = _get_client_for_device(mgr, device_ip)
    if err:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': err})}\n\n"]),
            media_type="text/event-stream"
        )

    x = float(data.get("x", 0))
    y = float(data.get("y", 0))
    w = float(data.get("w", 0))
    h = float(data.get("h", 0))

    if w <= 0 or h <= 0:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '框选区域无效'})}\n\n"]),
            media_type="text/event-stream"
        )

    mac_clean = normalize_mac(connected.get("mac", ""))

    # v11.15: 从内存缓存读取设备能力（不再反复访问硬盘）
    wb_rb_supported = True  # 默认支持
    from src.core.device_reader import get_device_reader
    reader = get_device_reader()
    if reader:
        func_data = reader.get_device_functions()
        if func_data:
            wb_func = func_data.get("white_balance", {})
            wb_rb_supported = wb_func.get("supported", True)

    searcher = WhiteBalanceSearcher(
        mgr=mgr, device_ip=device_ip, client=client,
        x=x, y=y, w=w, h=h,
        capture_func=capture_and_crop,
        cleanup_func=cleanup_temp,
        mac_clean=mac_clean,
        wb_rb_supported=wb_rb_supported  # v11.10: 传递R/B支持标志
    )

    # v8.41: 保存搜索器到全局变量
    global _wb_searcher
    _wb_searcher = searcher

    return StreamingResponse(
        guarded_search('wb', searcher.run),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )


@api_router.post("/vision/whitebalance-stop", summary="停止白平衡搜索", tags=["Vision"])
async def whitebalance_stop():
    """停止当前白平衡搜索并写入最佳值到设备"""
    global _wb_searcher
    if not _wb_searcher:
        return {"success": False, "message": "无活跃的白平衡搜索"}

    _wb_searcher._interrupt()
    _wb_searcher = None

    return {"success": True, "message": "白平衡搜索已停止"}


@api_router.post("/vision/focus-search", summary="反差对焦搜索", tags=["Vision"])
async def focus_search(data: dict):
    """反差对焦搜索 - SSE流式。路由层只验证输入和调用对焦模块。"""
    import json

    from src.advanced.scheduler import is_observation_blocking
    from src.controlpanel.autofocus import FocusSearcher

    if is_observation_blocking():
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '观测计划执行中，搜索功能已暂停'})}\n\n"]),
            media_type="text/event-stream"
        )

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': 'PTZ控制器未初始化'})}\n\n"]),
            media_type="text/event-stream"
        )

    connected = mgr.get_connected_device()
    if not connected:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '无活跃设备'})}\n\n"]),
            media_type="text/event-stream"
        )

    device_ip = connected["ip"]
    client, err = _get_client_for_device(mgr, device_ip)
    if err:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': err})}\n\n"]),
            media_type="text/event-stream"
        )

    x = float(data.get("x", 0))
    y = float(data.get("y", 0))
    w = float(data.get("w", 0))
    h = float(data.get("h", 0))

    if w <= 0 or h <= 0:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '框选区域无效'})}\n\n"]),
            media_type="text/event-stream"
        )

    searcher = FocusSearcher(
        mgr=mgr, device_ip=device_ip, client=client,
        x=x, y=y, w=w, h=h,
        capture_func=capture_and_crop,
        cleanup_func=cleanup_temp,
        mac_clean=normalize_mac(connected.get("mac", ""))
    )

    # v8.41: 保存搜索器到全局变量
    global _focus_searcher
    _focus_searcher = searcher

    return StreamingResponse(
        guarded_search('focus', searcher.run),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )


@api_router.post("/vision/focus-stop", summary="停止对焦搜索", tags=["Vision"])
async def focus_stop():
    """停止当前对焦搜索"""
    global _focus_searcher
    if not _focus_searcher:
        return {"success": False, "message": "无活跃的对焦搜索"}

    _focus_searcher._interrupt()
    _focus_searcher = None

    return {"success": True, "message": "对焦搜索已停止"}


@api_router.post("/vision/brightness-search", summary="局部亮度迭代搜索 (SSE)", tags=["Vision"])
async def brightness_search(data: dict):
    """局部亮度迭代搜索 - SSE流式。"""
    import json

    from src.advanced.scheduler import is_observation_blocking
    from src.controlpanel.brightness import BrightnessSearcher

    if is_observation_blocking():
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '观测计划执行中，搜索功能已暂停'})}\n\n"]),
            media_type="text/event-stream"
        )

    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': 'PTZ控制器未初始化'})}\n\n"]),
            media_type="text/event-stream"
        )

    connected = mgr.get_connected_device()
    if not connected:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '无活跃设备'})}\n\n"]),
            media_type="text/event-stream"
        )

    device_ip = connected["ip"]
    client, err = _get_client_for_device(mgr, device_ip)
    if err:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': err})}\n\n"]),
            media_type="text/event-stream"
        )

    x = float(data.get("x", 0))
    y = float(data.get("y", 0))
    w = float(data.get("w", 0))
    h = float(data.get("h", 0))

    if w <= 0 or h <= 0:
        return StreamingResponse(
            iter([f"data: {json.dumps({'type': 'error', 'message': '框选区域无效'})}\n\n"]),
            media_type="text/event-stream"
        )

    target = float(data.get("target", 50))
    shutter_idx = int(data.get("shutter_idx", 0))
    iris_idx = int(data.get("iris_idx", 0))
    gain = int(data.get("gain", 0))
    shutter_values = data.get("shutter_values", [])
    iris_values = data.get("iris_values", [])
    gain_min = int(data.get("gain_min", 0))
    gain_max = int(data.get("gain_max", 100))

    searcher = BrightnessSearcher(
        mgr=mgr, device_ip=device_ip, client=client,
        x=x, y=y, w=w, h=h,
        capture_func=capture_and_crop,
        cleanup_func=cleanup_temp,
        mac_clean=connected.get("mac", "").replace(":", "").replace("-", "").lower(),
        target=target, shutter_idx=shutter_idx, iris_idx=iris_idx,
        gain=gain, shutter_values=shutter_values, iris_values=iris_values,
        gain_min=gain_min, gain_max=gain_max
    )

    global _brightness_searcher
    _brightness_searcher = searcher

    return StreamingResponse(
        guarded_search('brightness', searcher.run),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )


@api_router.post("/vision/brightness-stop", summary="停止局部亮度搜索", tags=["Vision"])
async def brightness_stop():
    """停止当前局部亮度搜索"""
    global _brightness_searcher
    if not _brightness_searcher:
        return {"success": False, "message": "无活跃的亮度搜索"}

    _brightness_searcher._interrupt()
    _brightness_searcher = None

    return {"success": True, "message": "局部亮度搜索已停止"}


@api_router.post("/vision/stack-start", summary="启动星点叠加", tags=["Vision"])
async def stack_start(req: StackStartRequest) -> dict:
    """启动星点叠加会话。v11.12: 业务逻辑迁入 stack_engine.py。"""
    mgr: PTZDeviceController | None = _managers.get("ptz_controller")
    if not mgr:
        return {"success": False, "message": "PTZ控制器未初始化"}

    client, err = _get_active_client(mgr)
    if err:
        return {"success": False, "message": err}

    total_frames = int(req.total_exposure * 1000 / req.frame_exposure_ms)
    return await start_stacking(client, total_frames, req.frame_exposure_ms)


@api_router.post("/vision/stack-stop", summary="停止星点叠加", tags=["Vision"])
async def stack_stop() -> dict:
    """停止星点叠加。v11.12: 委托给 stack_engine.py。"""
    return await stop_stacking()


@api_router.post("/vision/stack-cancel", summary="取消星点叠加", tags=["Vision"])
async def stack_cancel() -> dict:
    """取消星点叠加。v11.12: 委托给 stack_engine.py。"""
    return await cancel_stacking()


# ================================================================ #
#  v8.80: 三星校准 API 端点 (v11.12: 业务逻辑迁入 rd2az.StarAlignManager)
# ================================================================ #


def _load_all_calibrations():
    """启动时加载所有已保存的校准数据。v11.12: 委托给 StarAlignManager。"""
    try:
        current_lat, current_lon = _get_station_coordinates()
    except ValueError as e:
        log_warn("staralign", "load_skip", str(e))
        return

    # 构建 mac_clean -> ip 映射
    mac_to_ip: dict[str, str] = {}
    mgr = _managers.get("ptz_controller")
    if mgr:
        for dev in mgr.list_stored_devices():
            dev_mac = normalize_mac(dev.get("mac", ""))
            dev_ip = dev.get("ip", "")
            if dev_mac and dev_ip:
                mac_to_ip[dev_mac] = dev_ip

    sam = get_staralign_manager()
    sam.load_all_calibrations(obs_lat=current_lat, obs_lon=current_lon, mac_to_ip=mac_to_ip)


def _get_star_catalog():
    """惰性加载星表。v11.12: 委托给 StarAlignManager。"""
    return get_staralign_manager().get_catalog()


# v8.80: 请求模型
class StarAlignStartRequest(BaseModel):
    device_ip: str

class StarAlignSyncRequest(BaseModel):
    device_ip: str
    star_hip: int
    ptz_pan: float = 0.0   # 已废弃,后端从设备读取
    ptz_tilt: float = 0.0  # 已废弃,后端从设备读取

class StarAlignComputeRequest(BaseModel):
    device_ip: str

class StarAlignSaveRequest(BaseModel):
    device_ip: str
    params: dict

class StarAlignGotoRequest(BaseModel):
    device_ip: str
    star_hip: int


@api_router.get("/staralign/catalog", summary="获取亮星表", tags=["StarAlign"])
async def staralign_catalog():
    """返回全部亮星列表,坐标由服务器台站配置提供。"""
    try:
        lat, lon = _get_station_coordinates()
        cat = _get_star_catalog()
        now_utc = datetime.now(UTC)
        # v11.30: astropy 逐星变换移出事件循环（上百颗星可造成秒级冻结）
        stars = await asyncio.to_thread(cat.get_visible, lat, lon, now_utc, min_alt=25, max_vmag=4.0)
        return {
            "success": True,
            "count": len(stars),
            "stars": [{
                "hip": s["hip"],
                "display_name": s["display_name"],
                "vmag": s["vmag"],
                "ra": s["ra_deg"],
                "dec": s["dec_deg"],
                "alt": round(s.get("alt", 0), 2),
                "az": round(s.get("az", 0), 2),
            } for s in stars]
        }
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/staralign/recommend", summary="获取推荐校准星", tags=["StarAlign"])
async def staralign_recommend(exclude_hips: str = Query("", description="逗号分隔的已选HIP编号")):
    """返回3颗推荐星,坐标由服务器台站配置提供。"""
    try:
        lat, lon = _get_station_coordinates()
        cat = _get_star_catalog()
        now_utc = datetime.now(UTC)
        excl = []
        if exclude_hips:
            excl = [int(h.strip()) for h in exclude_hips.split(",") if h.strip().isdigit()]
        # v11.30: astropy 计算移出事件循环
        stars = await asyncio.to_thread(cat.get_recommended, lat, lon, now_utc, exclude_hips=excl, count=3)
        return {
            "success": True,
            "recommended": [{
                "hip": s["hip"],
                "display_name": s["display_name"],
                "vmag": s["vmag"],
                "alt": round(s.get("alt", 0), 2),
                "az": round(s.get("az", 0), 2),
                "ra": s["ra_deg"],
                "dec": s["dec_deg"],
            } for s in stars]
        }
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/staralign/start", summary="初始化校准会话", tags=["StarAlign"])
async def staralign_start(req: StarAlignStartRequest):
    """初始化校准会话,返回北极星信息。v11.12: 委托给 StarAlignManager。"""
    try:
        obs_lat, obs_lon = _get_station_coordinates()
        sam = get_staralign_manager()
        # v11.30: astropy 变换移出事件循环
        return await asyncio.to_thread(sam.start_session, obs_lat=obs_lat, obs_lon=obs_lon, device_ip=req.device_ip)
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/staralign/apply", summary="应用校准点", tags=["StarAlign"])
async def staralign_apply(req: StarAlignSyncRequest):
    """记录一个校准点并更新模型。v11.30: 延迟落盘——只更新内存会话，
    写盘由 /staralign/finish 在用户点「完成」时执行。"""
    try:
        sam = get_staralign_manager()
        session = sam.get_session(req.device_ip)
        if not session:
            return {"success": False, "message": "校准会话不存在,请先 start"}

        # v11.30: 位置只从 DeviceReader 缓存读（零 ISAPI）
        reader = get_device_reader()
        pos = reader.get_position() if reader else None
        if not pos:
            return {"success": False, "message": "获取PTZ位置失败(缓存为空)"}

        # 调用 manager 的 apply_point（astropy solve 移出事件循环）
        result = await asyncio.to_thread(
            sam.apply_point,
            device_ip=req.device_ip,
            star_hip=req.star_hip,
            ptz_pan=pos["pan"],
            ptz_tilt=pos["tilt"],
        )

        return result
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/staralign/goto", summary="指向恒星", tags=["StarAlign"])
async def staralign_goto(req: StarAlignGotoRequest):
    """PTZ 指向目标恒星。v11.12: 使用 StarAlignManager 获取会话。"""
    try:
        sam = get_staralign_manager()
        session = sam.get_session(req.device_ip)
        if not session:
            return {"success": False, "message": "校准会话不存在,请先 start"}

        mgr = _managers.get("ptz_controller")
        if not mgr:
            return {"success": False, "message": "PTZ控制器未初始化"}

        target_ip = _resolve_device_id_to_ip(mgr, req.device_ip)
        if not target_ip:
            return {"success": False, "message": f"无法解析设备标识: {req.device_ip}"}

        ctrl, err = mgr._get_controller(target_ip)
        if err:
            return {"success": False, "message": err}

        # 读取 limit.json
        import json

        from src.advanced.device_path import get_data_path_read, get_device_info
        device_info = get_device_info(ctrl)
        limit_path = get_data_path_read(None, device_info["mac_clean"], "limit")
        limit_data = None
        if limit_path and limit_path.exists():
            try:
                content = await asyncio.to_thread(_read_file, limit_path)
                limit_data = json.loads(content)
            except Exception as e:
                log_warn("staralign", "limit_read_failed", str(e))

        solver = session.get("solver")
        # v11.30: 注入已加载的单例星表，消除每次 goto 重读整份 CSV
        goto = AstroGoto(ctrl, solver, limit_data=limit_data, catalog=sam.get_catalog())
        success, message = await asyncio.to_thread(
            goto.goto_star,
            star_hip=req.star_hip,
            obs_lat=session["obs_lat"],
            obs_lon=session["obs_lon"],
        )
        return {"success": success, "message": message}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/staralign/compute", summary="解算校准参数", tags=["StarAlign"])
async def staralign_compute(req: StarAlignComputeRequest):
    """解算5参数。v11.12: 委托给 StarAlignManager。"""
    try:
        sam = get_staralign_manager()
        return sam.compute(req.device_ip)
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/staralign/save", summary="保存校准参数", tags=["StarAlign"])
async def staralign_save(req: StarAlignSaveRequest):
    """保存校准参数到本地文件。v11.12: 委托给 StarAlignManager。"""
    try:
        sam = get_staralign_manager()
        session = sam.get_session(req.device_ip)
        if not session:
            return {"success": False, "message": "校准会话不存在"}

        # 获取设备信息
        mgr = _managers.get("ptz_controller")
        if not mgr:
            return {"success": False, "message": "PTZ控制器未初始化"}

        target_ip = _resolve_device_id_to_ip(mgr, req.device_ip)
        if not target_ip:
            return {"success": False, "message": f"无法解析设备标识: {req.device_ip}"}

        ctrl, err = mgr._get_controller(target_ip)
        if err:
            return {"success": False, "message": err}

        from src.advanced.device_path import get_device_info
        device_info = get_device_info(ctrl)
        mac_clean = device_info["mac_clean"]

        def _write_cal(path, content):
            with open(path, "w", encoding="utf-8") as f:
                f.write(content)

        result = sam.save_calibration(req.device_ip, mac_clean, _write_cal)
        if result.get("success"):
            session["calibration_result"] = req.params if req.params else result.get("save_data")
            log_info("staralign", "saved", {"device_ip": req.device_ip})
        return {"success": result.get("success", False), "message": result.get("message", ""), "path": result.get("path", "")}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/staralign/finish", summary="完成三星校准(落盘)", tags=["StarAlign"])
async def staralign_finish(req: StarAlignStartRequest):
    """v11.30: 完成校准 — 将内存会话落盘（延迟落盘语义的唯一写盘点）。"""
    try:
        sam = get_staralign_manager()
        session = sam.get_session(req.device_ip)
        if not session:
            return {"success": False, "message": "校准会话不存在"}

        mgr = _managers.get("ptz_controller")
        if not mgr:
            return {"success": False, "message": "PTZ控制器未初始化"}
        dev = mgr.get_connected_device() or {}
        mac_clean = normalize_mac(dev.get("mac") or "")
        if not mac_clean:
            return {"success": False, "message": "无法获取设备 MAC"}

        def _write_cal(path, content):
            with open(path, "w", encoding="utf-8") as f:
                f.write(content)

        result = await asyncio.to_thread(sam.save_calibration, req.device_ip, mac_clean, _write_cal)
        if result.get("success"):
            log_info("staralign", "finish", {"device_ip": req.device_ip, "path": result.get("path")})
        return {"success": result.get("success", False), "message": result.get("message", ""), "path": result.get("path", "")}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/staralign/abort", summary="取消三星校准(零写盘)", tags=["StarAlign"])
async def staralign_abort(req: StarAlignStartRequest):
    """v11.30: 取消校准 — 丢弃内存会话并回滚到磁盘旧校准，零写盘。"""
    try:
        sam = get_staralign_manager()
        mgr = _managers.get("ptz_controller")
        dev = (mgr.get_connected_device() or {}) if mgr else {}
        mac_clean = normalize_mac(dev.get("mac") or "")
        try:
            obs_lat, obs_lon = _get_station_coordinates()
        except ValueError:
            obs_lat, obs_lon = 0.0, 0.0
        result = await asyncio.to_thread(
            sam.abort_session, req.device_ip, obs_lat, obs_lon, mac_clean
        )
        log_info("staralign", "abort", {"device_ip": req.device_ip})
        return result
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/staralign/status", summary="查询校准状态", tags=["StarAlign"])
async def staralign_status(device_ip: str = Query(..., description="设备IP")):
    """返回当前校准会话状态。v11.12: 使用 StarAlignManager。"""
    try:
        sam = get_staralign_manager()
        session = sam.get_session(device_ip)
        if not session:
            return {"success": False, "message": "无校准会话"}
        solver: CalibrationSolver = session["solver"]
        status = solver.status()
        return {
            "success": True,
            **status,
            "points": session["points"],
        }
    except Exception as e:
        return {"success": False, "message": str(e)}

# ═══════════════════════════════════════════════════════════════════════════════
# 天体跟踪 API (v8.100) — v11.12: 业务逻辑迁入 astro_tracking.py
# ═══════════════════════════════════════════════════════════════════════════════


class TrackingStartRequest(BaseModel):
    """启动跟踪请求。观测站坐标由后端统一读取。"""
    device_ip: str
    mac: str
    rate_mode: str = "sidereal"  # sidereal | lunar | solar
    speed_tier: int = 1  # 已废弃 (v10.65): 速度档位已取消，保留字段兼容旧前端


class TrackingRateRequest(BaseModel):
    """切换跟踪速率模式请求"""
    rate: str  # sidereal | lunar | solar


class TrackingSpeedTierRequest(BaseModel):
    """切换跟踪速度档位请求"""
    tier: int  # 1-5


class TrackingIdentifyRequest(BaseModel):
    """目标识别请求"""
    name: str


@api_router.post("/tracking/start", summary="启动天体跟踪", tags=["Tracking"])
async def tracking_start(req: TrackingStartRequest) -> dict:
    """启动跟踪引擎。v11.12: 委托给 astro_tracking.py。"""
    try:
        reader = get_device_reader()
        if not reader:
            return {"success": False, "message": "DeviceReader未初始化"}

        try:
            obs_lat, obs_lon = _get_station_coordinates()
        except ValueError as e:
            return {"success": False, "message": str(e)}

        result = start_tracking(
            device_reader=reader,
            mac=req.mac,
            obs_lat=obs_lat,
            obs_lon=obs_lon,
            rate_mode=req.rate_mode,
            speed_tier=req.speed_tier,
        )

        if result.get("success"):
            log_info("tracking", "start", {"rate_mode": req.rate_mode})

        return result
    except RuntimeError as e:
        if "速度缓存未加载" in str(e) or "速度数据缺失" in str(e):
            return {
                "success": False,
                "message": "请先完成测速,测速数据缺失",
                "error_code": "SPEED_CACHE_MISSING",
            }
        return {"success": False, "message": f"启动跟踪失败: {e}"}
    except Exception as e:
        return {"success": False, "message": f"启动跟踪失败: {e}"}


@api_router.post("/tracking/stop", summary="停止天体跟踪", tags=["Tracking"])
async def tracking_stop() -> dict:
    """停止跟踪引擎。v11.12: 委托给 astro_tracking.py。"""
    try:
        result = stop_tracking()
        if result.get("success"):
            log_info("tracking", "stop", {})
        return result
    except Exception as e:
        return {"success": False, "message": f"停止跟踪失败: {e}"}


@api_router.post("/tracking/rate", summary="切换跟踪速率", tags=["Tracking"])
async def tracking_rate(req: TrackingRateRequest) -> dict:
    """切换跟踪速率模式。v11.12: 委托给 astro_tracking.py。"""
    try:
        result = set_tracking_rate(req.rate)
        if result.get("success"):
            log_info("tracking", "rate", {"rate": req.rate})
        return result
    except Exception as e:
        return {"success": False, "message": f"设置速率失败: {e}"}


@api_router.post("/tracking/speed-tier", summary="设置跟踪速度档位", tags=["Tracking"])
async def tracking_speed_tier(req: TrackingSpeedTierRequest) -> dict:
    """已废弃 (v10.65): 速度档位已取消，始终使用最慢速度。保留端点兼容旧前端。"""
    return {
        "success": True,
        "message": "速度档位已废弃 (v10.65)，始终使用最慢速度",
        "deprecated": True,
    }


@api_router.get("/tracking/status", summary="获取跟踪状态", tags=["Tracking"])
async def tracking_status() -> dict:
    """获取当前跟踪引擎状态。v11.12: 委托给 astro_tracking.py。"""
    try:
        return get_tracking_status()
    except Exception as e:
        return {"success": False, "message": f"获取状态失败: {e}"}


@api_router.post("/tracking/identify-target", summary="识别目标", tags=["Tracking"])
async def tracking_identify(req: TrackingIdentifyRequest) -> dict:
    """根据用户输入的名称识别目标类型和返回信息"""
    try:
        resolver = CelestialResolver()
        result = resolver.identify_target(req.name)
        return {"success": True, "data": result}

    except Exception as e:
        return {"success": False, "message": f"目标识别失败: {e}"}




# ================================================================ #
#  v8.102: SkyAtlas 天体目标选择 API
# ================================================================ #


@api_router.get("/skyatlas/solar", summary="获取太阳系天体实时坐标", tags=["SkyAtlas"])
async def skyatlas_solar() -> dict:
    """返回太阳、月球、五大行星的实时 RA/Dec 坐标"""
    try:
        sa = get_skyatlas()
        bodies = sa.get_solar_system_bodies()
        return {"success": True, "data": bodies}
    except Exception as e:
        return {"success": False, "message": f"获取失败: {e}"}


@api_router.get("/skyatlas/messier", summary="获取梅西耶天体列表", tags=["SkyAtlas"])
async def skyatlas_messier() -> dict:
    """返回110个梅西耶天体列表(用于下拉菜单)"""
    try:
        sa = get_skyatlas()
        targets = sa.get_messier_targets()
        return {"success": True, "data": targets}
    except Exception as e:
        return {"success": False, "message": f"获取失败: {e}"}


@api_router.get("/skyatlas/search", summary="模糊搜索天体", tags=["SkyAtlas"])
async def skyatlas_search(q: str = Query(..., description="搜索关键词,如 m42、猎户、星云")) -> dict:
    """模糊搜索天体,返回匹配列表"""
    try:
        sa = get_skyatlas()
        results = sa.search_target(q)
        return {"success": True, "query": q, "count": len(results), "data": results}
    except Exception as e:
        return {"success": False, "message": f"搜索失败: {e}"}


@api_router.get("/skyatlas/stellarium", summary="获取Stellarium当前选中目标", tags=["SkyAtlas"])
async def skyatlas_stellarium() -> dict:
    """从 Stellarium 远程获取当前选中目标(容错处理)"""
    try:
        sa = get_skyatlas()
        target = sa.get_stellarium_target()
        if target:
            return {"success": True, "data": target}
        return {"success": False, "message": "无法连接 Stellarium 或未选中目标"}
    except Exception as e:
        return {"success": False, "message": f"获取失败: {e}"}


# ── v8.102: SkyAtlas Goto + 坐标转换 ──

class SkyAtlasGotoRequest(BaseModel):
    """SkyAtlas 指向请求"""
    ra: float
    dec: float
    target_name: str = ""
    track_mode: str = "sidereal"


class SkyAtlasCoordConvertRequest(BaseModel):
    """SkyAtlas 坐标转换请求"""
    ra: float | None = None
    dec: float | None = None
    az: float | None = None
    alt: float | None = None


@api_router.post("/skyatlas/goto", summary="SkyAtlas 指向天体", tags=["SkyAtlas"])
async def skyatlas_goto(req: SkyAtlasGotoRequest) -> dict:
    """将 PTZ 指向指定 RA/Dec 坐标的天体。v11.12: 业务逻辑迁入 skyatlas.py。"""
    try:
        mgr = _managers.get("ptz_controller")
        if not mgr:
            return {"success": False, "message": "PTZDeviceController未初始化"}

        connected = mgr.get_connected_device()
        if not connected:
            return {"success": False, "message": "无已连接设备"}

        target_ip = connected.get("ip", "")
        if not target_ip:
            return {"success": False, "message": "活跃设备无IP"}

        ctrl, err = mgr._get_controller(target_ip)
        if err:
            return {"success": False, "message": f"获取控制器失败: {err}"}

        try:
            obs_lat, obs_lon = _get_station_coordinates()
        except ValueError as e:
            return {"success": False, "message": str(e)}

        # 读取校准参数和限位数据
        session = get_staralign_manager().sessions.get(target_ip)
        solver = None
        if session and session.get("solver") and session["solver"]._solved:
            solver = session["solver"]

        import json

        from src.advanced.device_path import get_data_path_read, get_device_info
        device_info = get_device_info(ctrl)
        limit_path = get_data_path_read(None, device_info["mac_clean"], "limit")
        limit_data = None
        if limit_path and limit_path.exists():
            try:
                content = await asyncio.to_thread(_read_file, limit_path)
                limit_data = json.loads(content)
            except Exception as e:
                log_warn("skyatlas", "limit_read_failed", str(e))

        result = await asyncio.to_thread(
            skyatlas_goto_service,
            ctrl, req.ra, req.dec, req.target_name,
            obs_lat, obs_lon, limit_data, solver,
        )

        if result.get("success"):
            log_info("skyatlas", "goto", {
                "target": req.target_name, "ra": req.ra, "dec": req.dec,
                "alt": result.get("alt"), "az": result.get("az"),
            })

        return result
    except Exception as e:
        return {"success": False, "message": f"指向失败: {e}"}


@api_router.post("/skyatlas/coord-convert", summary="SkyAtlas 坐标转换", tags=["SkyAtlas"])
async def skyatlas_coord_convert(req: SkyAtlasCoordConvertRequest) -> dict:
    """RA/Dec ↔ Alt/Az 双向坐标转换。v11.12: 业务逻辑迁入 skyatlas.py。"""
    try:
        try:
            obs_lat, obs_lon = _get_station_coordinates()
        except ValueError as e:
            return {"success": False, "message": str(e)}

        return coord_convert_service(
            ra=req.ra, dec=req.dec, az=req.az, alt=req.alt,
            obs_lat=obs_lat, obs_lon=obs_lon,
        )
    except Exception as e:
        return {"success": False, "message": f"坐标转换失败: {e}"}


@api_router.get("/altitude-chart", summary="获取PTZ指向高度曲线", tags=["Astronomy"])
async def get_altitude_chart(device_id: str | None = Query(None)) -> dict:
    """根据当前PTZ实际指向生成本地日期24小时高度曲线。

    PTZ位置 → 校准后的Alt/Az(无校准时使用设备角度)→ RA/Dec → 曲线。
    该端点只在前端检测到设备指向变化时调用;当前时间线由前端本地刷新。
    """
    try:
        mgr = _managers.get("ptz_controller")
        if not mgr:
            return {"success": False, "message": "PTZDeviceController未初始化"}

        if not device_id:
            connected = mgr.get_connected_device()
            device_id = connected.get("ip", "") if connected else ""
        if not device_id:
            return {"success": False, "message": "无已连接设备"}

        reader = get_device_reader()
        pos = reader.get_position() if reader else None
        if not pos:
            return {"success": False, "message": "未获取到PTZ位置数据"}

        target_ip = _resolve_device_id_to_ip(mgr, device_id)
        if not target_ip:
            return {"success": False, "message": f"无法解析设备标识: {device_id}"}

        pan = float(pos.get("pan", 0))
        tilt = float(pos.get("tilt", 0))

        # v1.1.50: 检查三星校准状态 — 未校准时不生成指向模型
        session = get_staralign_manager().sessions.get(target_ip)
        calibrated = bool(session and session.get("solver") and session["solver"]._solved)

        if not calibrated:
            return {
                "success": False,
                "message": "未执行三星校准，无法生成指向模型",
                "data": {"calibrated": False}
            }

        # 已校准：使用校准模型转换 PTZ → 真实 Alt/Az
        az, alt = session["solver"].ptz_to_true(pan / 10.0, tilt / 10.0)

        try:
            obs_lat, obs_lon = _get_station_coordinates()
        except ValueError as e:
            return {"success": False, "message": str(e)}

        # v9.63: 读取limit.json获取tilt_min作为纵轴起点
        alt_min = -20  # 默认值
        try:
            import json

            from src.advanced.device_path import get_data_path_read, get_device_info
            device_info = get_device_info(mgr)
            limit_path = get_data_path_read(None, device_info["mac_clean"], "limit")
            if limit_path and limit_path.exists():
                content = await asyncio.to_thread(_read_file, limit_path)
                limit_data = json.loads(content)
                tilt_min = limit_data.get("tilt_min", -200)
                alt_min = tilt_min / 10.0  # 0.1°→度
        except Exception:
            pass

        now_utc = datetime.now(UTC)
        ra, dec = CoordinateConverter().altaz_to_radec(
            float(alt), float(az), obs_lat, obs_lon, now_utc
        )
        data = get_altitude_chart_data(ra, dec, obs_lat, obs_lon, alt_min=alt_min)
        data.update({
            "device_id": device_id,
            "az": round(float(az), 4),
            "alt": round(float(alt), 4),
            "calibrated": calibrated,
            "obs_lat": round(float(obs_lat), 4),
        })
        return {"success": True, "data": data}
    except Exception as e:
        return {"success": False, "message": f"高度曲线计算失败: {e}"}


@api_router.get("/advanced/file-status", summary="获取配置文件状态", tags=["Advanced"])
async def get_file_status(mac: str = None):
    """获取所有配置文件的存在状态"""
    # v1.0.2: frozen 下 __file__ 指向 _internal/，统一走 config_paths
    from src.config_paths import DATA_DIR, get_app_dir

    app_dir = get_app_dir()
    data_dir = DATA_DIR

    result = {
        "station": [],  # 台站配置
        "device": [],   # 设备配置
        "logs": []      # 运行记录
    }

    # 台站配置文件
    station_files = [
        ("主控台功能配置", app_dir / "console_features.json"),
        ("本地配置", data_dir / "config" / "local.json"),
        ("存储配置", data_dir / "config" / "storage.json"),
        ("设备注册表", data_dir / "registry.json"),
        ("主控台快照", data_dir / "console_snapshot.json"),
        ("设备型号库", data_dir / "device_models.json"),
        ("PTZ型号库", data_dir / "ptz_models.json"),
    ]

    for name, path in station_files:
        result["station"].append({
            "name": name,
            "path": str(path),
            "exists": path.exists()
        })

    # 设备配置文件(需要 mac 参数)
    if mac:
        device_dir = data_dir / "devices" / mac
        device_files = [
            ("设备信息", device_dir / "info.json"),
            ("功能配置", device_dir / "function.json"),
            ("限位配置", device_dir / "limit.json"),
            ("速度配置", device_dir / "speed.json"),
            ("校准数据", device_dir / "calibration.json"),
            ("点击定位", device_dir / "CLICK_POINT.JSON"),
        ]

        for name, path in device_files:
            result["device"].append({
                "name": name,
                "path": str(path),
                "exists": path.exists()
            })

    # 运行记录文件
    log_files = [
        ("端到端测试日志", data_dir / "reports" / "e2e_logs.json"),
        ("本地报告", data_dir / "reports" / "localhost.json"),
        ("VCA全配置", data_dir / "reports" / "vca_full_config.json"),
        ("VCA探测", data_dir / "reports" / "vca_probe.json"),
        ("搜索基线", data_dir / "search_baseline.json"),
    ]

    for name, path in log_files:
        result["logs"].append({
            "name": name,
            "path": str(path),
            "exists": path.exists()
        })

    return {"success": True, "data": result}


@api_router.get("/advanced/calib-time", summary="获取校准文件时间", tags=["Advanced"])
async def get_calib_time(mac: str, type: str):
    """获取校准文件的时间(优先从文件名提取,否则用文件修改时间)"""
    import re
    from datetime import datetime

    # v1.0.2: frozen 下 __file__ 指向 _internal/，统一走 config_paths
    from src.config_paths import DEVICES_DIR

    device_dir = DEVICES_DIR / mac

    file_map = {
        "function": "function.json",
        "limit": "limit.json",
        "speed": "speed.json",
        "staralign": "calibration.json",
        "clickpoint": "CLICK_POINT.JSON"
    }

    if type not in file_map:
        return {"success": False, "message": "无效类型"}

    base_name = file_map[type]
    test_type = base_name.replace('.json', '').replace('.JSON', '')

    # 优先查找带时间戳的文件: {test_type}_YYYYMMDD_HHMMSS.json
    pattern = re.compile(rf'^{re.escape(test_type)}_(\d{{8}}_\d{{6}})\.json$', re.IGNORECASE)
    latest_file = None
    latest_time = None
    latest_time_str = None

    if device_dir.exists():
        for f in device_dir.iterdir():
            if f.is_file():
                match = pattern.match(f.name)
                if match:
                    time_str = match.group(1)  # YYYYMMDD_HHMMSS
                    file_time = datetime.strptime(time_str, "%Y%m%d_%H%M%S")
                    if latest_time is None or file_time > latest_time:
                        latest_time = file_time
                        latest_file = f
                        latest_time_str = file_time.strftime("%Y-%m-%d %H:%M:%S")

    # 如果找到带时间戳的文件,返回文件名中的时间
    if latest_file:
        return {"success": True, "data": {"time": latest_time_str}}

    # 否则查找固定名称文件
    file_path = device_dir / base_name
    if not file_path.exists():
        return {"success": True, "data": {"time": None}}

    # 使用文件修改时间
    mtime = file_path.stat().st_mtime
    dt = datetime.fromtimestamp(mtime)
    time_str = dt.strftime("%Y-%m-%d %H:%M:%S")

    return {"success": True, "data": {"time": time_str}}


# ================================================================
# 依赖环境检查 API (v1.0.9: 移除旧 updater.exe 端点，统一使用 /update/*)
# ================================================================

@api_router.get("/updater/env-check")
async def updater_env_check() -> dict[str, Any]:
    """依赖环境检查。

    检查 OBS、GStreamer、插件、MediaMTX 等组件状态。
    v1.1.36: OBS 路径取自运行时配置（load_obs_config），避免自定义安装路径下误报。
    """
    from src.obs.obs_config import load_obs_config
    from src.updater.env_checker import EnvChecker

    cfg = load_obs_config()
    checker = EnvChecker(obs_path=cfg.obs_exe.parent, gst_path=None)
    # v1.1.49: get_summary 含子进程探测（gst-inspect 等），移入线程池避免阻塞事件循环
    return await asyncio.to_thread(checker.get_summary)


# ================================================================ #
#  观测计划 API (v1.04)
# ================================================================ #


class ObservationPlanRequest(BaseModel):
    """观测计划请求。"""
    id: str | None = None
    type: str = "scheduled"  # scheduled | timelapse | sunrise_sunset | solar_transit
    name: str = ""
    target: dict | None = None
    start_time: str | None = None
    stop_time: str | None = None
    start_enabled: bool = True
    stop_enabled: bool = True
    action_screenshot: bool = False
    action_record: bool = False
    action_auto_stop: bool = True
    time_mode: str = "local"  # local | astronomical
    interval_seconds: int = 30
    astro_start_event: str = "sunset"
    astro_start_offset: int = 0
    astro_end_event: str = "astronomical_dawn"
    astro_end_offset: int = 0
    generate_video: bool = True
    end_time: str | None = None
    repeat_daily: bool = False  # v1.1.8: 每天重复（PTZhub 运行期间按时间自动执行）


@api_router.get("/observation/plans", summary="获取所有观测计划", tags=["Observation"])
async def observation_plans_list() -> dict:
    """获取所有观测计划。"""
    try:
        from src.advanced.scheduler import get_all_plans
        plans = get_all_plans()
        return {"success": True, "data": plans}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/observation/plans", summary="创建/更新观测计划", tags=["Observation"])
async def observation_plans_save(req: ObservationPlanRequest) -> dict:
    """创建或更新观测计划。"""
    try:
        from src.advanced.scheduler import save_plan
        plan = req.model_dump()
        result = save_plan(plan)
        return {"success": True, "data": result}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/observation/plans/{plan_id}/delete", summary="删除观测计划", tags=["Observation"])
async def observation_plans_delete(plan_id: str) -> dict:
    """删除观测计划。"""
    try:
        from src.advanced.scheduler import delete_plan
        ok = delete_plan(plan_id)
        return {"success": ok, "message": "已删除" if ok else "计划不存在"}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/observation/plans/{plan_id}/execute", summary="执行观测计划", tags=["Observation"])
async def observation_plans_execute(plan_id: str) -> dict:
    """立即执行观测计划。"""
    try:
        from src.advanced.scheduler import execute_plan
        return await execute_plan(plan_id)
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/observation/plans/{plan_id}/cancel", summary="取消观测计划", tags=["Observation"])
async def observation_plans_cancel(plan_id: str) -> dict:
    """取消正在执行的观测计划。"""
    try:
        from src.advanced.scheduler import cancel_plan
        ok = cancel_plan(plan_id)
        return {"success": ok, "message": "已取消" if ok else "无活跃任务"}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/observation/plans/{plan_id}/status", summary="获取计划执行状态", tags=["Observation"])
async def observation_plans_status(plan_id: str) -> dict:
    """获取观测计划执行状态。"""
    try:
        from src.advanced.scheduler import get_plan_status
        return get_plan_status(plan_id)
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/observation/astronomical-events", summary="获取当日天文事件", tags=["Observation"])
async def observation_astro_events(date: str | None = None) -> dict:
    """获取指定日期的天文事件时间（晨昏蒙影 + 日出日落）。

    复用 altitude_chart.py 中的计算函数。
    """
    try:
        from datetime import datetime as dt

        from src.advanced.scheduler import get_astronomical_events_for_date

        target = dt.strptime(date, "%Y-%m-%d") if date else dt.now()

        events = get_astronomical_events_for_date(target)
        return {"success": True, "data": events}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.get("/observation/active-functions", summary="获取观测功能运行状态", tags=["Observation"])
async def observation_active_functions() -> dict:
    """获取各观测功能（延时摄影/日出日落/中天记录）的运行状态。"""
    try:
        from src.advanced.scheduler import get_active_functions
        return {"success": True, "data": get_active_functions()}
    except Exception as e:
        return {"success": False, "message": str(e)}


# ================================================================ #
#  依赖下载 API (v1.03)
# ================================================================ #


# 依赖下载映射（v1.06 重构：每个组件自带 download_url，直连官方源）
DEPENDENCY_DOWNLOAD_MAP = {
    "ffmpeg": {
        "filename": "ffmpeg-release-essentials.zip",
        "description": "FFmpeg 视频处理工具",
        "download_url": "https://www.gyan.dev/ffmpeg/builds/ffmpeg-release-essentials.zip",
        # v1.1.24: 境外主源不通时回退自有 OTA 快照（rolling release 快照固定 9.0.1 essentials）
        "fallback_url": "https://ota.philtchang.xyz/packages/astrohub/deps/ffmpeg-9.0.1-essentials.zip",
    },
    "obs": {
        "filename": "OBS-Studio-32.2.2-Windows-x64-Installer.exe",
        "description": "OBS Studio 直播/录制软件",
        "download_url": "https://github.com/obsproject/obs-studio/releases/download/32.2.2/OBS-Studio-32.2.2-Windows-x64-Installer.exe",
    },
    "obs_plugins": {
        "filename": "astrohub-obs.dll",
        "description": "AstroHub OBS 插件",
        "download_url": "https://ota.philtchang.xyz/packages/astrohub/plugins/astrohub-obs.dll",
        # v1.1.4: 运行期修复附加 DLL（obs-gstreamer 注册 gstreamer-source kind，
        # 缺失时媒体源创建必败；发布时需补传至 OTA plugins 目录）
        # v1.1.19: 补 astrohub-multi-rtmp.dll——推流目标依赖，缺失则 target_ok 永假、
        # 监督器永不变绿；发布时三个 DLL 一并同步至 OTA plugins 目录
        "extra_files": [
            {
                "filename": "obs-gstreamer.dll",
                "download_url": "https://ota.philtchang.xyz/packages/astrohub/plugins/obs-gstreamer.dll",
            },
            {
                "filename": "astrohub-multi-rtmp.dll",
                "download_url": "https://ota.philtchang.xyz/packages/astrohub/plugins/astrohub-multi-rtmp.dll",
            },
        ],
    },
    "gstreamer": {
        "filename": "gstreamer-1.0-msvc-x86_64-1.28.4.exe",
        "description": "GStreamer 多媒体框架",
        "download_url": "https://gstreamer.freedesktop.org/data/pkg/windows/1.28.4/msvc/gstreamer-1.0-msvc-x86_64-1.28.4.exe",
    },
    "mediamtx": {
        "filename": "mediamtx.zip",
        "description": "MediaMTX 流媒体服务器",
        "download_url": "https://github.com/bluenviron/mediamtx/releases/download/v1.12.0/mediamtx_v1.12.0_windows_amd64.zip",
        # v1.1.24: GitHub 不通时回退自有 OTA deps 快照（sha256 与官方一致）
        "fallback_url": "https://ota.philtchang.xyz/packages/astrohub/deps/mediamtx_v1.12.0_windows_amd64.zip",
    },
}


def _dependency_urls(name: str) -> list[str]:
    """v1.1.24: 依赖下载源列表（官方主源 + 自有 OTA 回退源），供运行期下载器双源回退。"""
    info = DEPENDENCY_DOWNLOAD_MAP.get(name, {})
    urls = [info.get("download_url", "")]
    fallback = info.get("fallback_url")
    if fallback:
        urls.append(fallback)
    return [u for u in urls if u]


def _obs_effect_source_dir() -> Path:
    """v1.1.4: effect 数据源目录（冻结包用打包内置资源，dev 模式用源码目录）。"""
    import sys

    if getattr(sys, "frozen", False):
        base = Path(getattr(sys, "_MEIPASS", ""))
        return base / "obs-plugin" / "astrohub-obs" / "data" / "obs-plugins" / "astrohub-obs"
    return (
        Path(__file__).resolve().parents[2]
        / "obs-plugin" / "astrohub-obs" / "data" / "obs-plugins" / "astrohub-obs"
    )


def _gst_bin_source() -> Path | None:
    """v1.1.5: GST 运行时 DLL 同步源（ABI 校验解析根，强制刷新缓存）。"""
    from src.updater.env_checker import resolve_gst_root

    root = resolve_gst_root(force=True)
    return (root / "bin") if root is not None else None


@api_router.get("/dependencies/status", summary="检查依赖安装状态", tags=["Dependencies"])
async def dependencies_status() -> dict:
    """检查各依赖的安装状态和下载可用性。"""
    try:
        from src.updater.env_checker import EnvChecker
        checker = EnvChecker()
        results = checker.check_all()

        data = {}
        for comp_id, result in results.items():
            info = result.to_dict()
            download_info = DEPENDENCY_DOWNLOAD_MAP.get(comp_id, {})
            data[comp_id] = {
                "installed": info.get("status") in ("ok", "warning"),
                "status": info.get("status"),
                "path": info.get("path", ""),
                "version": info.get("version", ""),
                "message": info.get("message", ""),
                "download_url": download_info.get("download_url", "") if download_info else "",
                "description": download_info.get("description", comp_id),
            }

        return {"success": True, "data": data}
    except Exception as e:
        return {"success": False, "message": str(e)}


@api_router.post("/dependencies/download/{name}", summary="下载依赖", tags=["Dependencies"])
async def dependencies_download(name: str):
    """从官方源下载指定依赖到 data/downloads/ 目录，SSE 流式推送进度。"""
    import json as _json

    from fastapi.responses import StreamingResponse

    if name not in DEPENDENCY_DOWNLOAD_MAP:
        return StreamingResponse(
            iter([f"data: {_json.dumps({'success': False, 'message': f'未知依赖: {name}'})}\n\n"]),
            media_type="text/event-stream",
        )

    download_info = DEPENDENCY_DOWNLOAD_MAP[name]

    from src.config_paths import DOWNLOAD_DIR
    dest = DOWNLOAD_DIR / download_info["filename"]

    async def event_generator():
        DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
        try:
            import httpx
        except ImportError:
            yield f"data: {_json.dumps({'success': False, 'message': '服务端缺少 httpx 依赖，请重新安装或更新程序'})}\n\n"
            return
        last_err = ""
        for _url in _dependency_urls(name):
            try:
                async with (
                    httpx.AsyncClient(follow_redirects=True) as client,
                    client.stream("GET", _url, timeout=httpx.Timeout(connect=30, read=0, write=30, pool=30)) as resp,
                ):
                    if resp.status_code != 200:
                        last_err = f"HTTP {resp.status_code}"
                        continue
                    total = int(resp.headers.get("content-length", 0))
                    downloaded = 0
                    last_pct = -1.0

                    with open(dest, "wb") as f:
                        async for chunk in resp.aiter_bytes(chunk_size=65536):
                            f.write(chunk)
                            downloaded += len(chunk)
                            if total > 0:
                                pct = round(downloaded * 100 / total, 1)
                                # 节流：每 1% 才推送一次
                                if pct - last_pct >= 1.0 or pct >= 100:
                                    last_pct = pct
                                    yield f"data: {_json.dumps({'success': True, 'progress': pct, 'downloaded': downloaded, 'total': total})}\n\n"
                            else:
                                # 无 Content-Length 时每 5MB 推送一次
                                if downloaded % (5 * 1024 * 1024) < 65536:
                                    yield f"data: {_json.dumps({'success': True, 'progress': -1, 'downloaded': downloaded})}\n\n"
                break
            except Exception as e:
                last_err = str(e)
        else:
            log_warn("dependency", "download_failed", f"{name}: {last_err}")
            yield f"data: {_json.dumps({'success': False, 'message': f'下载失败(官方源与OTA回退源均尝试): {last_err}'})}\n\n"
            return
        yield f"data: {_json.dumps({'success': True, 'progress': 100, 'message': f'{name} 下载完成', 'data': {'path': str(dest), 'name': name}})}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")


@api_router.post("/dependencies/install/{name}", summary="下载并安装依赖 (SSE 阶段进度)", tags=["Dependencies"])
async def dependencies_install(name: str):
    """v1.1.0: 依赖修复闭环 — 下载 → 安装 → OBS 初始化，SSE 全程推送阶段事件。

    阶段: download(进度) → install(静默安装/解压) → init_obs(场景/源/滤镜) → done。
    """
    import json as _json

    if name not in DEPENDENCY_DOWNLOAD_MAP:
        return StreamingResponse(
            iter([f"data: {_json.dumps({'success': False, 'message': f'未知依赖: {name}'})}\n\n"]),
            media_type="text/event-stream",
        )

    download_info = DEPENDENCY_DOWNLOAD_MAP[name]

    from src.config_paths import DOWNLOAD_DIR

    dest = DOWNLOAD_DIR / download_info["filename"]

    async def event_generator():
        def ev(**kw: Any) -> str:
            return f"data: {_json.dumps(kw, ensure_ascii=False)}\n\n"

        # ---- 阶段 1: 下载（v1.1.24: 官方主源与 OTA 回退源双源回退） ----
        yield ev(success=True, phase="download", message=f"正在下载 {name}...")
        try:
            import httpx
        except ImportError:
            yield ev(success=False, message="服务端缺少 httpx 依赖，请重新安装或更新程序")
            return
        last_err = ""
        for _url in _dependency_urls(name):
            try:
                DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
                async with (
                    httpx.AsyncClient(follow_redirects=True) as client,
                    client.stream("GET", _url, timeout=httpx.Timeout(connect=30, read=0, write=30, pool=30)) as resp,
                ):
                    if resp.status_code != 200:
                        last_err = f"HTTP {resp.status_code}"
                        continue
                    total = int(resp.headers.get("content-length", 0))
                    downloaded = 0
                    last_pct = -1.0
                    with open(dest, "wb") as f:
                        async for chunk in resp.aiter_bytes(chunk_size=65536):
                            f.write(chunk)
                            downloaded += len(chunk)
                            if total > 0:
                                pct = round(downloaded * 100 / total, 1)
                                if pct - last_pct >= 5.0 or pct >= 100:
                                    last_pct = pct
                                    yield ev(success=True, phase="download", progress=pct,
                                             message=f"下载 {name}: {pct}%")
                break
            except Exception as e:
                last_err = str(e)
        else:
            log_warn("dependency", "install_download_failed", f"{name}: {last_err}")
            yield ev(success=False, message=f"下载失败(官方源与OTA回退源均尝试): {last_err}")
            return
        yield ev(success=True, phase="download", progress=100, message=f"{name} 下载完成")

        # ---- 阶段 2: 安装 ----
        import subprocess
        import zipfile

        yield ev(success=True, phase="install", message=f"正在安装 {name}...")
        try:
            if name in ("obs", "gstreamer"):
                # 官方安装程序静默安装
                if name == "obs":
                    cmd = [str(dest), "/S"]
                    yield ev(success=True, phase="install", message="OBS 静默安装中（约 1-2 分钟）...")
                else:
                    # v1.1.5: /DIR 改无空格 msvc 根（空格截断静默失败陷阱）
                    cmd = [str(dest), "/SILENT", "/NORESTART", "/ALLUSERS", "/TYPE=runtime",
                           "/DIR=C:\\gstreamer\\1.0\\msvc_x86_64"]
                    yield ev(success=True, phase="install", message="GStreamer 静默安装中...")
                proc = await asyncio.to_thread(
                    subprocess.run, cmd, capture_output=True, timeout=900,
                    creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0,
                )
                if proc.returncode != 0:
                    yield ev(success=False, message=f"{name} 静默安装返回码 {proc.returncode}，请在下载目录手动运行安装包")
                    return
                # v1.1.5: GStreamer 装后同步运行时 DLL 到 OBS bin（OBS 安全 DLL 搜索不查 PATH）
                if name == "gstreamer":
                    import shutil

                    gst_bin = _gst_bin_source()
                    if gst_bin is not None and gst_bin.exists():
                        from src.obs.obs_config import load_obs_config

                        obs_bin = Path(load_obs_config().bin_path)

                        def _copy_gst_after_install() -> int:
                            copied = 0
                            for dll in gst_bin.glob("*.dll"):
                                try:
                                    shutil.copy2(dll, obs_bin / dll.name)
                                    copied += 1
                                except Exception:
                                    pass
                            return copied

                        cnt = await asyncio.to_thread(_copy_gst_after_install)
                        yield ev(success=True, phase="install",
                                 message=f"GStreamer 运行时 DLL 已同步到 OBS bin（{cnt} 个）")
            elif name in ("mediamtx", "ffmpeg"):
                # zip 解压就位
                from src.updater.version_manager import get_app_dir

                if name == "mediamtx":
                    target = get_app_dir() / "bin" / "mediamtx"
                    strip = 0
                    yield ev(success=True, phase="install", message="正在解压缩 MediaMTX...")
                else:
                    target = get_app_dir() / "tools" / "ffmpeg"
                    strip = 1  # essentials zip 含顶层目录
                    yield ev(success=True, phase="install", message="正在解压缩 FFmpeg...")
                target.mkdir(parents=True, exist_ok=True)

                def _extract() -> None:
                    with zipfile.ZipFile(dest) as zf:
                        for info in zf.infolist():
                            parts = info.filename.split("/")
                            if strip:
                                if len(parts) <= 1:
                                    continue
                                info.filename = "/".join(parts[strip:])
                                if not info.filename:
                                    continue
                            zf.extract(info, target)

                await asyncio.to_thread(_extract)
            elif name == "obs_plugins":
                # v1.1.4: 插件完整修复——主 DLL + 附加 DLL + GST 运行时 DLL 进 OBS bin
                # + effect 数据部署（对齐 NSIS 安装期职责，治运行期修复死端）
                from src.obs.obs_config import load_obs_config

                obs_cfg = load_obs_config()
                plugin_dir = Path(obs_cfg.plugin_path)
                plugin_dir.mkdir(parents=True, exist_ok=True)
                yield ev(success=True, phase="install", message="正在安装 OBS 插件...")
                import shutil
                await asyncio.to_thread(shutil.copy2, dest, plugin_dir / dest.name)
                # 附加插件 DLL（obs-gstreamer.dll）
                for item in DEPENDENCY_DOWNLOAD_MAP[name].get("extra_files", []):
                    extra_dest = DOWNLOAD_DIR / item["filename"]
                    try:
                        async with httpx.AsyncClient(follow_redirects=True) as ecx:
                            r = await ecx.get(item["download_url"], timeout=180)
                            r.raise_for_status()
                            await asyncio.to_thread(extra_dest.write_bytes, r.content)
                        await asyncio.to_thread(shutil.copy2, extra_dest, plugin_dir / item["filename"])
                        yield ev(success=True, phase="install", message=f"{item['filename']} 已部署")
                    except Exception as e:
                        log_warn("dependency", "extra_plugin_failed", f"{item['filename']}: {e}")
                        yield ev(success=True, phase="install",
                                 message=f"{item['filename']} 获取失败（可重装安装包补齐）: {e}")
                # GST 运行时核心 DLL 复制到 OBS bin（OBS 安全 DLL 搜索不查 PATH，对齐 nsi xcopy）
                # v1.1.5: 同步源改 ABI 校验解析根（不再写死 PF 根）
                gst_bin = _gst_bin_source()
                obs_bin = Path(obs_cfg.bin_path)
                if gst_bin is not None and gst_bin.exists() and obs_bin.exists():
                    def _copy_gst() -> int:
                        copied = 0
                        for dll in gst_bin.glob("*.dll"):
                            try:
                                shutil.copy2(dll, obs_bin / dll.name)
                                copied += 1
                            except Exception:
                                pass
                        return copied

                    cnt = await asyncio.to_thread(_copy_gst)
                    yield ev(success=True, phase="install", message=f"GStreamer 运行时 DLL 已同步到 OBS bin（{cnt} 个）")
                # effect 数据文件（astro_stack.effect 等，滤镜着色器依赖）
                effect_src = _obs_effect_source_dir()
                if effect_src.exists():
                    effect_dst = plugin_dir.parent.parent / "data" / "obs-plugins" / "astrohub-obs"

                    def _copy_effect() -> int:
                        effect_dst.mkdir(parents=True, exist_ok=True)
                        n = 0
                        for f in effect_src.rglob("*"):
                            if f.is_file():
                                shutil.copy2(f, effect_dst / f.relative_to(effect_src))
                                n += 1
                        return n

                    ecnt = await asyncio.to_thread(_copy_effect)
                    yield ev(success=True, phase="install", message=f"effect 数据文件已部署（{ecnt} 个）")
            else:
                yield ev(success=False, message=f"{name} 暂不支持自动安装")
                return
        except Exception as e:
            log_warn("dependency", "install_failed", f"{name}: {e}")
            yield ev(success=False, message=f"安装失败: {e}")
            return
        yield ev(success=True, phase="install", message=f"{name} 安装完成")

        # ---- 阶段 3: OBS 初始化（场景/媒体源/滤镜）----
        yield ev(success=True, phase="init_obs", message="正在初始化 OBS 场景/媒体源/滤镜...")
        try:
            from src.obs.obs_manager import get_obs_manager

            mgr = get_obs_manager()
            await mgr.start(spawn_obs=True)
            yield ev(success=True, phase="init_obs", message="OBS 场景图检查/创建完成")
        except Exception as e:
            log_warn("dependency", "obs_init_failed", str(e))
            yield ev(success=True, phase="init_obs", message=f"OBS 初始化未完成（可稍后重试）: {e}")

        yield ev(success=True, phase="done", message=f"{name} 下载+安装完成", data={"name": name})

    return StreamingResponse(event_generator(), media_type="text/event-stream")


# ================================================================ #
#  License 授权认证 API
# ================================================================ #


class LicenseActivateRequest(BaseModel):
    """序列号激活请求。"""
    serial: str


@api_router.get("/license/status", summary="查询授权状态", tags=["License"])
async def license_status() -> dict:
    """查询当前授权状态。

    返回授权状态、序列号（脱敏）、硬件指纹摘要等信息。
    此端点不受授权保护，未授权用户也可调用。
    """
    from src.license.license_manager import get_license_manager

    license_mgr = get_license_manager()
    return license_mgr.get_status_dict()


@api_router.post("/license/activate", summary="激活序列号", tags=["License"])
async def license_activate(req: LicenseActivateRequest) -> dict:
    """使用序列号激活软件。

    流程：
    1. 采集本机硬件指纹
    2. 调用授权服务器验证序列号
    3. 绑定硬件指纹
    4. 保存 license 文件

    此端点不受授权保护，未授权用户可调用。
    """
    serial = req.serial.strip().upper()

    # 基本格式校验
    if len(serial) < 20:
        return {"success": False, "message": "序列号格式错误"}

    from src.license.license_manager import get_license_manager

    license_mgr = get_license_manager()
    result = await license_mgr.activate(serial)

    return {
        "success": result.success,
        "message": result.message,
        "license_type": result.license_type,
        "expires_at": result.expires_at,
    }


@api_router.post("/license/deactivate", summary="解绑设备", tags=["License"])
async def license_deactivate() -> dict:
    """解绑当前设备。

    解绑后需要重新输入序列号激活。
    此端点不受授权保护。
    """
    from src.license.license_manager import get_license_manager

    license_mgr = get_license_manager()
    result = await license_mgr.deactivate()
    return result


@api_router.get("/license/fingerprint", summary="获取硬件指纹", tags=["License"])
async def license_fingerprint() -> dict:
    """获取当前机器的硬件指纹摘要。

    用于用户联系客服时核对设备信息。
    此端点不受授权保护。
    """
    from src.license.fingerprint import collect_fingerprint

    fingerprint = collect_fingerprint()
    return {
        "success": True,
        "fingerprint": fingerprint[:16].upper() if fingerprint else "",
        "full_fingerprint": fingerprint,
    }