import os
import re
import subprocess
import json
import time
from pathlib import Path
from typing import Any, Dict, Optional, Tuple

from fastapi import APIRouter, HTTPException, Header, Depends, BackgroundTasks
from pydantic import BaseModel
import logging

logger = logging.getLogger(__name__)

# 路由统一挂在 /api/system 下，前端直接调用 /api/system/...
router = APIRouter(prefix="/api/system")

# 管理员鉴权（JWT；未启用登录时兼容 X-Admin-Token）
from api.auth_deps import require_system_admin  # noqa: E402
from database.models import Account  # noqa: E402

LOG_GROUPS = ("error", "warning", "info")

# 后端服务启动时间（用于前端展示“运行多久/是否已重启”）
_BACKEND_STARTED_AT_MS: int = int(time.time() * 1000)

# 系统元信息存储（优先 Redis；用于记录重启时间等）
def _system_meta_prefix() -> str:
    return (os.getenv("SYSTEM_META_PREFIX", "ats:system").strip() or "ats:system")


def _system_meta_key(name: str) -> str:
    name = (name or "").strip()
    return f"{_system_meta_prefix()}:{name}"


def _system_meta_read(name: str) -> Optional[Dict[str, Any]]:
    client = _get_redis_client_for_logs()
    if client is None:
        return None
    try:
        raw = client.get(_system_meta_key(name))
        if not raw:
            return None
        return json.loads(raw)
    except Exception:
        return None


def _system_meta_write(name: str, payload: Dict[str, Any], ttl_sec: int = 30 * 24 * 3600) -> None:
    client = _get_redis_client_for_logs()
    if client is None:
        return
    try:
        client.setex(_system_meta_key(name), int(ttl_sec), json.dumps(payload, ensure_ascii=False))
    except Exception:
        return

# 避免 Redis 异常刷屏（前端可能自动刷新）
_last_logs_redis_err_ts: float = 0.0


def _logs_prefix() -> str:
    return (os.getenv("REDIS_LOG_LIST_PREFIX", "ats:logs").strip() or "ats:logs")


def _logs_key_for_group(group: str) -> str:
    group = (group or "error").strip().lower()
    # 兼容旧配置：REDIS_LOG_LIST_KEY 仅用于 error
    if group == "error":
        legacy = os.getenv("REDIS_LOG_LIST_KEY", "").strip()
        if legacy:
            return legacy

    env_key = os.getenv(f"REDIS_LOG_LIST_KEY_{group.upper()}", "").strip()
    if env_key:
        return env_key

    return f"{_logs_prefix()}:{group}"


def _logs_config_key() -> str:
    return (os.getenv("REDIS_LOG_CONFIG_KEY", "ats:logs:config").strip() or "ats:logs:config")


def _logs_stats_prefix() -> str:
    return (os.getenv("REDIS_LOG_STATS_PREFIX", "ats:logs:stats:added").strip() or "ats:logs:stats:added")


def _beijing_yyyymmdd() -> str:
    from datetime import datetime, timezone, timedelta

    beijing_tz = timezone(timedelta(hours=8))
    return datetime.now(tz=beijing_tz).strftime("%Y%m%d")


def _default_logs_config() -> Dict[str, Any]:
    return {
        "max_len": {"error": 2000, "warning": 2000, "info": 2000},
        "enabled": {"error": True, "warning": True, "info": True},
        "dedupe_consecutive": True,
        "include_debug_in_info": False,
        "keys": {g: _logs_key_for_group(g) for g in LOG_GROUPS},
        "config_key": _logs_config_key(),
        "stats_prefix": _logs_stats_prefix(),
    }


def _merge_logs_config(defaults: Dict[str, Any], redis_hash: Dict[str, str]) -> Dict[str, Any]:
    cfg = defaults

    for g in LOG_GROUPS:
        v = redis_hash.get(f"max_len:{g}")
        if v is not None:
            try:
                n = int(str(v).strip())
                if n > 0:
                    cfg["max_len"][g] = n
            except Exception:
                pass

        ev = redis_hash.get(f"enabled:{g}")
        if ev is not None:
            s = str(ev).strip().lower()
            cfg["enabled"][g] = s in ("1", "true", "yes", "y", "on")

    for k in ("dedupe_consecutive", "include_debug_in_info"):
        vv = redis_hash.get(k)
        if vv is not None:
            s = str(vv).strip().lower()
            cfg[k] = s in ("1", "true", "yes", "y", "on")

    return cfg


def _read_logs_config(client) -> Dict[str, Any]:
    defaults = _default_logs_config()
    try:
        raw = client.hgetall(_logs_config_key()) or {}
        return _merge_logs_config(defaults, raw)
    except Exception:
        return defaults


def _write_logs_config_and_trim(client, cfg: Dict[str, Any]) -> Dict[str, Any]:
    mapping: Dict[str, str] = {}
    for g in LOG_GROUPS:
        mapping[f"max_len:{g}"] = str(int(cfg["max_len"][g]))
        mapping[f"enabled:{g}"] = "1" if cfg["enabled"][g] else "0"
    mapping["dedupe_consecutive"] = "1" if cfg.get("dedupe_consecutive") else "0"
    mapping["include_debug_in_info"] = "1" if cfg.get("include_debug_in_info") else "0"

    # 注意：AWS Valkey/Redis 集群模式下，MULTI/EXEC 不能跨 slot
    # 这里会同时操作多个 key（config hash + 3 个 list），所以必须禁用 transaction
    pipe = client.pipeline(transaction=False)
    pipe.hset(_logs_config_key(), mapping=mapping)
    for g in LOG_GROUPS:
        key = _logs_key_for_group(g)
        max_len = int(cfg["max_len"][g])
        if max_len > 0:
            pipe.ltrim(key, 0, max_len - 1)
    pipe.execute()
    return cfg


class LogsConfigUpdate(BaseModel):
    max_len: Optional[Dict[str, int]] = None
    enabled: Optional[Dict[str, bool]] = None
    dedupe_consecutive: Optional[bool] = None
    include_debug_in_info: Optional[bool] = None


def _beijing_time_str() -> str:
    from datetime import datetime, timezone, timedelta

    beijing_tz = timezone(timedelta(hours=8))
    return datetime.now(tz=beijing_tz).strftime("%Y-%m-%d %H:%M:%S")


@router.post("/logs/test-write")
async def logs_test_write(
    _admin: Dict[str, Any] = Depends(require_system_admin),
) -> Dict[str, Any]:
    """
    写入 3 条测试日志到 Redis（error/warning/info），用于验证“是否写入到同一台 Redis、同一组 key”。
    """
    client = _get_redis_client_for_logs()
    if client is None:
        raise HTTPException(status_code=503, detail="Redis 不可用，无法写入测试日志")

    cfg = _read_logs_config(client)
    now_ms = int(__import__("time").time() * 1000)
    time_str = _beijing_time_str()

    entries = {
        "error": {"level": "ERROR", "message": f"[TEST] backend 写入 error 测试日志 @ {time_str}"},
        "warning": {"level": "WARNING", "message": f"[TEST] backend 写入 warning 测试日志 @ {time_str}"},
        "info": {"level": "INFO", "message": f"[TEST] backend 写入 info 测试日志 @ {time_str}"},
    }

    day = _beijing_yyyymmdd()
    stats_prefix = _logs_stats_prefix()

    # 集群模式下禁用 transaction，避免 CROSSSLOT
    pipe = client.pipeline(transaction=False)
    for g in LOG_GROUPS:
        key = _logs_key_for_group(g)
        max_len = int(cfg["max_len"][g])
        entry = {
            "ts": now_ms,
            "time": time_str,
            "service": "backend",
            "level": entries[g]["level"],
            "logger": "api.routes.system",
            "message": entries[g]["message"],
            "hostname": os.getenv("HOSTNAME", ""),
            "signature": f"backend|{entries[g]['level']}|test|{entries[g]['message']}",
            "count": 1,
        }
        pipe.lpush(key, json.dumps(entry, ensure_ascii=False))
        if max_len > 0:
            pipe.ltrim(key, 0, max_len - 1)
        pipe.incr(f"{stats_prefix}:{day}:{g}", 1)
        pipe.expire(f"{stats_prefix}:{day}:{g}", 14 * 24 * 3600)

    pipe.execute()

    # 返回写入后的 LLEN，便于你确认
    # 单 key LLEN 查询也不需要 transaction
    pipe2 = client.pipeline(transaction=False)
    for g in LOG_GROUPS:
        pipe2.llen(_logs_key_for_group(g))
    llens = pipe2.execute()

    return {
        "message": "ok",
        "keys": {g: _logs_key_for_group(g) for g in LOG_GROUPS},
        "llen": {g: int(llens[i] or 0) for i, g in enumerate(LOG_GROUPS)},
        "note": "如果你在前端仍看不到，说明前端请求的后端实例/Redis key/筛选条件不一致。",
    }


@router.post("/trading/trigger-scan")
async def trigger_scan(
    _admin: Dict[str, Any] = Depends(require_system_admin),
) -> Dict[str, Any]:
    """
    触发手动扫描：
    通过设置 Redis 信号（ats:trigger-scan），通知所有运行中的 strategy 进程立即执行扫描。
    """
    client = _get_redis_client_for_logs()
    if client is None:
        raise HTTPException(status_code=503, detail="Redis 不可用，无法触发扫描")

    try:
        # 设置触发信号（当前时间戳），让 strategy 检测到变化
        import time
        ts = int(time.time())
        # 使用 setex 设置 600秒过期，防止永久残留（虽然 strategy 只关心变化，但过期是个好习惯）
        # 注意：strategy 端比较的是时间戳大小，所以只要比上一次大即可
        client.setex("ats:trigger-scan", 600, str(ts))
        
        return {
            "message": "已发送扫描触发信号",
            "timestamp": ts
        }
    except Exception as e:
        logger.error(f"触发扫描失败: {e}")
        raise HTTPException(status_code=500, detail=f"触发扫描失败: {e}")


def _get_redis_client_for_logs():
    """
    获取 Redis 客户端（优先复用 config_manager 的连接；失败则自行创建）。
    返回：redis.Redis 或 None
    """
    # 1) 复用 config_manager（避免重复连接）
    try:
        import config_manager  # backend/config_manager.py（已负责加载 .env）

        cm = getattr(config_manager, "config_manager", None)
        if cm is not None:
            redis_client = getattr(cm, "_redis_client", None)
            redis_connected = getattr(cm, "_redis_connected", False)
            if redis_client is not None and redis_connected:
                try:
                    redis_client.ping()
                    return redis_client
                except Exception:
                    pass
    except Exception:
        pass

    # 2) 自行创建
    try:
        import redis  # type: ignore

        redis_url = os.getenv("REDIS_URL", "redis://localhost:6379")
        redis_use_tls = os.getenv("REDIS_USE_TLS", "False").lower() == "true"
        redis_username = os.getenv("REDIS_USERNAME", None)
        redis_password = os.getenv("REDIS_PASSWORD", None)
        ssl_cert_reqs = os.getenv("REDIS_SSL_CERT_REQS", "required")
        ssl_ca_certs = os.getenv("REDIS_SSL_CA_CERTS", None)

        kwargs: Dict[str, Any] = {
            "decode_responses": True,
            "username": redis_username,
            "password": redis_password,
            "socket_connect_timeout": 1,
            "socket_timeout": 1,
        }
        if redis_url.startswith("rediss://") or redis_use_tls:
            kwargs["ssl_cert_reqs"] = ssl_cert_reqs
            if ssl_ca_certs:
                kwargs["ssl_ca_certs"] = ssl_ca_certs
            if ssl_cert_reqs == "none":
                kwargs["ssl_check_hostname"] = False
            elif ssl_cert_reqs == "required":
                kwargs["ssl_check_hostname"] = True
            else:
                kwargs["ssl_check_hostname"] = False

        client = redis.from_url(redis_url, **kwargs)
        client.ping()
        return client
    except Exception as e:
        # 把错误尽量打到 api.log（你现在看的文件）
        global _last_logs_redis_err_ts
        import time as _t

        now = _t.time()
        if now - _last_logs_redis_err_ts > 30:
            _last_logs_redis_err_ts = now
            logger.warning(f"日志模块 Redis 连接失败。REDIS_URL={os.getenv('REDIS_URL', '')} err={e}")
        return None


@router.get("/logs")
async def get_logs(
    limit: int = 200,
    group: str = "error",
    start: int = 0,
    service: Optional[str] = None,
    level: Optional[str] = None,
    _admin: Dict[str, Any] = Depends(require_system_admin),
) -> Dict[str, Any]:
    """
    从 Redis List 读取最新日志（默认 group=error -> ats:logs:error）。

    参数：
    - limit: 返回条数（最大 2000）
    - group: 日志分组（error / warning / info）
    - service: 过滤（backend / trading_system）
    - level: 过滤（ERROR / CRITICAL ...）
    """
    if limit <= 0:
        limit = 200
    if limit > 20000:
        limit = 20000

    if start < 0:
        start = 0

    # 定义管理员不需要关注的日志模式（噪声过滤）
    IGNORED_PATTERNS = [
        "API密钥未配置",
        "请在配置界面设置该账号的BINANCE_API_KEY",
    ]

    group = (group or "error").strip().lower()
    if group not in LOG_GROUPS:
        raise HTTPException(status_code=400, detail=f"非法 group：{group}（可选：{', '.join(LOG_GROUPS)}）")

    list_key = _logs_key_for_group(group)

    client = _get_redis_client_for_logs()
    if client is None:
        raise HTTPException(status_code=503, detail="Redis 不可用，无法读取日志")

    try:
        llen_total = int(client.llen(list_key) or 0)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"读取 Redis 日志失败: {e}")

    if llen_total <= 0:
        return {
            "group": group,
            "key": list_key,
            "start": start,
            "limit": limit,
            "llen_total": 0,
            "next_start": start,
            "has_more": False,
            "count": 0,
            "items": [],
        }

    # 分页扫描：为了支持 service/level 过滤，这里会向后多取一些直到凑够 limit 或到末尾
    # 保护：最多扫描 limit*10 条，避免过滤太严格导致无限扫描
    max_scan = min(llen_total, start + limit * 10)
    pos = start
    scanned = 0
    items: list[Dict[str, Any]] = []

    try:
        while len(items) < limit and pos < llen_total and pos < max_scan:
            chunk_size = min(500, limit, max_scan - pos)
            end = pos + chunk_size - 1
            raw_batch = client.lrange(list_key, pos, end)
            scanned += len(raw_batch or [])

            for raw in raw_batch or []:
                try:
                    obj = raw
                    if isinstance(raw, bytes):
                        obj = raw.decode("utf-8", errors="ignore")
                    if not isinstance(obj, str):
                        continue
                    parsed = json.loads(obj)
                    if not isinstance(parsed, dict):
                        continue
                    if service and str(parsed.get("service")) != service:
                        continue
                    if level and str(parsed.get("level")) != level:
                        continue
                    
                    # 噪声过滤
                    msg = str(parsed.get("message", ""))
                    if any(p in msg for p in IGNORED_PATTERNS):
                        continue

                    items.append(parsed)
                    if len(items) >= limit:
                        break
                except Exception:
                    continue

            pos = end + 1

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"读取 Redis 日志失败: {e}")

    return {
        "group": group,
        "key": list_key,
        "start": start,
        "limit": limit,
        "llen_total": llen_total,
        "scanned": scanned,
        "next_start": pos,
        "has_more": pos < llen_total,
        "count": len(items),
        "items": items,
    }


@router.get("/logs/overview")
async def logs_overview(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:

    client = _get_redis_client_for_logs()
    if client is None:
        raise HTTPException(status_code=503, detail="Redis 不可用，无法读取日志概览")

    try:
        cfg = _read_logs_config(client)

        day = _beijing_yyyymmdd()
        stats_prefix = _logs_stats_prefix()

        # 集群模式下禁用 transaction，避免 CROSSSLOT
        pipe = client.pipeline(transaction=False)
        for g in LOG_GROUPS:
            pipe.llen(_logs_key_for_group(g))
        for g in LOG_GROUPS:
            pipe.get(f"{stats_prefix}:{day}:{g}")
        res = pipe.execute()

        llen_vals = res[: len(LOG_GROUPS)]
        added_vals = res[len(LOG_GROUPS) :]

        llen: Dict[str, int] = {}
        added_today: Dict[str, int] = {}
        for i, g in enumerate(LOG_GROUPS):
            try:
                llen[g] = int(llen_vals[i] or 0)
            except Exception:
                llen[g] = 0
            try:
                added_today[g] = int(added_vals[i] or 0)
            except Exception:
                added_today[g] = 0

        return {
            "config": cfg,
            "stats": {
                "day": day,
                "llen": llen,
                "added_today": added_today,
            },
            "meta": {
                "redis_url": os.getenv("REDIS_URL", ""),
                "keys": {g: _logs_key_for_group(g) for g in LOG_GROUPS},
            },
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"logs_overview 失败: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail=f"logs_overview failed: {e}")


@router.put("/logs/config")
async def update_logs_config(
    payload: LogsConfigUpdate,
    _admin: Dict[str, Any] = Depends(require_system_admin),
) -> Dict[str, Any]:
    client = _get_redis_client_for_logs()
    if client is None:
        raise HTTPException(status_code=503, detail="Redis 不可用，无法更新日志配置")

    cfg = _read_logs_config(client)

    if payload.max_len:
        for g, v in payload.max_len.items():
            gg = (g or "").strip().lower()
            if gg not in LOG_GROUPS:
                continue
            try:
                n = int(v)
                if n < 100:
                    n = 100
                if n > 20000:
                    n = 20000
                cfg["max_len"][gg] = n
            except Exception:
                continue

    if payload.enabled:
        for g, v in payload.enabled.items():
            gg = (g or "").strip().lower()
            if gg not in LOG_GROUPS:
                continue
            cfg["enabled"][gg] = bool(v)

    if payload.dedupe_consecutive is not None:
        cfg["dedupe_consecutive"] = bool(payload.dedupe_consecutive)

    if payload.include_debug_in_info is not None:
        cfg["include_debug_in_info"] = bool(payload.include_debug_in_info)

    cfg = _write_logs_config_and_trim(client, cfg)
    return {"message": "ok", "config": cfg}


def _require_admin(token: Optional[str], provided: Optional[str]) -> None:
    """
    可选的简单保护：如果环境变量配置了 SYSTEM_CONTROL_TOKEN，则要求请求携带 X-Admin-Token。
    生产环境强烈建议通过 Nginx 额外做鉴权 / IP 白名单。
    """
    if not token:
        return
    if not provided or provided != token:
        raise HTTPException(status_code=401, detail="Unauthorized")


#
# 注意：require_system_admin 已迁移到 api.auth_deps，避免导入不一致导致 uvicorn 启动失败


def _build_supervisorctl_cmd(args: list[str]) -> list[str]:
    supervisorctl_path = os.getenv("SUPERVISORCTL_PATH", "supervisorctl")
    supervisor_conf = os.getenv("SUPERVISOR_CONF", "").strip()
    use_sudo = os.getenv("SUPERVISOR_USE_SUDO", "false").lower() == "true"

    # 如果没显式配置 SUPERVISOR_CONF，就尝试自动探测常见路径（宝塔/系统）
    if not supervisor_conf:
        candidates = [
            "/www/server/panel/plugin/supervisor/supervisord.conf",
            "/www/server/panel/plugin/supervisor/supervisor.conf",
            "/etc/supervisor/supervisord.conf",
            "/etc/supervisord.conf",
        ]
        for p in candidates:
            try:
                if Path(p).exists():
                    supervisor_conf = p
                    break
            except Exception:
                continue

    cmd: list[str] = []
    if use_sudo:
        # 需要你在 sudoers 配置 NOPASSWD（sudo -n 才不会卡住）
        cmd += ["sudo", "-n"]
    cmd += [supervisorctl_path]
    if supervisor_conf:
        cmd += ["-c", supervisor_conf]
    cmd += args
    return cmd


def _run_supervisorctl(args: list[str]) -> str:
    cmd = _build_supervisorctl_cmd(args)
    try:
        res = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
    except subprocess.TimeoutExpired:
        raise RuntimeError("supervisorctl 超时（10s）")

    out = (res.stdout or "").strip()
    err = (res.stderr or "").strip()
    combined = "\n".join([s for s in [out, err] if s]).strip()
    # supervisorctl 约定：
    # - status 在存在 STOPPED/FATAL 等进程时可能返回 exit=3，但输出仍然有效
    ok_rc = {0}
    if args and args[0] == "status":
        ok_rc.add(3)
    if res.returncode not in ok_rc:
        raise RuntimeError(combined or f"supervisorctl failed (exit={res.returncode})")
    return combined or out


def _parse_supervisor_status(raw: str) -> Tuple[bool, Optional[int], str]:
    """
    典型输出：
    - auto_sys                          RUNNING   pid 1234, uptime 0:10:00
    - auto_sys                          STOPPED   Not started
    """
    if "RUNNING" in raw:
        m = re.search(r"\bpid\s+(\d+)\b", raw)
        pid = int(m.group(1)) if m else None
        return True, pid, "RUNNING"
    for state in ["STOPPED", "FATAL", "EXITED", "BACKOFF", "STARTING", "UNKNOWN"]:
        if state in raw:
            return False, None, state
    return False, None, "UNKNOWN"

def _list_supervisor_process_names(status_all_raw: str) -> list[str]:
    names: list[str] = []
    if not status_all_raw:
        return names
    for ln in status_all_raw.splitlines():
        s = (ln or "").strip()
        if not s:
            continue
        # 每行格式：<name> <STATE> ...
        name = s.split(None, 1)[0].strip()
        if name:
            names.append(name)
    return names


def _get_program_name() -> str:
    # 你给的宝塔配置是 [program:auto_sys]
    return os.getenv("SUPERVISOR_TRADING_PROGRAM", "auto_sys").strip() or "auto_sys"


def _select_best_process_name(program: str, status_all_raw: str) -> Optional[str]:
    """
    从 `supervisorctl status` 全量输出中，找到最匹配的真实进程名。
    兼容 supervisor 的 group:process 格式，例如：auto_sys:auto_sys_00
    """
    if not status_all_raw:
        return None

    lines = [ln.strip() for ln in status_all_raw.splitlines() if ln.strip()]
    names: list[str] = []
    for ln in lines:
        name = ln.split(None, 1)[0].strip()
        if name:
            names.append(name)

    # 精确优先：program / program_00 / program:program_00
    preferred = [program, f"{program}_00", f"{program}:{program}_00"]
    for cand in preferred:
        if cand in names:
            return cand

    # 次优：任意以 program_ 开头
    for name in names:
        if name.startswith(program + "_"):
            return name

    # 次优：任意以 program: 开头
    for name in names:
        if name.startswith(program + ":"):
            return name

    return None


def _status_with_fallback(program: str) -> Tuple[str, Optional[str], Optional[str]]:
    """
    - 优先 `status <program>`
    - 若 no such process：返回全量 status，并尝试解析真实 name（例如 auto_sys:auto_sys_00）
    返回：(raw, resolved_name, status_all)
    """
    try:
        raw = _run_supervisorctl(["status", program])
        return raw, program, None
    except Exception as e:
        msg = str(e).lower()
        if "no such process" not in msg:
            raise

    status_all = _run_supervisorctl(["status"])
    resolved = _select_best_process_name(program, status_all)
    if resolved:
        try:
            raw = _run_supervisorctl(["status", resolved])
            return raw, resolved, status_all
        except Exception:
            # 兜底：至少把全量输出返回，方便你确认真实进程名
            return status_all, None, status_all
    return status_all, None, status_all


def _action_with_fallback(action: str, program: str) -> Tuple[str, Optional[str], Optional[str]]:
    """
    对 start/stop/restart 做兜底：如果 program 不存在，尝试解析真实 name 再执行。
    返回：(output, resolved_name, status_all)
    """
    try:
        out = _run_supervisorctl([action, program])
        return out, program, None
    except Exception as e:
        msg = str(e).lower()
        if "no such process" not in msg:
            raise

    status_all = _run_supervisorctl(["status"])
    resolved = _select_best_process_name(program, status_all)
    if not resolved:
        # 没找到就把全量输出带上，方便定位
        raise RuntimeError(f"no such process: {program}. 当前 supervisor 进程列表：\n{status_all}")

    out = _run_supervisorctl([action, resolved])
    return out, resolved, status_all



def _run_fix_script():
    """Run the fix_trade_records.py script in a subprocess"""
    try:
        script_path = Path(__file__).parent.parent.parent.parent / "scripts" / "fix_trade_records.py"
        if not script_path.exists():
            logger.error(f"Fix script not found at {script_path}")
            return
            
        logger.info(f"Starting trade record fix script: {script_path}")
        
        # Ensure project root is in PYTHONPATH
        env = os.environ.copy()
        project_root = Path(__file__).parent.parent.parent.parent
        env["PYTHONPATH"] = f"{env.get('PYTHONPATH', '')}:{project_root}"
        
        process = subprocess.Popen(
            ["python3", str(script_path)],
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        stdout, stderr = process.communicate()
        
        if process.returncode == 0:
            logger.info(f"Trade record fix completed successfully:\n{stdout}")
        else:
            logger.error(f"Trade record fix failed (exit code {process.returncode}):\n{stderr}")
            
    except Exception as e:
        logger.error(f"Error running trade record fix script: {e}")

@router.post("/fix-trade-records")
async def fix_trade_records(
    background_tasks: BackgroundTasks,
    _admin: Dict[str, Any] = Depends(require_system_admin)
):
    """
    Trigger the trade record fix script (time inversion & commission backfill).
    Runs in background.
    """
    background_tasks.add_task(_run_fix_script)
    return {"message": "Trade fix task started in background"}

@router.post("/clear-cache")
async def clear_cache(
    _admin: Dict[str, Any] = Depends(require_system_admin),
    x_account_id: Optional[int] = Header(default=None, alias="X-Account-Id"),
) -> Dict[str, Any]:
    """
    清理配置缓存（Redis Hash: trading_config），并从数据库回灌到 Redis。
    """
    try:
        import config_manager

        account_id = int(x_account_id or 1)
        cm = None
        if hasattr(config_manager, "ConfigManager") and hasattr(config_manager.ConfigManager, "for_account"):
            cm = config_manager.ConfigManager.for_account(account_id)
        else:
            cm = getattr(config_manager, "config_manager", None)
        if cm is None:
            raise HTTPException(status_code=500, detail="config_manager 未初始化")

        deleted_keys: list[str] = []

        # 1) 清 backend 本地 cache
        try:
            cm._cache = {}
        except Exception:
            pass

        # 2) 清 Redis 缓存 key（Hash: trading_config）
        try:
            redis_client = getattr(cm, "_redis_client", None)
            redis_connected = getattr(cm, "_redis_connected", False)
            if redis_client is not None and redis_connected:
                try:
                    redis_client.ping()
                except Exception:
                    redis_connected = False

            if redis_client is not None and redis_connected:
                try:
                    key = getattr(cm, "_redis_hash_key", "trading_config")
                    redis_client.delete(key)
                    deleted_keys.append(str(key))
                    # 兼容：老 key（仅 default 账号）
                    legacy = getattr(cm, "_legacy_hash_key", None)
                    if legacy and legacy != key:
                        redis_client.delete(legacy)
                        deleted_keys.append(str(legacy))
                except Exception as e:
                    logger.warning(f"删除 Redis key 失败: {e}")

                # 可选：实时推荐缓存（如果存在）
                try:
                    redis_client.delete("recommendations:realtime")
                    deleted_keys.append("recommendations:realtime")
                except Exception:
                    pass
        except Exception as e:
            logger.warning(f"清 Redis 缓存失败: {e}")

        # 3) 立刻从 DB 回灌到 Redis（避免 trading_system 读到空）
        try:
            cm.reload()
        except Exception as e:
            logger.warning(f"回灌配置到 Redis 失败（仍可能使用DB/本地cache）: {e}")

        return {
            "message": "缓存已清理并回灌",
            "deleted_keys": deleted_keys,
            "note": "如果你使用 supervisor 管理交易系统，请点击“重启交易系统”让新 Key 立即生效。",
        }
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"清理缓存失败: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail=str(e))


@router.get("/trading/services")
async def list_trading_services(_admin: Dict[str, Any] = Depends(require_system_admin)):
    """获取所有交易服务状态（包括所有账号）"""
    try:
        # 获取所有 supervisor 进程状态
        status_all = _run_supervisorctl(["status"])
        
        services = []
        summary = {"total": 0, "running": 0, "stopped": 0, "unknown": 0}
        
        # 解析每一行
        # 格式通常是: name state description
        for line in status_all.splitlines():
            line = line.strip()
            if not line:
                continue
            
            parts = line.split(None, 2)
            if len(parts) < 2:
                continue
                
            name = parts[0]
            state = parts[1]
            desc = parts[2] if len(parts) > 2 else ""
            
            # 只关注 auto_sys 开头的服务
            if name.startswith("auto_sys"):
                is_running = state == "RUNNING"
                pid = None
                if is_running:
                    # Parse PID from desc: "pid 1234, uptime ..."
                    m = re.search(r"pid\s+(\d+)", desc)
                    if m:
                        pid = int(m.group(1))
                
                services.append({
                    "program": name,
                    "state": state,
                    "running": is_running,
                    "pid": pid,
                    "description": desc
                })
                
                summary["total"] += 1
                if is_running:
                    summary["running"] += 1
                elif state in ["STOPPED", "EXITED", "FATAL"]:
                    summary["stopped"] += 1
                else:
                    summary["unknown"] += 1
                    
        return {
            "summary": summary,
            "services": services,
            "raw": status_all
        }
    except Exception as e:
        # supervisor 未安装/未运行时（如 unix socket 不存在）避免刷 ERROR，改为 WARNING 并返回友好说明
        err_msg = str(e).strip()
        if not err_msg:
            err_msg = repr(e)
        is_supervisor_unavailable = (
            "no such file" in err_msg.lower()
            or "connection refused" in err_msg.lower()
            or "sock" in err_msg.lower()
            or "unix://" in err_msg.lower()
        )
        if is_supervisor_unavailable:
            logger.warning(f"列出服务失败（supervisor 未运行或不可用）: {err_msg}")
            return {
                "summary": {"total": 0, "running": 0, "stopped": 0, "unknown": 0},
                "services": [],
                "error": "supervisor 未安装或未运行，请检查 supervisord 或配置 SUPERVISOR_CONF"
            }
        logger.error(f"列出服务失败: {e}")
        return {
            "summary": {"total": 0, "running": 0, "stopped": 0, "unknown": 0},
            "services": [],
            "error": err_msg
        }


@router.get("/trading/status")
async def trading_status(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:

    program = _get_program_name()
    try:
        raw, resolved_name, status_all = _status_with_fallback(program)
        running, pid, state = _parse_supervisor_status(raw)
        meta = _system_meta_read("trading:last_restart") or {}
        return {
            "mode": "supervisor",
            "program": program,
            "resolved_name": resolved_name,
            "running": running,
            "pid": pid,
            "state": state,
            "raw": raw,
            "status_all": status_all,
            "meta": meta,
        }
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"supervisorctl status 失败: {e}. 你可能需要配置 SUPERVISOR_CONF / SUPERVISOR_TRADING_PROGRAM / SUPERVISOR_USE_SUDO",
        )


@router.post("/trading/start")
async def trading_start(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:

    program = _get_program_name()
    try:
        out, resolved_name, status_all = _action_with_fallback("start", program)
        raw, resolved_name2, status_all2 = _status_with_fallback(resolved_name or program)
        running, pid, state = _parse_supervisor_status(raw)
        return {
            "message": "交易系统已启动（supervisor）",
            "output": out,
            "status": {
                "mode": "supervisor",
                "program": program,
                "resolved_name": resolved_name2 or resolved_name,
                "running": running,
                "pid": pid,
                "state": state,
                "raw": raw,
                "status_all": status_all2 or status_all,
            },
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"supervisorctl start 失败: {e}")


@router.post("/trading/stop")
async def trading_stop(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:

    program = _get_program_name()
    try:
        out, resolved_name, status_all = _action_with_fallback("stop", program)
        raw, resolved_name2, status_all2 = _status_with_fallback(resolved_name or program)
        running, pid, state = _parse_supervisor_status(raw)
        return {
            "message": "交易系统已停止（supervisor）",
            "output": out,
            "status": {
                "mode": "supervisor",
                "program": program,
                "resolved_name": resolved_name2 or resolved_name,
                "running": running,
                "pid": pid,
                "state": state,
                "raw": raw,
                "status_all": status_all2 or status_all,
            },
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"supervisorctl stop 失败: {e}")


@router.post("/trading/restart")
async def trading_restart(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:

    program = _get_program_name()
    try:
        requested_at = _beijing_time_str()
        requested_at_ms = int(time.time() * 1000)
        out, resolved_name, status_all = _action_with_fallback("restart", program)
        raw, resolved_name2, status_all2 = _status_with_fallback(resolved_name or program)
        running, pid, state = _parse_supervisor_status(raw)

        # 记录交易系统重启时间（用于前端展示）
        _system_meta_write(
            "trading:last_restart",
            {
                "requested_at": requested_at,
                "requested_at_ms": requested_at_ms,
                "pid": pid,
                "program": resolved_name2 or resolved_name or program,
            },
        )
        return {
            "message": "交易系统已重启（supervisor）",
            "output": out,
            "status": {
                "mode": "supervisor",
                "program": program,
                "resolved_name": resolved_name2 or resolved_name,
                "running": running,
                "pid": pid,
                "state": state,
                "raw": raw,
                "status_all": status_all2 or status_all,
            },
            "meta": {
                "requested_at": requested_at,
                "requested_at_ms": requested_at_ms,
            },
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"supervisorctl restart 失败: {e}")


@router.post("/trading/stop-all")
async def trading_stop_all(
    _admin: Dict[str, Any] = Depends(require_system_admin),
    prefix: str = "auto_sys_acc",
    include_default: bool = False,
) -> Dict[str, Any]:
    """
    一键停止所有账号交易进程（supervisor）。
    """
    try:
        prefix = (prefix or "auto_sys_acc").strip()
        if not prefix:
            prefix = "auto_sys_acc"

        # 先读取全量 status，拿到有哪些进程
        status_all = _run_supervisorctl(["status"])
        names = _list_supervisor_process_names(status_all)

        targets: list[str] = []
        for n in names:
            if n.startswith(prefix):
                targets.append(n)

        if include_default:
            default_prog = _get_program_name()
            if default_prog and default_prog not in targets and default_prog in names:
                targets.append(default_prog)

        if not targets:
            return {
                "message": "未找到可停止的交易进程",
                "prefix": prefix,
                "include_default": include_default,
                "count": 0,
                "targets": [],
                "status_all": status_all,
            }

        results: list[Dict[str, Any]] = []
        ok = 0
        failed = 0
        for prog in targets:
            try:
                out = _run_supervisorctl(["stop", prog])
                raw = _run_supervisorctl(["status", prog])
                running, pid, state = _parse_supervisor_status(raw)
                results.append(
                    {
                        "program": prog,
                        "ok": True,
                        "output": out,
                        "status": {"running": running, "pid": pid, "state": state, "raw": raw},
                    }
                )
                ok += 1
            except Exception as e:
                failed += 1
                results.append({"program": prog, "ok": False, "error": str(e)})

        return {
            "message": "已发起批量停止",
            "prefix": prefix,
            "include_default": include_default,
            "count": len(targets),
            "ok": ok,
            "failed": failed,
            "targets": targets,
            "results": results,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"批量停止失败: {e}")


@router.post("/trading/restart-all")
async def trading_restart_all(
    _admin: Dict[str, Any] = Depends(require_system_admin),
    prefix: str = "auto_sys_acc",
    include_default: bool = False,
    do_update: bool = True,
) -> Dict[str, Any]:
    """
    一键重启所有账号交易进程（supervisor）。

    - 默认重启所有以 auto_sys_acc 开头的 program（例如 auto_sys_acc1/2/3...）
    - 可选 include_default=true：同时包含 SUPERVISOR_TRADING_PROGRAM（默认 auto_sys）
    - 可选 do_update=true：先执行 supervisorctl reread/update 再重启（确保新 ini 生效）
    """
    try:
        prefix = (prefix or "auto_sys_acc").strip()
        if not prefix:
            prefix = "auto_sys_acc"

        # 先读取全量 status，拿到有哪些进程
        status_all = _run_supervisorctl(["status"])
        names = _list_supervisor_process_names(status_all)

        targets: list[str] = []
        skipped_disabled: list[Dict[str, Any]] = []
        for n in names:
            if n.startswith(prefix):
                # 若能解析出 account_id，则跳过 disabled 的账号
                try:
                    m = re.match(rf"^{re.escape(prefix)}(\d+)$", n)
                    if m:
                        aid = int(m.group(1))
                        row = Account.get(aid)
                        st = (row.get("status") if isinstance(row, dict) else None) or "active"
                        if str(st).strip().lower() != "active":
                            skipped_disabled.append({"program": n, "account_id": aid, "status": st})
                            continue
                except Exception:
                    # 解析失败/查库失败：不影响批量重启流程
                    pass
                targets.append(n)

        if include_default:
            default_prog = _get_program_name()
            if default_prog and default_prog not in targets and default_prog in names:
                targets.append(default_prog)

        if not targets:
            return {
                "message": "未找到可重启的交易进程",
                "prefix": prefix,
                "include_default": include_default,
                "count": 0,
                "targets": [],
                "status_all": status_all,
                "skipped_disabled": skipped_disabled,
            }

        reread_out = ""
        update_out = ""
        if do_update:
            try:
                reread_out = _run_supervisorctl(["reread"])
            except Exception as e:
                reread_out = f"failed: {e}"
            try:
                update_out = _run_supervisorctl(["update"])
            except Exception as e:
                update_out = f"failed: {e}"

        results: list[Dict[str, Any]] = []
        ok = 0
        failed = 0
        for prog in targets:
            try:
                out = _run_supervisorctl(["restart", prog])
                raw = _run_supervisorctl(["status", prog])
                running, pid, state = _parse_supervisor_status(raw)
                results.append(
                    {
                        "program": prog,
                        "ok": True,
                        "output": out,
                        "status": {"running": running, "pid": pid, "state": state, "raw": raw},
                    }
                )
                ok += 1
            except Exception as e:
                failed += 1
                results.append({"program": prog, "ok": False, "error": str(e)})

        return {
            "message": "已发起批量重启",
            "prefix": prefix,
            "include_default": include_default,
            "do_update": do_update,
            "count": len(targets),
            "ok": ok,
            "failed": failed,
            "reread": reread_out,
            "update": update_out,
            "targets": targets,
            "results": results,
            "skipped_disabled": skipped_disabled,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"批量重启失败: {e}")


@router.get("/market-overview")
async def market_overview(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """
    市场行情概览：拉取 Binance 公开接口，展示与策略过滤对应的数据。
    供全局配置页展示，帮助用户确认当前策略方案是否匹配市场。
    """
    try:
        from market_overview import get_market_overview
    except ImportError:
        try:
            from backend.market_overview import get_market_overview
        except ImportError:
            import sys
            backend_dir = Path(__file__).parent.parent.parent
            if str(backend_dir) not in sys.path:
                sys.path.insert(0, str(backend_dir))
            from market_overview import get_market_overview

    data = get_market_overview()

    # 获取当前策略配置，用于对比
    beta_enabled = False
    beta_threshold = -0.005
    market_scheme = "normal"
    try:
        from config_manager import GlobalStrategyConfigManager
        mgr = GlobalStrategyConfigManager()
        beta_enabled = str(mgr.get("BETA_FILTER_ENABLED", "true")).lower() in ("true", "1", "yes")
        try:
            beta_threshold = float(mgr.get("BETA_FILTER_THRESHOLD", -0.005))
        except (TypeError, ValueError):
            pass
        market_scheme = str(mgr.get("MARKET_SCHEME", "normal")).strip().lower() or "normal"
    except Exception:
        pass

    # 计算大盘共振是否触发（与 strategy._check_beta_filter 一致）
    threshold_pct = beta_threshold * 100
    triggered = False
    if beta_enabled:
        for key in ["btc_15m_change_pct", "btc_1h_change_pct", "eth_15m_change_pct", "eth_1h_change_pct"]:
            val = data.get(key)
            if val is not None and val < threshold_pct:
                triggered = True
                break

    data["config"] = {
        "BETA_FILTER_ENABLED": beta_enabled,
        "BETA_FILTER_THRESHOLD": beta_threshold,
        "BETA_FILTER_THRESHOLD_PCT": round(threshold_pct, 2),
        "MARKET_SCHEME": market_scheme,
    }
    data["beta_filter_triggered"] = triggered

    # 策略执行概览：当前执行方案与配置项执行情况（易读文字）
    get_strategy_execution_overview = None
    try:
        from market_overview import get_strategy_execution_overview
    except ImportError:
        try:
            from backend.market_overview import get_strategy_execution_overview
        except ImportError:
            pass
        if get_strategy_execution_overview is None:
            try:
                import sys
                backend_dir = Path(__file__).resolve().parent.parent.parent
                if str(backend_dir) not in sys.path:
                    sys.path.insert(0, str(backend_dir))
                from market_overview import get_strategy_execution_overview
            except ImportError:
                pass
    if get_strategy_execution_overview is not None:
        try:
            data["strategy_execution_overview"] = get_strategy_execution_overview()
        except Exception as e:
            data["strategy_execution_overview"] = {"sections": [{"title": "加载失败", "content": str(e)}]}
    else:
        data["strategy_execution_overview"] = {
            "sections": [{"title": "策略执行概览暂不可用", "content": "请确认后端已重启并已部署最新代码；若已重启仍无数据，请检查 backend/market_overview.py 与 config_manager 是否可正常导入。"}]
        }
    return data


@router.get("/backend/status")
async def backend_status(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """
    查看后端服务状态（当前 uvicorn 进程）。

    说明：
    - pid 使用 os.getpid()（当前 FastAPI 进程）
    - last_restart 从 Redis 读取（若可用）
    """
    meta = _system_meta_read("backend:last_restart") or {}
    return {
        "running": True,
        "pid": os.getpid(),
        "started_at_ms": _BACKEND_STARTED_AT_MS,
        "started_at": _beijing_time_str(),
        "meta": meta,
    }


@router.post("/backend/restart")
async def backend_restart(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """
    重启后端服务（uvicorn）。

    实现方式：
    - 后端启动脚本为 nohup uvicorn ... &
    - 这里通过后台启动 backend/restart.sh 来完成：
      1) grep 找到 uvicorn api.main:app 进程并 kill
      2) 再执行 backend/start.sh 拉起新进程

    注意：
    - 为了让接口能先返回，这里会延迟 1s 再执行 restart.sh
    """
    backend_dir = Path(__file__).parent.parent.parent  # backend/
    restart_script = backend_dir / "restart.sh"
    if not restart_script.exists():
        raise HTTPException(status_code=500, detail=f"找不到重启脚本: {restart_script}")

    requested_at = _beijing_time_str()
    requested_at_ms = int(time.time() * 1000)
    cur_pid = os.getpid()

    _system_meta_write(
        "backend:last_restart",
        {
            "requested_at": requested_at,
            "requested_at_ms": requested_at_ms,
            "pid_before": cur_pid,
            "script": str(restart_script),
        },
    )

    # 后台执行：sleep 1 后再重启，保证当前请求可以返回
    cmd = ["bash", "-lc", f"sleep 1; '{restart_script}'"]
    try:
        subprocess.Popen(
            cmd,
            cwd=str(backend_dir),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"启动重启脚本失败: {e}")

    return {
        "message": "已发起后端重启（1s 后执行）",
        "pid_before": cur_pid,
        "requested_at": requested_at,
        "requested_at_ms": requested_at_ms,
        "script": str(restart_script),
        "note": "重启期间接口可能短暂不可用，页面可等待 3-5 秒后刷新状态。",
    }


@router.post("/backend/stop")
async def backend_stop(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """
    停止后端服务（uvicorn）。
    警告：停止后 API 将不可用，必须手动登录服务器启动！
    """
    backend_dir = Path(__file__).parent.parent.parent  # backend/
    stop_script = backend_dir / "stop.sh"
    if not stop_script.exists():
        raise HTTPException(status_code=500, detail=f"找不到停止脚本: {stop_script}")

    cur_pid = os.getpid()
    
    # 后台执行：sleep 1 后再停止，保证当前请求可以返回
    cmd = ["bash", "-lc", f"sleep 1; '{stop_script}'"]
    try:
        subprocess.Popen(
            cmd,
            cwd=str(backend_dir),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"启动停止脚本失败: {e}")

    return {
        "message": "已发起后端停止（1s 后执行）",
        "pid_before": cur_pid,
        "script": str(stop_script),
        "warning": "后端服务停止后，Web 界面将无法访问，请手动在服务器启动！",
    }


def _recommendations_process_running() -> Tuple[bool, Optional[int]]:
    """检查推荐服务进程是否运行，返回 (running, pid)。兼容 pgrep/ps 及 supervisor 等启动方式。"""
    # 1. 优先 pgrep（Linux/macOS 常见）
    for pattern in ["trading_system.recommendations_main", "recommendations_main1", "recommendations_main"]:
        try:
            result = subprocess.run(
                ["pgrep", "-f", pattern],
                capture_output=True,
                text=True,
                timeout=5,
            )
            if result.returncode == 0 and result.stdout.strip():
                pids = [x for x in result.stdout.strip().split() if x.isdigit()]
                if pids:
                    return True, int(pids[0])
        except (FileNotFoundError, subprocess.TimeoutExpired, ValueError):
            break

    # 2. 回退：ps + grep（pgrep 不可用或匹配失败时）
    try:
        result = subprocess.run(
            ["sh", "-c", "ps aux 2>/dev/null | grep -E 'recommendations_main1|recommendations_main|trading_system.recommendations' | grep -v grep | head -1"],
            capture_output=True,
            text=True,
            timeout=5,
        )
        if result.returncode == 0 and result.stdout.strip():
            parts = result.stdout.strip().split()
            if len(parts) >= 2 and parts[1].isdigit():
                return True, int(parts[1])
    except (FileNotFoundError, subprocess.TimeoutExpired, ValueError):
        pass
    return False, None


@router.get("/recommendations/status")
async def recommendations_status(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """查看推荐服务状态（recommendations_main 进程）"""
    running, pid = _recommendations_process_running()
    return {
        "running": running,
        "pid": pid,
    }


@router.post("/recommendations/restart")
async def recommendations_restart(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """重启推荐服务（recommendations_main）"""
    backend_dir = Path(__file__).parent.parent.parent
    restart_script = backend_dir / "restart_recommendations.sh"
    if not restart_script.exists():
        raise HTTPException(status_code=500, detail=f"找不到重启脚本: {restart_script}")
    running_before, pid_before = _recommendations_process_running()
    cmd = ["bash", "-lc", f"sleep 1; '{restart_script}'"]
    try:
        subprocess.Popen(
            cmd,
            cwd=str(backend_dir),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"启动重启脚本失败: {e}")
    return {
        "message": "已发起推荐服务重启（1s 后执行）",
        "pid_before": pid_before,
        "running_before": running_before,
        "script": str(restart_script),
    }


@router.post("/recommendations/stop")
async def recommendations_stop(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """停止推荐服务"""
    backend_dir = Path(__file__).parent.parent.parent
    stop_script = backend_dir / "stop_recommendations.sh"
    if not stop_script.exists():
        raise HTTPException(status_code=500, detail=f"找不到停止脚本: {stop_script}")
    running_before, pid_before = _recommendations_process_running()
    cmd = ["bash", "-lc", f"sleep 1; '{stop_script}'"]
    try:
        subprocess.Popen(
            cmd,
            cwd=str(backend_dir),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"启动停止脚本失败: {e}")
    return {
        "message": "已发起推荐服务停止",
        "pid_before": pid_before,
        "running_before": running_before,
    }


@router.post("/recommendations/start")
async def recommendations_start(_admin: Dict[str, Any] = Depends(require_system_admin)) -> Dict[str, Any]:
    """启动推荐服务（若已运行则跳过）"""
    running, pid = _recommendations_process_running()
    if running:
        return {"message": "推荐服务已在运行中", "pid": pid, "skipped": True}
    backend_dir = Path(__file__).parent.parent.parent
    start_script = backend_dir / "start_recommendations.sh"
    if not start_script.exists():
        raise HTTPException(status_code=500, detail=f"找不到启动脚本: {start_script}")
    cmd = ["bash", "-lc", f"sleep 1; '{start_script}'"]
    try:
        subprocess.Popen(
            cmd,
            cwd=str(backend_dir),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"启动脚本执行失败: {e}")
    return {"message": "已发起推荐服务启动（1s 后执行）", "script": str(start_script)}

