Upload 3 files

app.py

@@ -1,6 +1,6 @@
-import asyncio
-import atexit
-import hashlib
+import asyncio
+import atexit
+import hashlib
 import json
 import logging
 import os
@@ -15,20 +15,20 @@ 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)
-
+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"
@@ -47,180 +47,190 @@ DEFAULT_TIMEZONE = "Asia/Shanghai"
 MAX_LOG_LINES = 1200
 MAX_TEMPLATE_LENGTH = 2000
 PASSWORD_ITERATIONS = 210000
-
-DEFAULT_USER_CONFIG = {
-    "multiTask": True,
-    "taskCount": 5,
-    "proxyAddress": "",
-    "messageTemplate": "[续火花]",
-    "hitokotoTypes": ["文学", "影视", "诗词", "哲学"],
-    "scheduler": {
-        "enabled": True,
-        "timezone": DEFAULT_TIMEZONE,
-        "hour": 9,
-        "minute": 0,
-        "runOnStartup": False,
-    },
-}
-
+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
-
-
-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]
-
-
+db_status_lock = threading.Lock()
+db_status = {
+    "connected": None,
+    "last_checked_at": None,
+    "last_ok_at": None,
+    "last_error": "",
+}
+scheduler_bootstrapped = 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:
@@ -231,22 +241,22 @@ def _ensure_data_layout():
         _init_db_schema()
         _migrate_legacy_file_data_if_needed()
         db_initialized = True
-
-
-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 _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)
@@ -274,6 +284,35 @@ def _merge_config_with_defaults(raw_cfg: Any):
     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:
@@ -331,8 +370,18 @@ def _build_mysql_conn_kwargs():
     return kwargs
 
 
+def _db_connect():
+    try:
+        conn = pymysql.connect(**_build_mysql_conn_kwargs())
+    except Exception as exc:
+        _update_db_status(False, exc)
+        raise
+    _update_db_status(True)
+    return conn
+
+
 def _db_query_all(query: str, params=()):
-    conn = pymysql.connect(**_build_mysql_conn_kwargs())
+    conn = _db_connect()
     try:
         with conn.cursor() as cursor:
             cursor.execute(query, params)
@@ -342,7 +391,7 @@ def _db_query_all(query: str, params=()):
 
 
 def _db_query_one(query: str, params=()):
-    conn = pymysql.connect(**_build_mysql_conn_kwargs())
+    conn = _db_connect()
     try:
         with conn.cursor() as cursor:
             cursor.execute(query, params)
@@ -352,7 +401,7 @@ def _db_query_one(query: str, params=()):
 
 
 def _db_execute(query: str, params=()):
-    conn = pymysql.connect(**_build_mysql_conn_kwargs())
+    conn = _db_connect()
     try:
         with conn.cursor() as cursor:
             cursor.execute(query, params)
@@ -362,6 +411,7 @@ def _db_execute(query: str, params=()):
 
 
 def _init_db_schema():
+    _ensure_data_layout()
     _db_execute(
         f"""
         CREATE TABLE IF NOT EXISTS `{USERS_TABLE}` (
@@ -455,6 +505,7 @@ def _migrate_legacy_file_data_if_needed():
 
 
 def _load_users_meta():
+    _ensure_data_layout()
     rows = _db_query_all(
         f"""
         SELECT username, unique_id, password_hash, password_salt, created_at
@@ -466,6 +517,7 @@ def _load_users_meta():
 
 
 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
@@ -477,6 +529,7 @@ def _load_user_row(username: str):
 
 
 def _user_exists(username: str):
+    _ensure_data_layout()
     row = _db_query_one(
         f"SELECT 1 AS ok FROM `{USERS_TABLE}` WHERE username=%s",
         (username,),
@@ -513,9 +566,10 @@ def _create_user_record(
 
 
 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)
@@ -576,387 +630,449 @@ def _save_user_users_data(username: str, users_data: list):
     )
     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 _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
-    _ensure_data_layout()
-    with scheduler_lock:
-        if scheduler is None:
-            scheduler = BackgroundScheduler(timezone=DEFAULT_TIMEZONE)
-            scheduler.start()
-
-    users_map = _load_users_meta()
-    for username in users_map.keys():
-        _schedule_user_job(username)
-        cfg = _load_user_config(username)
-        run_on_startup = bool(cfg.get("scheduler", {}).get("runOnStartup", False))
-        if run_on_startup:
-            _start_background_run(username, "startup")
-
-
-def _stop_scheduler():
-    global scheduler
-    with scheduler_lock:
-        if scheduler and scheduler.running:
-            scheduler.shutdown(wait=False)
-            logger.info("Scheduler stopped.")
-        scheduler = None
-
-
-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():
-    _ensure_data_layout()
-    _start_scheduler()
-    atexit.register(_stop_scheduler)
-
-
-@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} 已注册。"})
-
+
+
+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
+
+    for username in users_map.keys():
+        _schedule_user_job(username)
+        if run_startup_tasks:
+            cfg = _load_user_config(username)
+            run_on_startup = bool(cfg.get("scheduler", {}).get("runOnStartup", False))
+            if run_on_startup:
+                _start_background_run(username, "startup")
+
+    scheduler_bootstrapped = 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, scheduler_bootstrapped
+    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,
+        )
+
+    try:
+        _ensure_data_layout()
+        users_map = _load_users_meta()
+        _sync_user_jobs_from_meta(users_map, run_startup_tasks=True)
+    except Exception as exc:
+        scheduler_bootstrapped = False
+        logger.warning("Scheduler bootstrap skipped, database unavailable. error=%s", exc)
+
+
+def _stop_scheduler():
+    global scheduler, scheduler_bootstrapped
+    with scheduler_lock:
+        if scheduler and scheduler.running:
+            scheduler.shutdown(wait=False)
+            logger.info("Scheduler stopped.")
+        scheduler = None
+        scheduler_bootstrapped = 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():
+    _start_scheduler()
+    atexit.register(_stop_scheduler)
+
+
+@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)
@@ -982,285 +1098,341 @@ async def api_register(password: str = Form(...), users_file: UploadFile = File(
 
     _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)
-    users_map = _load_users_meta()
-
-    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),
-            }
-        )
-
-    return {
-        "ok": True,
-        "users": rows,
-        "task_count": len(rows),
-    }
-
-
-@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,
-            "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}/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()
-
+
+    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)
@@ -1268,17 +1440,17 @@ async def api_admin_delete_user(request: Request, username: str):
     _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()
+
+
+@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()

templates/admin.html

@@ -1,9 +1,9 @@
-<!doctype html>
+<!doctype html>
 <html lang="zh-CN">
 <head>
   <meta charset="utf-8">
   <meta name="viewport" content="width=device-width, initial-scale=1">
-  <title>DouYin Spark Flow - Admin 控制台</title>
+  <title>DouYin Spark Helper - Admin 控制台</title>
   <link rel="stylesheet" href="/static/style.css">
 </head>
 <body class="dash-body">
@@ -13,6 +13,7 @@
       <p>用户管理 + 定时任务总览</p>
     </div>
     <div class="top-actions">
+      <button id="retryAllFailedBtn" class="btn ghost">一键重试失败任务</button>
       <button id="refreshBtn" class="btn ghost">刷新</button>
       <button id="logoutBtn" class="btn ghost">退出登录</button>
     </div>
@@ -22,8 +23,9 @@
     <section class="panel">
       <h2>用户与任务总览</h2>
       <p id="summary" class="muted">加载中...</p>
+      <div id="dbAlert" class="alert warning" style="display:none;"></div>
       <div class="table-wrap">
-        <table>
+        <table class="admin-table">
           <thead>
             <tr>
               <th>发起用户</th>
@@ -45,12 +47,94 @@
       </div>
       <p id="adminMsg" class="msg"></p>
     </section>
+
+    <section id="detailPanel" class="panel" style="display:none;">
+      <div class="panel-header">
+        <h2 id="detailTitle">任务详情</h2>
+        <div class="top-actions">
+          <button id="detailRetryBtn" class="btn" style="display:none;">重试失败任务</button>
+          <button id="detailRefreshBtn" class="btn">刷新详情</button>
+        </div>
+      </div>
+
+      <div id="detailMeta" class="detail-grid"></div>
+
+      <div class="detail-split">
+        <article class="detail-card">
+          <h3>消息内容</h3>
+          <pre id="detailMessage" class="detail-message">-</pre>
+        </article>
+        <article class="detail-card">
+          <h3>接收方（去重后）</h3>
+          <pre id="detailTargets" class="detail-message">-</pre>
+        </article>
+      </div>
+
+      <article class="detail-card">
+        <h3>账号明细</h3>
+        <div class="table-wrap">
+          <table class="admin-table detail-history">
+            <thead>
+              <tr>
+                <th>账号昵称</th>
+                <th>unique_id</th>
+                <th>目标数</th>
+                <th>cookies 数</th>
+                <th>目标列表</th>
+              </tr>
+            </thead>
+            <tbody id="detailAccountsBody">
+              <tr><td colspan="5">暂无数据</td></tr>
+            </tbody>
+          </table>
+        </div>
+      </article>
+
+      <article class="detail-card">
+        <h3>执行历史</h3>
+        <div class="table-wrap">
+          <table class="admin-table detail-history">
+            <thead>
+              <tr>
+                <th>触发方式</th>
+                <th>开始时间</th>
+                <th>结束时间</th>
+                <th>状态</th>
+                <th>耗时</th>
+                <th>信息</th>
+              </tr>
+            </thead>
+            <tbody id="detailHistoryBody">
+              <tr><td colspan="6">暂无记录</td></tr>
+            </tbody>
+          </table>
+        </div>
+      </article>
+
+      <article class="detail-card">
+        <h3>任务日志</h3>
+        <pre id="detailLogs" class="detail-logbox">暂无日志。</pre>
+      </article>
+    </section>
   </main>
 
   <script>
     const adminBody = document.getElementById("adminBody");
     const summary = document.getElementById("summary");
+    const retryAllFailedBtn = document.getElementById("retryAllFailedBtn");
+    const dbAlert = document.getElementById("dbAlert");
     const adminMsg = document.getElementById("adminMsg");
+    const detailPanel = document.getElementById("detailPanel");
+    const detailTitle = document.getElementById("detailTitle");
+    const detailMeta = document.getElementById("detailMeta");
+    const detailMessage = document.getElementById("detailMessage");
+    const detailTargets = document.getElementById("detailTargets");
+    const detailAccountsBody = document.getElementById("detailAccountsBody");
+    const detailHistoryBody = document.getElementById("detailHistoryBody");
+    const detailLogs = document.getElementById("detailLogs");
+    const detailRetryBtn = document.getElementById("detailRetryBtn");
+    let currentDetailUser = null;
+    let dbConnected = true;
 
     function setMsg(msg, isError = false) {
       adminMsg.textContent = msg || "";
@@ -85,9 +169,49 @@
         .replace(/'/g, "&#39;");
     }
 
+    function renderDbStatus(status, fallbackMessage = "") {
+      dbConnected = !status || status.connected !== false;
+      if (dbConnected) {
+        dbAlert.style.display = "none";
+        dbAlert.textContent = "";
+        return;
+      }
+
+      const parts = ["当前无法连接 SQL 服务器。"];
+      if (fallbackMessage) {
+        parts.push(fallbackMessage);
+      } else if (status && status.last_error) {
+        parts.push(status.last_error);
+      }
+      if (status && status.last_checked_at && status.last_checked_at !== "-") {
+        parts.push(`最近检查：${status.last_checked_at}`);
+      }
+      if (status && status.last_ok_at && status.last_ok_at !== "-") {
+        parts.push(`最近成功连接：${status.last_ok_at}`);
+      }
+
+      dbAlert.textContent = parts.join(" ");
+      dbAlert.style.display = "block";
+    }
+
+    function clearDetailPanel() {
+      currentDetailUser = null;
+      detailPanel.style.display = "none";
+      detailRetryBtn.style.display = "none";
+      detailRetryBtn.dataset.user = "";
+      detailTitle.textContent = "任务详情";
+      detailMeta.innerHTML = "";
+      detailMessage.textContent = "-";
+      detailTargets.textContent = "-";
+      detailLogs.textContent = "暂无日志。";
+      detailAccountsBody.innerHTML = '<tr><td colspan="5">暂无数据</td></tr>';
+      detailHistoryBody.innerHTML = '<tr><td colspan="6">暂无记录</td></tr>';
+    }
+
     function renderRows(users) {
       if (!users || users.length === 0) {
         adminBody.innerHTML = '<tr><td colspan="10">暂无用户</td></tr>';
+        clearDetailPanel();
         return;
       }
 
@@ -100,7 +224,9 @@
               <td>${escapeHtml(u.created_at || "-")}</td>
               <td colspan="6" style="color:#c0392b;">数据异常：${escapeHtml(u.error)}</td>
               <td>
-                <button class="btn admin-delete-user" data-user="${escapeHtml(u.username)}">删除用户</button>
+                <div class="admin-actions">
+                  <button class="btn admin-delete-user" data-user="${escapeHtml(u.username)}">删用户</button>
+                </div>
               </td>
             </tr>
           `;
@@ -108,6 +234,10 @@
 
         const targets = Array.isArray(u.targets) ? u.targets : [];
         const targetText = targets.length ? targets.join(" / ") : "-";
+        const statusText = u.is_running ? "运行中" : (u.last_status || "-");
+        const retryAction = u.can_retry
+          ? `<button class="btn admin-retry-task" data-user="${escapeHtml(u.username)}">重试失败</button>`
+          : "";
 
         return `
           <tr>
@@ -116,12 +246,14 @@
             <td>${escapeHtml(u.created_at)}</td>
             <td>${u.scheduler_enabled ? "启用" : "禁用"}</td>
             <td>${escapeHtml(u.schedule_time)} (${escapeHtml(u.schedule_timezone)})</td>
-            <td class="clamp-cell">${escapeHtml(u.message_template || "")}</td>
-            <td class="clamp-cell">${escapeHtml(targetText)}</td>
+            <td class="ellipsis-cell" title="${escapeHtml(u.message_template || "")}">${escapeHtml(u.message_template || "-")}</td>
+            <td class="ellipsis-cell" title="${escapeHtml(targetText)}">${escapeHtml(targetText)}</td>
             <td>${escapeHtml(u.next_run || "-")}</td>
-            <td>${u.is_running ? "运行中" : escapeHtml(u.last_status || "-")}</td>
+            <td>${escapeHtml(statusText)}</td>
             <td>
               <div class="admin-actions">
+                <button class="btn admin-view-task" data-user="${escapeHtml(u.username)}">详情/日志</button>
+                ${retryAction}
                 <button class="btn admin-del-task" data-user="${escapeHtml(u.username)}">删任务</button>
                 <button class="btn admin-delete-user" data-user="${escapeHtml(u.username)}">删用户</button>
               </div>
@@ -130,6 +262,13 @@
         `;
       }).join("");
 
+      document.querySelectorAll(".admin-view-task").forEach((btn) => {
+        btn.addEventListener("click", async () => {
+          const username = btn.dataset.user;
+          await loadTaskDetail(username, true);
+        });
+      });
+
       document.querySelectorAll(".admin-del-task").forEach((btn) => {
         btn.addEventListener("click", async () => {
           const username = btn.dataset.user;
@@ -142,12 +281,22 @@
             });
             setMsg(data.message || "已删除任务");
             await loadOverview();
+            if (currentDetailUser === username) {
+              await loadTaskDetail(username, false);
+            }
           } catch (err) {
             setMsg(err.message, true);
           }
         });
       });
 
+      document.querySelectorAll(".admin-retry-task").forEach((btn) => {
+        btn.addEventListener("click", async () => {
+          const username = btn.dataset.user;
+          await retryFailedTask(username);
+        });
+      });
+
       document.querySelectorAll(".admin-delete-user").forEach((btn) => {
         btn.addEventListener("click", async () => {
           const username = btn.dataset.user;
@@ -158,6 +307,9 @@
               method: "DELETE",
             });
             setMsg(data.message || "用户已删除");
+            if (currentDetailUser === username) {
+              clearDetailPanel();
+            }
             await loadOverview();
           } catch (err) {
             setMsg(err.message, true);
@@ -166,29 +318,180 @@
       });
     }
 
+    function renderDetail(task) {
+      currentDetailUser = task.username;
+      detailPanel.style.display = "block";
+      detailTitle.textContent = `任务详情 · ${task.username}`;
+
+      const runtime = task.runtime || {};
+      const config = task.config || {};
+      const targets = Array.isArray(task.targets) ? task.targets : [];
+
+      detailMeta.innerHTML = `
+        <div class="detail-item"><span>发起用户</span><strong>${escapeHtml(task.username || "-")}</strong></div>
+        <div class="detail-item"><span>唯一标识</span><strong>${escapeHtml(task.unique_id || "-")}</strong></div>
+        <div class="detail-item"><span>注册时间</span><strong>${escapeHtml(task.created_at || "-")}</strong></div>
+        <div class="detail-item"><span>定时状态</span><strong>${task.scheduler_enabled ? "启用" : "禁用"}</strong></div>
+        <div class="detail-item"><span>发送时间</span><strong>${escapeHtml(task.schedule_time || "-")} (${escapeHtml(task.schedule_timezone || "-")})</strong></div>
+        <div class="detail-item"><span>下一次执行</span><strong>${escapeHtml(runtime.next_run || "-")}</strong></div>
+        <div class="detail-item"><span>最近状态</span><strong>${runtime.is_running ? "运行中" : escapeHtml(runtime.last_status || "-")}</strong></div>
+        <div class="detail-item"><span>最近开始</span><strong>${escapeHtml(runtime.last_start || "-")}</strong></div>
+        <div class="detail-item"><span>账号数 / 目标数</span><strong>${runtime.account_count || 0} / ${runtime.target_count || 0}</strong></div>
+        <div class="detail-item"><span>并发设置</span><strong>multiTask=${config.multiTask ? "true" : "false"}, taskCount=${config.taskCount || 1}</strong></div>
+      `;
+
+      detailMessage.textContent = task.message_template || "-";
+      detailTargets.textContent = targets.length ? targets.join("\n") : "-";
+      detailLogs.textContent = task.logs || "暂无日志。";
+      detailRetryBtn.dataset.user = task.username || "";
+      detailRetryBtn.style.display = task.can_retry ? "inline-flex" : "none";
+
+      const accounts = Array.isArray(task.accounts) ? task.accounts : [];
+      if (!accounts.length) {
+        detailAccountsBody.innerHTML = '<tr><td colspan="5">暂无账号明细</td></tr>';
+      } else {
+        detailAccountsBody.innerHTML = accounts.map((item) => {
+          const t = Array.isArray(item.targets) ? item.targets : [];
+          return `
+            <tr>
+              <td>${escapeHtml(item.username || "-")}</td>
+              <td>${escapeHtml(item.unique_id || "-")}</td>
+              <td>${item.target_count || 0}</td>
+              <td>${item.cookie_count || 0}</td>
+              <td>${escapeHtml(t.join(" / ") || "-")}</td>
+            </tr>
+          `;
+        }).join("");
+      }
+
+      const history = Array.isArray(task.history) ? task.history : [];
+      if (!history.length) {
+        detailHistoryBody.innerHTML = '<tr><td colspan="6">暂无记录</td></tr>';
+      } else {
+        detailHistoryBody.innerHTML = history.map((row) => `
+          <tr>
+            <td>${escapeHtml(row.trigger || "-")}</td>
+            <td>${escapeHtml(row.start || "-")}</td>
+            <td>${escapeHtml(row.end || "-")}</td>
+            <td>${escapeHtml(row.status || "-")}</td>
+            <td>${escapeHtml(row.duration || "-")}</td>
+            <td>${escapeHtml(row.message || "-")}</td>
+          </tr>
+        `).join("");
+      }
+    }
+
+    async function loadTaskDetail(username, showMessage = false) {
+      try {
+        const data = await requestJSON(`/api/admin/tasks/${encodeURIComponent(username)}?log_limit=1200`);
+        renderDetail(data.task || {});
+        if (showMessage) {
+          setMsg(`已加载 ${username} 的任务详情与日志。`);
+        }
+      } catch (err) {
+        setMsg(err.message, true);
+      }
+    }
+
+    async function retryAllFailedTasks() {
+      if (!dbConnected) {
+        setMsg("SQL 连接异常，暂时无法批量重试失败任务。", true);
+        return;
+      }
+
+      setMsg("正在批量重试失败任务...");
+      try {
+        const data = await requestJSON("/api/admin/tasks/retry-failed", {
+          method: "POST",
+          body: "{}",
+        });
+        setMsg(data.message || "已开始批量重试失败任务");
+        await loadOverview();
+        if (currentDetailUser && dbConnected) {
+          await loadTaskDetail(currentDetailUser, false);
+        }
+      } catch (err) {
+        setMsg(err.message, true);
+      }
+    }
+
+    async function retryFailedTask(username) {
+      if (!username) {
+        setMsg("未找到可重试的任务用户。", true);
+        return;
+      }
+
+      setMsg("正在重试失败任务...");
+      try {
+        const data = await requestJSON(`/api/admin/tasks/${encodeURIComponent(username)}/retry`, {
+          method: "POST",
+          body: "{}",
+        });
+        setMsg(data.message || "已开始重试失败任务");
+        await loadOverview();
+        if (currentDetailUser === username && dbConnected) {
+          await loadTaskDetail(username, false);
+        }
+      } catch (err) {
+        setMsg(err.message, true);
+      }
+    }
+
     async function loadOverview() {
       const data = await requestJSON("/api/admin/overview");
-      summary.textContent = `共 ${data.task_count || 0} 个用户任务。`;
+      renderDbStatus(data.db_status || null, data.message || "");
+      summary.textContent = dbConnected
+        ? `共 ${data.task_count || 0} 个用户任务。`
+        : "SQL 连接异常，当前未能加载用户任务。";
       renderRows(data.users || []);
+      if (!dbConnected) {
+        clearDetailPanel();
+      }
+      return data;
     }
 
+    retryAllFailedBtn.addEventListener("click", async () => {
+      await retryAllFailedTasks();
+    });
+
     document.getElementById("refreshBtn").addEventListener("click", async () => {
       try {
         await loadOverview();
+        if (dbConnected && currentDetailUser) {
+          await loadTaskDetail(currentDetailUser, false);
+        }
       } catch (err) {
         setMsg(err.message, true);
       }
     });
 
+    document.getElementById("detailRefreshBtn").addEventListener("click", async () => {
+      if (!currentDetailUser) {
+        setMsg("请先在上方选择一个任务。", true);
+        return;
+      }
+      await loadTaskDetail(currentDetailUser, true);
+    });
+
+    detailRetryBtn.addEventListener("click", async () => {
+      await retryFailedTask(detailRetryBtn.dataset.user || currentDetailUser);
+    });
+
     document.getElementById("logoutBtn").addEventListener("click", async () => {
       await fetch("/api/logout", { method: "POST", credentials: "same-origin" });
       window.location.href = "/admin";
     });
 
     loadOverview().catch((err) => setMsg(err.message, true));
-    setInterval(() => {
-      loadOverview().catch(() => {});
+    setInterval(async () => {
+      try {
+        await loadOverview();
+        if (dbConnected && currentDetailUser) {
+          await loadTaskDetail(currentDetailUser, false);
+        }
+      } catch (_) {
+      }
     }, 8000);
   </script>
 </body>
-</html>
+</html>