app.py

import asyncio
import atexit
import hashlib
import json
import logging
import os
import secrets
import threading
import traceback
import time
from collections import deque
from datetime import datetime
from pathlib import Path
from typing import Any, Optional
from urllib.parse import parse_qsl, unquote, urlsplit

import pymysql
import uvicorn
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile, status
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel, Field

from core.tasks import runTasks
from utils.logger import setup_logger


logger = setup_logger(level=logging.DEBUG)

BASE_DIR = Path(__file__).resolve().parent
TEMPLATES_DIR = BASE_DIR / "templates"
STATIC_DIR = BASE_DIR / "static"
ROOT_CONFIG_PATH = BASE_DIR / "config.json"
LEGACY_DATA_DIR = BASE_DIR / "data"
LEGACY_USERS_META_PATH = LEGACY_DATA_DIR / "users.json"
MYSQL_DSN_TEMPLATE = "mysql://SQL_PASSWORD@mysql-2bace9cd-cacode.i.aivencloud.com:21260/defaultdb?ssl-mode=REQUIRED"
MYSQL_DSN_ENV = "MYSQL_DSN_TEMPLATE"
MYSQL_PASSWORD_ENV = "SQL_PASSWORD"
MYSQL_USER_ENV = "MYSQL_USER"
MYSQL_CA_CERT_ENV = "MYSQL_CA_CERT_PATH"
MYSQL_DEFAULT_USER = "avnadmin"
USERS_TABLE = "app_users"
SESSION_COOKIE_NAME = "sparkflow_auth"
DEFAULT_TIMEZONE = "Asia/Shanghai"
MAX_LOG_LINES = 1200
MAX_TEMPLATE_LENGTH = 2000
PASSWORD_ITERATIONS = 210000
FAILED_RETRY_JOB_ID = "_system_retry_failed_tasks"
FAILED_RETRY_INTERVAL_HOURS = 1

DEFAULT_USER_CONFIG = {
    "multiTask": True,
    "taskCount": 5,
    "proxyAddress": "",
    "messageTemplate": "[续火花]",
    "hitokotoTypes": ["文学", "影视", "诗词", "哲学"],
    "scheduler": {
        "enabled": True,
        "timezone": DEFAULT_TIMEZONE,
        "hour": 9,
        "minute": 0,
        "runOnStartup": False,
    },
}

AUTH_SESSIONS: dict[str, dict[str, str]] = {}
db_init_lock = threading.Lock()
scheduler_lock = threading.Lock()
runtime_map_lock = threading.Lock()
db_initialized = False
db_status_lock = threading.Lock()
db_status = {
    "connected": None,
    "last_checked_at": None,
    "last_ok_at": None,
    "last_error": "",
}
scheduler_bootstrapped = False
scheduler_bootstrap_lock = threading.Lock()
scheduler_bootstrap_running = False


class UserRuntimeState:
    def __init__(self, username: str):
        self.username = username
        self._run_lock = threading.Lock()
        self._state_lock = threading.Lock()
        self.is_running = False
        self.last_status = "未开始"
        self.last_error = ""
        self.last_trigger = "-"
        self.last_start = None
        self.last_end = None
        self.next_run = None
        self.schedule_hour = 9
        self.schedule_minute = 0
        self.schedule_timezone = DEFAULT_TIMEZONE
        self.history = deque(maxlen=50)
        self.logs = deque(maxlen=2000)

    def _format_ts(self, value: Optional[datetime]):
        if not value:
            return "-"
        return value.strftime("%Y-%m-%d %H:%M:%S")

    def schedule_time(self):
        return f"{self.schedule_hour:02d}:{self.schedule_minute:02d}"

    def _set_running(self, value: bool):
        with self._state_lock:
            self.is_running = value

    def add_log(self, message: str):
        ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with self._state_lock:
            self.logs.append(f"{ts} [{self.username}] {message}")

    def update_schedule(self, hour: int, minute: int, timezone: str):
        with self._state_lock:
            self.schedule_hour = hour
            self.schedule_minute = minute
            self.schedule_timezone = timezone

    def update_next_run(self, next_run):
        with self._state_lock:
            self.next_run = next_run

    def snapshot(self, account_count: int, target_count: int):
        with self._state_lock:
            return {
                "is_running": self.is_running,
                "last_status": self.last_status,
                "last_error": self.last_error,
                "last_trigger": self.last_trigger,
                "last_start": self._format_ts(self.last_start),
                "last_end": self._format_ts(self.last_end),
                "next_run": self._format_ts(self.next_run),
                "account_count": account_count,
                "target_count": target_count,
                "schedule_time": self.schedule_time(),
                "schedule_timezone": self.schedule_timezone,
            }

    def history_rows(self):
        with self._state_lock:
            return list(self.history)[::-1]

    def recent_logs(self, limit=MAX_LOG_LINES):
        with self._state_lock:
            lines = list(self.logs)[-max(1, limit):]
        return "\n".join(lines) if lines else "暂无日志。"

    def run_once(self, trigger: str):
        if not self._run_lock.acquire(blocking=False):
            self.add_log(f"任务已在运行中，忽略触发：{trigger}")
            return False, "已有任务在运行，本次触发已跳过。"

        self._set_running(True)
        with self._state_lock:
            self.last_trigger = trigger
            self.last_start = datetime.now()
            self.last_end = None
            self.last_error = ""
            self.last_status = "运行中"
        self.add_log(f"任务开始执行，触发方式：{trigger}")

        ok = True
        message = "任务执行完成。"
        try:
            asyncio.run(_run_user_tasks(self.username))
            with self._state_lock:
                self.last_status = "成功"
        except Exception as exc:
            ok = False
            message = f"任务执行失败：{exc}"
            with self._state_lock:
                self.last_status = "失败"
                self.last_error = repr(exc)
            self.add_log(f"任务失败：{exc}")
            logger.error("Task failed. user=%s trigger=%s error=%s", self.username, trigger, exc)
            logger.debug("Task traceback:\n%s", traceback.format_exc())
        finally:
            end_at = datetime.now()
            with self._state_lock:
                self.last_end = end_at
                duration = (self.last_end - self.last_start).total_seconds()
                self.history.append(
                    {
                        "trigger": trigger,
                        "start": self._format_ts(self.last_start),
                        "end": self._format_ts(self.last_end),
                        "status": self.last_status,
                        "duration": f"{duration:.2f}s",
                        "message": self.last_error or "OK",
                    }
                )
                current_status = self.last_status
            self.add_log(f"任务结束，状态={current_status}，耗时={duration:.2f}s")
            self._set_running(False)
            self._run_lock.release()
        return ok, message


runtime_map: dict[str, UserRuntimeState] = {}
scheduler = None


class UserLoginPayload(BaseModel):
    username: str
    password: str


class AdminLoginPayload(BaseModel):
    password: str


class SchedulePayload(BaseModel):
    time: str


class MessageTemplatePayload(BaseModel):
    message: str


class UserTargetsItem(BaseModel):
    unique_id: str
    targets: list[str] = Field(default_factory=list)


class UserTargetsPayload(BaseModel):
    users: list[UserTargetsItem]


def _ensure_data_layout():
    global db_initialized
    if db_initialized:
        return

    logger.info("DB layout ensure begin.")
    with db_init_lock:
        if db_initialized:
            logger.info("DB layout already initialized by another worker.")
            return
        logger.info("DB schema initialization begin.")
        _init_db_schema()
        db_initialized = True
        logger.info("DB schema initialization complete.")

    try:
        logger.info("Legacy migration stage begin.")
        _migrate_legacy_file_data_if_needed()
        logger.info("Legacy migration stage complete.")
    except Exception as exc:
        logger.warning("Legacy data migration skipped due to error: %s", exc)


def _hash_password(password: str, salt_hex: Optional[str] = None):
    salt = bytes.fromhex(salt_hex) if salt_hex else secrets.token_bytes(16)
    digest = hashlib.pbkdf2_hmac(
        "sha256",
        password.encode("utf-8"),
        salt,
        PASSWORD_ITERATIONS,
    )
    return {
        "salt": salt.hex(),
        "hash": digest.hex(),
    }


def _verify_password(password: str, salt_hex: str, expected_hash: str):
    data = _hash_password(password, salt_hex=salt_hex)
    return secrets.compare_digest(data["hash"], expected_hash)


def _deep_copy_json(value):
    return json.loads(json.dumps(value, ensure_ascii=False))


def _merge_config_with_defaults(raw_cfg: Any):
    base = _deep_copy_json(DEFAULT_USER_CONFIG)
    if not isinstance(raw_cfg, dict):
        return base

    merged = _deep_copy_json(base)
    merged.update(raw_cfg)
    base_scheduler = base.get("scheduler", {})
    merged_scheduler = raw_cfg.get("scheduler", {})
    if isinstance(merged_scheduler, dict):
        scheduler = _deep_copy_json(base_scheduler)
        scheduler.update(merged_scheduler)
        merged["scheduler"] = scheduler
    else:
        merged["scheduler"] = _deep_copy_json(base_scheduler)
    return merged


def _format_common_ts(value: Optional[datetime]):
    if not value:
        return "-"
    return value.strftime("%Y-%m-%d %H:%M:%S")


def _update_db_status(connected: bool, error: Optional[Exception] = None):
    now = datetime.now()
    with db_status_lock:
        db_status["connected"] = connected
        db_status["last_checked_at"] = now
        if connected:
            db_status["last_ok_at"] = now
            db_status["last_error"] = ""
        else:
            db_status["last_error"] = str(error or "数据库连接失败")


def _build_db_status_payload():
    with db_status_lock:
        connected = db_status.get("connected")
        return {
            "connected": connected,
            "last_checked_at": _format_common_ts(db_status.get("last_checked_at")),
            "last_ok_at": _format_common_ts(db_status.get("last_ok_at")),
            "last_error": str(db_status.get("last_error") or ""),
        }


def _resolve_mysql_dsn():
    raw = os.getenv(MYSQL_DSN_ENV, MYSQL_DSN_TEMPLATE).strip()
    if "SQL_PASSWORD" in raw:
        secret = os.getenv(MYSQL_PASSWORD_ENV, "").strip()
        if not secret:
            raise RuntimeError(f"环境变量 {MYSQL_PASSWORD_ENV} 未设置，无法连接 MySQL。")
        raw = raw.replace("SQL_PASSWORD", secret, 1)
    return raw


def _build_mysql_conn_kwargs():
    dsn = _resolve_mysql_dsn()
    parsed = urlsplit(dsn)
    if parsed.scheme not in ("mysql", "mysql+pymysql"):
        raise RuntimeError(f"不支持的 MySQL DSN 协议：{parsed.scheme}")

    host = parsed.hostname
    if not host:
        raise RuntimeError("MySQL DSN 缺少主机地址。")

    user = unquote(parsed.username or "")
    password = unquote(parsed.password) if parsed.password is not None else None
    if user and password is None:
        password = user
        user = os.getenv(MYSQL_USER_ENV, MYSQL_DEFAULT_USER).strip() or MYSQL_DEFAULT_USER
    if not user:
        user = os.getenv(MYSQL_USER_ENV, MYSQL_DEFAULT_USER).strip() or MYSQL_DEFAULT_USER
    if not password:
        password = os.getenv(MYSQL_PASSWORD_ENV, "").strip()
    if not password:
        raise RuntimeError("MySQL 密码为空，请检查 SQL_PASSWORD 环境变量。")

    db_name = parsed.path.lstrip("/") or "defaultdb"
    query = {k.lower(): v for k, v in parse_qsl(parsed.query, keep_blank_values=True)}
    ssl_mode = str(query.get("ssl-mode", query.get("ssl_mode", ""))).upper()

    kwargs = {
        "host": host,
        "port": parsed.port or 3306,
        "user": user,
        "password": password,
        "database": db_name,
        "charset": "utf8mb4",
        "autocommit": True,
        "connect_timeout": int(os.getenv("MYSQL_CONNECT_TIMEOUT", "4")),
        "read_timeout": int(os.getenv("MYSQL_READ_TIMEOUT", "8")),
        "write_timeout": int(os.getenv("MYSQL_WRITE_TIMEOUT", "8")),
        "cursorclass": pymysql.cursors.DictCursor,
    }

    if ssl_mode in {"REQUIRED", "VERIFY_CA", "VERIFY_IDENTITY"}:
        ca_file = Path(os.getenv(MYSQL_CA_CERT_ENV, str(BASE_DIR / "camysql2.pem"))).resolve()
        if not ca_file.exists():
            raise RuntimeError(f"MySQL CA 证书不存在：{ca_file}")
        kwargs["ssl"] = {"ca": str(ca_file)}

    return kwargs


def _db_connect():
    kwargs = _build_mysql_conn_kwargs()
    host = kwargs.get("host")
    port = kwargs.get("port")
    database = kwargs.get("database")
    connect_timeout = kwargs.get("connect_timeout")
    read_timeout = kwargs.get("read_timeout")
    write_timeout = kwargs.get("write_timeout")
    has_ssl = bool(kwargs.get("ssl"))
    started_at = time.perf_counter()

    logger.info(
        "MySQL connect begin. host=%s port=%s db=%s connect_timeout=%ss read_timeout=%ss write_timeout=%ss ssl=%s",
        host,
        port,
        database,
        connect_timeout,
        read_timeout,
        write_timeout,
        has_ssl,
    )
    try:
        conn = pymysql.connect(**kwargs)
    except Exception as exc:
        elapsed = time.perf_counter() - started_at
        logger.warning(
            "MySQL connect failed after %.2fs. host=%s port=%s db=%s error=%s",
            elapsed,
            host,
            port,
            database,
            exc,
        )
        _update_db_status(False, exc)
        raise

    elapsed = time.perf_counter() - started_at
    logger.info("MySQL connect success. host=%s port=%s db=%s elapsed=%.2fs", host, port, database, elapsed)
    _update_db_status(True)
    return conn


def _db_query_all(query: str, params=()):
    conn = _db_connect()
    try:
        with conn.cursor() as cursor:
            cursor.execute(query, params)
            return cursor.fetchall()
    finally:
        conn.close()


def _db_query_one(query: str, params=()):
    conn = _db_connect()
    try:
        with conn.cursor() as cursor:
            cursor.execute(query, params)
            return cursor.fetchone()
    finally:
        conn.close()


def _db_execute(query: str, params=()):
    conn = _db_connect()
    try:
        with conn.cursor() as cursor:
            cursor.execute(query, params)
            return cursor.rowcount
    finally:
        conn.close()


def _init_db_schema():
    logger.info("DB schema creation SQL begin.")
    _db_execute(
        f"""
        CREATE TABLE IF NOT EXISTS `{USERS_TABLE}` (
            `username` VARCHAR(128) NOT NULL,
            `unique_id` VARCHAR(255) NOT NULL,
            `password_hash` VARCHAR(128) NOT NULL,
            `password_salt` VARCHAR(64) NOT NULL,
            `created_at` VARCHAR(32) NOT NULL,
            `config_json` LONGTEXT NOT NULL,
            `users_data_json` LONGTEXT NOT NULL,
            `updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
            PRIMARY KEY (`username`),
            UNIQUE KEY `uniq_unique_id` (`unique_id`)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
        """
    )
    logger.info("DB schema creation SQL complete.")


def _legacy_load_json(path: Path, default):
    if not path.exists():
        return default
    with path.open("r", encoding="utf-8") as f:
        return json.load(f)


def _migrate_legacy_file_data_if_needed():
    logger.info("Legacy migration check. path=%s", LEGACY_USERS_META_PATH)
    if not LEGACY_USERS_META_PATH.exists():
        logger.info("Legacy migration skipped: users.json not found.")
        return

    row = _db_query_one(f"SELECT COUNT(*) AS cnt FROM `{USERS_TABLE}`")
    existing_count = int(row.get("cnt", 0)) if row else 0
    if existing_count > 0:
        logger.info("Legacy migration skipped: database already has %s users.", existing_count)
        return

    try:
        raw = _legacy_load_json(LEGACY_USERS_META_PATH, {"users": []})
    except Exception as exc:
        logger.warning("读取旧版 users.json 失败，跳过迁移：%s", exc)
        return

    users = raw.get("users", []) if isinstance(raw, dict) else []
    if not users:
        logger.info("Legacy migration skipped: legacy users list is empty.")
        return

    logger.info("Legacy migration loaded %s legacy users.", len(users))
    migrated = 0
    for item in users:
        username = str(item.get("username", "")).strip()
        unique_id = str(item.get("unique_id", "")).strip()
        password_hash = str(item.get("password_hash", "")).strip()
        password_salt = str(item.get("password_salt", "")).strip()
        created_at = str(item.get("created_at", "")).strip() or datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        if not (username and unique_id and password_hash and password_salt):
            logger.warning("旧版用户数据不完整，跳过：%s", username or "<empty>")
            continue

        cfg = _get_default_user_config()
        users_data = []
        tenant_rel = str(item.get("tenant_dir", "")).strip()
        if tenant_rel:
            tenant_dir = (BASE_DIR / tenant_rel).resolve()
            cfg_path = tenant_dir / "config.json"
            users_data_path = tenant_dir / "usersData.json"
            try:
                cfg = _merge_config_with_defaults(_legacy_load_json(cfg_path, cfg))
            except Exception as exc:
                logger.warning("读取旧版配置失败，使用默认配置。user=%s error=%s", username, exc)
            try:
                users_data = _legacy_load_json(users_data_path, [])
            except Exception as exc:
                logger.warning("读取旧版 usersData 失败。user=%s error=%s", username, exc)

        if not isinstance(users_data, list):
            users_data = []

        try:
            _create_user_record(
                username=username,
                unique_id=unique_id,
                password_hash=password_hash,
                password_salt=password_salt,
                created_at=created_at,
                config_payload=cfg,
                users_data_payload=users_data,
            )
            migrated += 1
        except Exception as exc:
            logger.warning("迁移用户失败。user=%s error=%s", username, exc)

    logger.info("Legacy migration finished. migrated=%s total=%s", migrated, len(users))
    if migrated > 0:
        logger.info("已完成旧版文件数据迁移，共迁移 %s 个用户。", migrated)


def _load_users_meta():
    logger.info("Load users meta begin.")
    _ensure_data_layout()
    rows = _db_query_all(
        f"""
        SELECT username, unique_id, password_hash, password_salt, created_at
        FROM `{USERS_TABLE}`
        ORDER BY username ASC
        """
    )
    logger.info("Load users meta complete. count=%s", len(rows))
    return {str(row["username"]): row for row in rows}


def _load_user_row(username: str):
    _ensure_data_layout()
    return _db_query_one(
        f"""
        SELECT username, unique_id, password_hash, password_salt, created_at, config_json, users_data_json
        FROM `{USERS_TABLE}`
        WHERE username=%s
        """,
        (username,),
    )


def _user_exists(username: str):
    _ensure_data_layout()
    row = _db_query_one(
        f"SELECT 1 AS ok FROM `{USERS_TABLE}` WHERE username=%s",
        (username,),
    )
    return bool(row)


def _create_user_record(
    *,
    username: str,
    unique_id: str,
    password_hash: str,
    password_salt: str,
    created_at: str,
    config_payload: dict[str, Any],
    users_data_payload: list[dict[str, Any]],
):
    _db_execute(
        f"""
        INSERT INTO `{USERS_TABLE}`
        (username, unique_id, password_hash, password_salt, created_at, config_json, users_data_json)
        VALUES (%s, %s, %s, %s, %s, %s, %s)
        """,
        (
            username,
            unique_id,
            password_hash,
            password_salt,
            created_at,
            json.dumps(config_payload, ensure_ascii=False),
            json.dumps(users_data_payload, ensure_ascii=False),
        ),
    )


def _delete_user_record(username: str):
    _ensure_data_layout()
    return _db_execute(f"DELETE FROM `{USERS_TABLE}` WHERE username=%s", (username,))


def _get_user_meta_or_404(username: str):
    users_map = _load_users_meta()
    user = users_map.get(username)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user


def _get_default_user_config():
    if ROOT_CONFIG_PATH.exists():
        try:
            with ROOT_CONFIG_PATH.open("r", encoding="utf-8") as f:
                root_cfg = json.load(f)
            return _merge_config_with_defaults(root_cfg)
        except Exception:
            logger.warning("Failed to read root config.json. fallback to DEFAULT_USER_CONFIG")
    return _deep_copy_json(DEFAULT_USER_CONFIG)


def _load_user_config(username: str):
    row = _load_user_row(username)
    if not row:
        raise FileNotFoundError(f"用户 {username} 不存在")
    try:
        payload = json.loads(row.get("config_json", "{}"))
    except Exception as exc:
        raise ValueError(f"用户 {username} 的配置数据损坏：{exc}")
    return _merge_config_with_defaults(payload)


def _save_user_config(username: str, cfg: dict):
    normalized = _merge_config_with_defaults(cfg)
    changed = _db_execute(
        f"UPDATE `{USERS_TABLE}` SET config_json=%s WHERE username=%s",
        (json.dumps(normalized, ensure_ascii=False), username),
    )
    if changed == 0 and not _user_exists(username):
        raise FileNotFoundError(f"用户 {username} 不存在")


def _load_user_users_data(username: str):
    row = _load_user_row(username)
    if not row:
        raise FileNotFoundError(f"用户 {username} 不存在")
    try:
        data = json.loads(row.get("users_data_json", "[]"))
    except Exception as exc:
        raise ValueError(f"用户 {username} 的 usersData 数据损坏：{exc}")
    if not isinstance(data, list):
        raise ValueError("usersData.json 必须是数组")
    return data


def _save_user_users_data(username: str, users_data: list):
    changed = _db_execute(
        f"UPDATE `{USERS_TABLE}` SET users_data_json=%s WHERE username=%s",
        (json.dumps(users_data, ensure_ascii=False), username),
    )
    if changed == 0 and not _user_exists(username):
        raise FileNotFoundError(f"用户 {username} 不存在")


def _sanitize_targets(values):
    cleaned = []
    seen = set()
    for value in values or []:
        text = str(value).strip()
        if not text or text in seen:
            continue
        seen.add(text)
        cleaned.append(text)
    return cleaned


def _validate_and_normalize_users_data(raw_bytes: bytes):
    try:
        payload = json.loads(raw_bytes.decode("utf-8"))
    except Exception as exc:
        raise ValueError(f"上传文件不是合法 JSON：{exc}")

    if not isinstance(payload, list) or not payload:
        raise ValueError("usersData.json 必须是非空数组")

    normalized = []
    for idx, item in enumerate(payload):
        if not isinstance(item, dict):
            raise ValueError(f"第 {idx + 1} 条用户数据格式错误（必须是对象）")

        unique_id = str(item.get("unique_id", "")).strip()
        username = str(item.get("username", "")).strip()
        cookies = item.get("cookies", [])
        targets = item.get("targets", [])

        if not unique_id:
            raise ValueError(f"第 {idx + 1} 条缺少 unique_id")
        if not username:
            raise ValueError(f"第 {idx + 1} 条缺少 username")
        if not isinstance(cookies, list) or not cookies:
            raise ValueError(f"第 {idx + 1} 条 cookies 不能为空且必须是数组")
        if not isinstance(targets, list):
            raise ValueError(f"第 {idx + 1} 条 targets 必须是数组")

        normalized.append(
            {
                "unique_id": unique_id,
                "username": username,
                "cookies": cookies,
                "targets": _sanitize_targets(targets),
            }
        )

    primary_username = normalized[0]["username"]
    primary_unique_id = normalized[0]["unique_id"]
    return normalized, primary_username, primary_unique_id


def _count_targets(users_data: list):
    return sum(len(user.get("targets", [])) for user in users_data)


def _get_runtime(username: str):
    with runtime_map_lock:
        runtime = runtime_map.get(username)
        if runtime is None:
            runtime = UserRuntimeState(username=username)
            runtime_map[username] = runtime
        return runtime


def _delete_runtime(username: str):
    with runtime_map_lock:
        runtime_map.pop(username, None)


def _session_from_request(request: Request):
    token = request.cookies.get(SESSION_COOKIE_NAME)
    if not token:
        return None
    return AUTH_SESSIONS.get(token)


def _require_user_session(request: Request):
    session = _session_from_request(request)
    if not session or session.get("role") != "user":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="未登录或登录已失效",
        )
    return session


def _require_admin_session(request: Request):
    session = _session_from_request(request)
    if not session or session.get("role") != "admin":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="未登录或登录已失效",
        )
    return session


def _parse_time_string(value: str):
    parts = value.strip().split(":")
    if len(parts) not in (2, 3):
        raise ValueError("时间格式错误，必须是 HH:MM")
    hour = int(parts[0])
    minute = int(parts[1])
    if hour < 0 or hour > 23 or minute < 0 or minute > 59:
        raise ValueError("时间范围错误，小时 0-23，分钟 0-59")
    return hour, minute


def _build_editor_state(username: str):
    cfg = _load_user_config(username)
    users = _load_user_users_data(username)
    return {
        "message_template": str(cfg.get("messageTemplate", "")),
        "users": [
            {
                "unique_id": str(user.get("unique_id", "")),
                "username": str(user.get("username", "未知用户")),
                "targets": _sanitize_targets(user.get("targets", [])),
            }
            for user in users
        ],
    }


def _scheduler_job_id(username: str):
    return f"daily_task::{username}"


def _run_scheduled_once(username: str):
    runtime = _get_runtime(username)
    runtime.run_once("schedule")
    if scheduler:
        job = scheduler.get_job(_scheduler_job_id(username))
        runtime.update_next_run(job.next_run_time if job else None)


async def _run_user_tasks(username: str):
    cfg = _load_user_config(username)
    users_data = _load_user_users_data(username)
    await runTasks(config=cfg, userData=users_data)


def _sync_user_jobs_from_meta(users_map: dict[str, Any], run_startup_tasks: bool = False):
    global scheduler_bootstrapped

    logger.info("Sync user jobs begin. count=%s run_startup_tasks=%s", len(users_map), run_startup_tasks)
    for username in users_map.keys():
        logger.info("Sync user job. username=%s", username)
        _schedule_user_job(username)
        if run_startup_tasks:
            cfg = _load_user_config(username)
            run_on_startup = bool(cfg.get("scheduler", {}).get("runOnStartup", False))
            logger.info("Startup run flag loaded. username=%s run_on_startup=%s", username, run_on_startup)
            if run_on_startup:
                logger.info("Trigger startup run. username=%s", username)
                _start_background_run(username, "startup")

    scheduler_bootstrapped = True
    logger.info("Sync user jobs complete. count=%s", len(users_map))


def _start_scheduler_bootstrap(run_startup_tasks: bool):
    global scheduler_bootstrapped, scheduler_bootstrap_running

    with scheduler_bootstrap_lock:
        if scheduler_bootstrap_running:
            logger.info("Scheduler bootstrap already running; skip duplicate start.")
            return False
        scheduler_bootstrap_running = True

    def _worker():
        global scheduler_bootstrapped, scheduler_bootstrap_running
        try:
            logger.info("Scheduler bootstrap started. run_startup_tasks=%s", run_startup_tasks)

            logger.info("Bootstrap stage begin: ensure_data_layout")
            _ensure_data_layout()
            logger.info("Bootstrap stage complete: ensure_data_layout")

            logger.info("Bootstrap stage begin: load_users_meta")
            users_map = _load_users_meta()
            logger.info("Bootstrap stage complete: load_users_meta count=%s", len(users_map))

            logger.info("Bootstrap stage begin: sync_user_jobs")
            _sync_user_jobs_from_meta(users_map, run_startup_tasks=run_startup_tasks)
            logger.info("Bootstrap stage complete: sync_user_jobs count=%s", len(users_map))

            logger.info("Scheduler bootstrap completed. users=%s", len(users_map))
        except Exception as exc:
            scheduler_bootstrapped = False
            logger.warning("Scheduler bootstrap skipped, database unavailable. error=%s", exc)
        finally:
            with scheduler_bootstrap_lock:
                scheduler_bootstrap_running = False

    thread = threading.Thread(target=_worker, daemon=True, name="scheduler-bootstrap")
    thread.start()
    return True


def _retry_failed_tasks_once(trigger: str, *, raise_on_db_error: bool = False):
    try:
        users_map = _load_users_meta()
        if not scheduler_bootstrapped:
            _sync_user_jobs_from_meta(users_map, run_startup_tasks=False)
    except Exception as exc:
        logger.warning("Failed to load users for failed-task retry. error=%s", exc)
        if raise_on_db_error:
            raise
        return []

    triggered = []
    for username in users_map.keys():
        runtime = _get_runtime(username)
        snapshot = runtime.snapshot(account_count=0, target_count=0)
        if snapshot.get("is_running") or snapshot.get("last_status") != "失败":
            continue

        try:
            cfg = _load_user_config(username)
        except Exception as exc:
            runtime.add_log(f"自动重试前加载配置失败：{exc}")
            continue

        if not bool(cfg.get("scheduler", {}).get("enabled", True)):
            continue

        runtime.add_log(f"检测到失败任务，准备执行自动重试：{trigger}")
        _start_background_run(username, trigger)
        triggered.append(username)

    if triggered:
        logger.info("Retried failed tasks for users: %s", ", ".join(triggered))
    return triggered


def _retry_failed_tasks_job():
    _retry_failed_tasks_once("hourly_retry")


def _schedule_user_job(username: str):
    global scheduler

    cfg = _load_user_config(username)
    scheduler_cfg = cfg.get("scheduler", {}) if isinstance(cfg, dict) else {}
    enabled = bool(scheduler_cfg.get("enabled", True))
    timezone = str(scheduler_cfg.get("timezone", DEFAULT_TIMEZONE))
    hour = int(scheduler_cfg.get("hour", 9))
    minute = int(scheduler_cfg.get("minute", 0))

    runtime = _get_runtime(username)
    runtime.update_schedule(hour, minute, timezone)

    with scheduler_lock:
        if scheduler is None:
            scheduler = BackgroundScheduler(timezone=timezone)
            scheduler.start()

        job_id = _scheduler_job_id(username)
        if not enabled:
            if scheduler.get_job(job_id):
                scheduler.remove_job(job_id)
            runtime.update_next_run(None)
            runtime.add_log("定时任务已禁用")
            return

        scheduler.add_job(
            _run_scheduled_once,
            args=[username],
            trigger=CronTrigger(hour=hour, minute=minute, timezone=timezone),
            id=job_id,
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        job = scheduler.get_job(job_id)
        runtime.update_next_run(job.next_run_time if job else None)
        runtime.add_log(f"定时任务更新为 {hour:02d}:{minute:02d} ({timezone})")


def _remove_user_schedule_job(username: str):
    with scheduler_lock:
        if scheduler is None:
            return
        job_id = _scheduler_job_id(username)
        if scheduler.get_job(job_id):
            scheduler.remove_job(job_id)


def _start_background_run(username: str, trigger: str):
    runtime = _get_runtime(username)

    def _worker():
        runtime.run_once(trigger)
        if scheduler:
            job = scheduler.get_job(_scheduler_job_id(username))
            runtime.update_next_run(job.next_run_time if job else None)

    thread = threading.Thread(target=_worker, daemon=True)
    thread.start()
    return True


def _start_scheduler():
    global scheduler
    with scheduler_lock:
        if scheduler is None:
            scheduler = BackgroundScheduler(timezone=DEFAULT_TIMEZONE)
            scheduler.start()
        scheduler.add_job(
            _retry_failed_tasks_job,
            trigger="interval",
            hours=FAILED_RETRY_INTERVAL_HOURS,
            id=FAILED_RETRY_JOB_ID,
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )

    _start_scheduler_bootstrap(run_startup_tasks=True)


def _stop_scheduler():
    global scheduler, scheduler_bootstrapped, scheduler_bootstrap_running
    with scheduler_lock:
        if scheduler and scheduler.running:
            scheduler.shutdown(wait=False)
            logger.info("Scheduler stopped.")
        scheduler = None
        scheduler_bootstrapped = False
    with scheduler_bootstrap_lock:
        scheduler_bootstrap_running = False


app = FastAPI(title="DouYin Spark Flow Dashboard")
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
templates = Jinja2Templates(directory=str(TEMPLATES_DIR))


@app.on_event("startup")
async def on_startup():
    logger.info("Application startup begin.")
    _start_scheduler()
    atexit.register(_stop_scheduler)
    logger.info("Application startup complete.")


@app.on_event("shutdown")
async def on_shutdown():
    _stop_scheduler()


@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request):
    session = _session_from_request(request)
    if not session:
        return RedirectResponse(url="/login", status_code=303)
    if session.get("role") == "admin":
        return RedirectResponse(url="/admin", status_code=303)

    username = session.get("username")
    runtime = _get_runtime(username)
    return templates.TemplateResponse(
        "dashboard.html",
        {
            "request": request,
            "default_time": runtime.schedule_time(),
            "username": username,
        },
    )


@app.get("/login", response_class=HTMLResponse)
async def login_page(request: Request):
    session = _session_from_request(request)
    if session:
        if session.get("role") == "admin":
            return RedirectResponse(url="/admin", status_code=303)
        return RedirectResponse(url="/", status_code=303)
    return templates.TemplateResponse("login.html", {"request": request})


@app.get("/register", response_class=HTMLResponse)
async def register_page(request: Request):
    session = _session_from_request(request)
    if session:
        if session.get("role") == "admin":
            return RedirectResponse(url="/admin", status_code=303)
        return RedirectResponse(url="/", status_code=303)
    return templates.TemplateResponse("register.html", {"request": request})


@app.get("/admin", response_class=HTMLResponse)
async def admin_page(request: Request):
    session = _session_from_request(request)
    if not session or session.get("role") != "admin":
        return templates.TemplateResponse(
            "admin_login.html",
            {
                "request": request,
                "password_missing": not bool(os.getenv("PASSWORD")),
            },
        )
    return templates.TemplateResponse("admin.html", {"request": request})


@app.post("/api/login")
async def api_login(payload: UserLoginPayload):
    username = payload.username.strip()
    if not username:
        return JSONResponse(status_code=400, content={"ok": False, "message": "用户名不能为空。"})

    users_map = _load_users_meta()
    user = users_map.get(username)
    if not user:
        return JSONResponse(status_code=401, content={"ok": False, "message": "用户名或密码错误。"})

    if not _verify_password(payload.password, user.get("password_salt", ""), user.get("password_hash", "")):
        return JSONResponse(status_code=401, content={"ok": False, "message": "用户名或密码错误。"})

    token = secrets.token_urlsafe(32)
    AUTH_SESSIONS[token] = {"role": "user", "username": username}

    response = JSONResponse({"ok": True, "message": "登录成功。"})
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=token,
        httponly=True,
        samesite="lax",
        max_age=7 * 24 * 3600,
    )
    return response


@app.post("/api/admin/login")
async def api_admin_login(payload: AdminLoginPayload):
    expected_password = os.getenv("PASSWORD")
    if not expected_password:
        return JSONResponse(
            status_code=500,
            content={"ok": False, "message": "服务端未配置 PASSWORD 环境变量。"},
        )

    if payload.password != expected_password:
        return JSONResponse(status_code=401, content={"ok": False, "message": "密码错误。"})

    token = secrets.token_urlsafe(32)
    AUTH_SESSIONS[token] = {"role": "admin", "username": "admin"}
    response = JSONResponse({"ok": True, "message": "登录成功。"})
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=token,
        httponly=True,
        samesite="lax",
        max_age=7 * 24 * 3600,
    )
    return response


@app.post("/api/register")
async def api_register(password: str = Form(...), users_file: UploadFile = File(...)):
    if len(password.strip()) < 4:
        return JSONResponse(status_code=400, content={"ok": False, "message": "密码至少 4 位。"})

    if not users_file.filename.lower().endswith(".json"):
        return JSONResponse(status_code=400, content={"ok": False, "message": "请上传 usersData.json 文件。"})

    try:
        raw = await users_file.read()
        users_data, username, unique_id = _validate_and_normalize_users_data(raw)
    except Exception as exc:
        return JSONResponse(status_code=400, content={"ok": False, "message": str(exc)})

    users_map = _load_users_meta()
    if username in users_map:
        return JSONResponse(status_code=409, content={"ok": False, "message": f"用户名 {username} 已注册。"})

    for existing in users_map.values():
        if str(existing.get("unique_id", "")).strip() == unique_id:
            return JSONResponse(status_code=409, content={"ok": False, "message": f"unique_id {unique_id} 已注册。"})

    default_config = _get_default_user_config()
    default_config.setdefault("scheduler", {})
    default_config["scheduler"].setdefault("enabled", True)
    default_config["scheduler"].setdefault("timezone", DEFAULT_TIMEZONE)
    default_config["scheduler"].setdefault("hour", 9)
    default_config["scheduler"].setdefault("minute", 0)
    default_config["scheduler"].setdefault("runOnStartup", False)

    hash_data = _hash_password(password.strip())
    created_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    try:
        _create_user_record(
            username=username,
            unique_id=unique_id,
            password_hash=hash_data["hash"],
            password_salt=hash_data["salt"],
            created_at=created_at,
            config_payload=default_config,
            users_data_payload=users_data,
        )
    except pymysql.err.IntegrityError:
        return JSONResponse(status_code=409, content={"ok": False, "message": "用户名或 unique_id 已注册。"})

    _schedule_user_job(username)
    _get_runtime(username).add_log("用户已注册并完成定时任务初始化")

    return {
        "ok": True,
        "message": "注册成功，请使用用户名和密码登录。",
        "username": username,
    }


@app.post("/api/logout")
async def api_logout(request: Request):
    token = request.cookies.get(SESSION_COOKIE_NAME)
    if token:
        AUTH_SESSIONS.pop(token, None)
    response = JSONResponse({"ok": True})
    response.delete_cookie(SESSION_COOKIE_NAME)
    return response


@app.get("/api/status")
async def api_status(request: Request):
    session = _require_user_session(request)
    username = session["username"]
    runtime = _get_runtime(username)
    users_data = _load_user_users_data(username)
    return {
        "ok": True,
        "runtime": runtime.snapshot(
            account_count=len(users_data),
            target_count=_count_targets(users_data),
        ),
        "history": runtime.history_rows(),
    }


@app.get("/api/logs")
async def api_logs(request: Request, limit: int = MAX_LOG_LINES):
    session = _require_user_session(request)
    username = session["username"]
    runtime = _get_runtime(username)
    limit = min(max(100, limit), 3000)
    return {"ok": True, "logs": runtime.recent_logs(limit=limit)}


@app.post("/api/run")
async def api_run(request: Request):
    session = _require_user_session(request)
    username = session["username"]
    runtime = _get_runtime(username)

    if runtime.is_running:
        return JSONResponse(
            status_code=409,
            content={"ok": False, "message": "已有任务正在执行，请稍后再试。"},
        )

    _start_background_run(username, "manual")
    return {"ok": True, "message": "任务已开始执行。"}


@app.post("/api/schedule")
async def api_schedule(request: Request, payload: SchedulePayload):
    session = _require_user_session(request)
    username = session["username"]

    try:
        hour, minute = _parse_time_string(payload.time)
    except Exception as exc:
        return JSONResponse(status_code=400, content={"ok": False, "message": str(exc)})

    cfg = _load_user_config(username)
    scheduler_cfg = cfg.setdefault("scheduler", {})
    scheduler_cfg["enabled"] = True
    scheduler_cfg["hour"] = hour
    scheduler_cfg["minute"] = minute
    scheduler_cfg["timezone"] = str(scheduler_cfg.get("timezone", DEFAULT_TIMEZONE))
    scheduler_cfg["runOnStartup"] = bool(scheduler_cfg.get("runOnStartup", False))
    _save_user_config(username, cfg)

    _schedule_user_job(username)
    runtime = _get_runtime(username)
    return {
        "ok": True,
        "message": f"定时任务已更新为每天 {hour:02d}:{minute:02d}。",
        "time": f"{hour:02d}:{minute:02d}",
        "next_run": runtime.snapshot(0, 0)["next_run"],
    }


@app.get("/api/editor/state")
async def api_editor_state(request: Request):
    session = _require_user_session(request)
    username = session["username"]
    return {"ok": True, **_build_editor_state(username)}


@app.post("/api/editor/message")
async def api_editor_message(request: Request, payload: MessageTemplatePayload):
    session = _require_user_session(request)
    username = session["username"]

    message = payload.message.strip()
    if not message:
        return JSONResponse(status_code=400, content={"ok": False, "message": "消息内容不能为空。"})
    if len(message) > MAX_TEMPLATE_LENGTH:
        return JSONResponse(
            status_code=400,
            content={"ok": False, "message": f"消息内容过长，最多 {MAX_TEMPLATE_LENGTH} 字符。"},
        )

    cfg = _load_user_config(username)
    cfg["messageTemplate"] = message
    _save_user_config(username, cfg)
    _get_runtime(username).add_log("消息模板已更新")
    return {"ok": True, "message": "消息模板已保存。"}


@app.post("/api/editor/targets")
async def api_editor_targets(request: Request, payload: UserTargetsPayload):
    session = _require_user_session(request)
    username = session["username"]

    users_data = _load_user_users_data(username)
    updates = {item.unique_id: _sanitize_targets(item.targets) for item in payload.users}

    updated = 0
    for user in users_data:
        uid = str(user.get("unique_id", ""))
        if uid in updates:
            user["targets"] = updates[uid]
            updated += 1

    _save_user_users_data(username, users_data)
    _get_runtime(username).add_log(f"目标好友已更新，涉及账号数：{updated}")
    return {"ok": True, "message": f"目标好友已保存（{updated} 个账号）。"}


@app.get("/api/admin/overview")
async def api_admin_overview(request: Request):
    _require_admin_session(request)
    try:
        users_map = _load_users_meta()
        if not scheduler_bootstrapped:
            _sync_user_jobs_from_meta(users_map, run_startup_tasks=False)
    except Exception as exc:
        logger.warning("Admin overview failed to reach database. error=%s", exc)
        return {
            "ok": True,
            "users": [],
            "task_count": 0,
            "db_status": _build_db_status_payload(),
            "message": f"无法连接 SQL 服务器：{exc}",
        }

    rows = []
    for username, meta in sorted(users_map.items(), key=lambda x: x[0]):
        try:
            cfg = _load_user_config(username)
            users_data = _load_user_users_data(username)
        except Exception as exc:
            rows.append(
                {
                    "username": username,
                    "unique_id": meta.get("unique_id", ""),
                    "created_at": meta.get("created_at", "-"),
                    "error": str(exc),
                }
            )
            continue

        scheduler_cfg = cfg.get("scheduler", {})
        runtime = _get_runtime(username)
        runtime_snapshot = runtime.snapshot(
            account_count=len(users_data),
            target_count=_count_targets(users_data),
        )

        receivers = []
        for item in users_data:
            receivers.extend(item.get("targets", []))

        rows.append(
            {
                "username": username,
                "unique_id": meta.get("unique_id", ""),
                "created_at": meta.get("created_at", "-"),
                "scheduler_enabled": bool(scheduler_cfg.get("enabled", True)),
                "schedule_time": f"{int(scheduler_cfg.get('hour', 9)):02d}:{int(scheduler_cfg.get('minute', 0)):02d}",
                "schedule_timezone": str(scheduler_cfg.get("timezone", DEFAULT_TIMEZONE)),
                "message_template": str(cfg.get("messageTemplate", "")),
                "targets": receivers,
                "target_count": len(receivers),
                "next_run": runtime_snapshot.get("next_run", "-"),
                "last_status": runtime_snapshot.get("last_status", "-"),
                "last_start": runtime_snapshot.get("last_start", "-"),
                "is_running": runtime_snapshot.get("is_running", False),
                "can_retry": bool(
                    not runtime_snapshot.get("is_running", False)
                    and runtime_snapshot.get("last_status") == "失败"
                ),
            }
        )

    return {
        "ok": True,
        "users": rows,
        "task_count": len(rows),
        "db_status": _build_db_status_payload(),
    }


@app.get("/api/admin/tasks/{username}")
async def api_admin_task_detail(request: Request, username: str, log_limit: int = MAX_LOG_LINES):
    _require_admin_session(request)
    username = username.strip()
    user_meta = _get_user_meta_or_404(username)

    try:
        cfg = _load_user_config(username)
        users_data = _load_user_users_data(username)
    except Exception as exc:
        return JSONResponse(
            status_code=500,
            content={"ok": False, "message": f"加载任务详情失败：{exc}"},
        )

    scheduler_cfg = cfg.get("scheduler", {})
    runtime = _get_runtime(username)
    target_count = _count_targets(users_data)
    snapshot = runtime.snapshot(account_count=len(users_data), target_count=target_count)

    accounts = []
    all_targets = []
    for item in users_data:
        targets = _sanitize_targets(item.get("targets", []))
        all_targets.extend(targets)
        accounts.append(
            {
                "username": str(item.get("username", "未知用户")),
                "unique_id": str(item.get("unique_id", "")),
                "target_count": len(targets),
                "targets": targets,
                "cookie_count": len(item.get("cookies", [])) if isinstance(item.get("cookies", []), list) else 0,
            }
        )

    log_limit = min(max(100, log_limit), 3000)
    return {
        "ok": True,
        "task": {
            "username": username,
            "unique_id": user_meta.get("unique_id", ""),
            "created_at": user_meta.get("created_at", "-"),
            "scheduler_enabled": bool(scheduler_cfg.get("enabled", True)),
            "schedule_time": f"{int(scheduler_cfg.get('hour', 9)):02d}:{int(scheduler_cfg.get('minute', 0)):02d}",
            "schedule_timezone": str(scheduler_cfg.get("timezone", DEFAULT_TIMEZONE)),
            "message_template": str(cfg.get("messageTemplate", "")),
            "targets": all_targets,
            "target_count": len(all_targets),
            "runtime": snapshot,
            "can_retry": bool(not snapshot.get("is_running", False) and snapshot.get("last_status") == "失败"),
            "history": runtime.history_rows(),
            "logs": runtime.recent_logs(limit=log_limit),
            "config": {
                "multiTask": bool(cfg.get("multiTask", True)),
                "taskCount": int(cfg.get("taskCount", 1) or 1),
                "hitokotoTypes": cfg.get("hitokotoTypes", []),
                "proxyAddress": str(cfg.get("proxyAddress", "")),
            },
            "accounts": accounts,
        },
    }


@app.post("/api/admin/tasks/{username}/retry")
async def api_admin_retry_task(request: Request, username: str):
    _require_admin_session(request)
    username = username.strip()
    _get_user_meta_or_404(username)

    runtime = _get_runtime(username)
    snapshot = runtime.snapshot(account_count=0, target_count=0)
    if snapshot.get("is_running"):
        return JSONResponse(status_code=409, content={"ok": False, "message": "任务正在运行中，请稍后再试。"})
    if snapshot.get("last_status") != "失败":
        return JSONResponse(status_code=409, content={"ok": False, "message": "当前任务不是失败状态，无需重试。"})

    runtime.add_log("管理员手动触发失败任务重试")
    _start_background_run(username, "admin_retry")
    return {"ok": True, "message": f"已开始重试 {username} 的失败任务。"}


@app.post("/api/admin/tasks/retry-failed")
async def api_admin_retry_all_failed_tasks(request: Request):
    _require_admin_session(request)

    try:
        usernames = _retry_failed_tasks_once("admin_bulk_retry", raise_on_db_error=True)
    except Exception as exc:
        return JSONResponse(status_code=503, content={"ok": False, "message": f"???? SQL ????{exc}"})

    if not usernames:
        return {"ok": True, "message": "?????????????", "count": 0, "usernames": []}

    return {
        "ok": True,
        "message": f"??????? {len(usernames)} ??????",
        "count": len(usernames),
        "usernames": usernames,
    }


@app.post("/api/admin/tasks/{username}/delete")
async def api_admin_delete_task(request: Request, username: str):
    _require_admin_session(request)
    username = username.strip()
    _get_user_meta_or_404(username)

    cfg = _load_user_config(username)
    scheduler_cfg = cfg.setdefault("scheduler", {})
    scheduler_cfg["enabled"] = False
    _save_user_config(username, cfg)

    _remove_user_schedule_job(username)
    runtime = _get_runtime(username)
    runtime.update_next_run(None)
    runtime.add_log("管理员已删除（禁用）该用户定时任务")

    return {"ok": True, "message": f"已删除用户 {username} 的定时任务。"}


@app.delete("/api/admin/users/{username}")
async def api_admin_delete_user(request: Request, username: str):
    _require_admin_session(request)
    username = username.strip()

    _get_user_meta_or_404(username)

    _remove_user_schedule_job(username)
    _delete_user_record(username)
    _delete_runtime(username)

    return {"ok": True, "message": f"用户 {username} 已删除。"}


@app.get("/health")
async def health():
    return {"ok": True, "status": "alive"}


def run_server():
    port = int(os.getenv("PORT", "7860"))
    uvicorn.run("app:app", host="0.0.0.0", port=port, workers=1)


if __name__ == "__main__":
    run_server()