re-xxm/reaction_board/cli.py

# -*- coding: utf-8 -*-
"""看板命令行入口。

    python app.py init        # 建库 / 迁移 / 写入黑名单种子（离线）
    python app.py ingest      # 手动入库 B 站视频链接 / BV 号 / av 号（只写本地库）
    python app.py snapshot    # 采集当日七项快照与粉丝数
    python app.py snapshot-seed  # 用上一完成日克隆当日壳（离线，可 --date）
    python app.py discover    # 执行一次搜索发现并写入候选池
    python app.py candidates  # 查看候选池（离线）
    python app.py auth-check  # 检查 Cookie 登录态
    python app.py run         # 按配置周期常驻运行（--once 只跑一次）
    python app.py status      # 查看库内状态（离线）

收藏夹重排序使用独立的 ``favsort.py``（唯一使用 ``bilibili.write`` 的功能）。
P2 起 ``python app.py`` 将在此基础上启动本地看板服务。
"""

from __future__ import annotations

import argparse
import json
import sys
import threading
import time
from pathlib import Path
from typing import Sequence

from .bilibili.auth_flow import AuthService
from .bilibili.client import BilibiliClient, Pacer
from .bilibili.provider import ClientProvider
from .bilibili.search import SearchClient
from .config import ConfigError, ConfigStore, default_config_path
from .database import Database
from .logging_json import JsonLogger
from .notify.telegram import TelegramClient
from .repository import Repository
from .scheduler import ScheduledTask, Scheduler, TaskControl
from .services.discovery import DiscoveryService
from .services.library import LibraryService
from .services.snapshot import SnapshotService
from .services.telegram_poller import TelegramPoller
from .services.up_profile import UpProfileService
from . import task_state as task_state_mod
from .timeutil import (
    in_snapshot_quiet_window,
    interval_next_run,
    is_date,
    next_after_snapshot_quiet_window,
    next_daily_run,
    parse_hhmm,
    shanghai_date,
    shanghai_datetime,
)

ROOT = Path(__file__).resolve().parent.parent
DEFAULT_CONFIG = default_config_path(ROOT)


def _build_parser() -> argparse.ArgumentParser:
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument(
        "--config",
        default=argparse.SUPPRESS,
        help="配置文件路径（默认优先 config.jsonc，否则 config.json）",
    )

    parser = argparse.ArgumentParser(
        prog="app.py",
        description="咻咻满 Reaction 数据看板 —— 数据底座、发现与每日快照",
    )
    parser.add_argument(
        "--config",
        default=str(DEFAULT_CONFIG),
        help="配置文件路径（默认优先 config.jsonc，否则 config.json）",
    )
    sub = parser.add_subparsers(dest="command", required=False)

    sub.add_parser("init", parents=[common], help="建库、迁移并写入黑名单种子")
    snapshot = sub.add_parser(
        "snapshot", parents=[common], help="采集当日七项快照与粉丝数"
    )
    snapshot.add_argument(
        "--force", action="store_true", help="忽略当日完成标记，重新采集"
    )
    snapshot_seed = sub.add_parser(
        "snapshot-seed",
        parents=[common],
        help="用上一完成日克隆当日快照壳（离线，不调 B 站）",
    )
    snapshot_seed.add_argument(
        "--date",
        default=None,
        metavar="YYYY-MM-DD",
        help="目标统计日（默认上海时区今天）",
    )
    discover = sub.add_parser(
        "discover", parents=[common], help="执行一次搜索发现并写入候选池"
    )
    discover.add_argument(
        "--no-notify",
        action="store_true",
        help="只写候选池，不发送 Telegram 通知",
    )
    candidates = sub.add_parser(
        "candidates", parents=[common], help="查看候选池（离线）"
    )
    candidates.add_argument(
        "--decision",
        choices=("accepted", "rejected"),
        default=None,
        help="只查看指定决策状态的候选",
    )
    ingest = sub.add_parser(
        "ingest", parents=[common],
        help="手动入库：B 站视频链接、BV 号或 av 号（只写本地库）",
    )
    ingest.add_argument(
        "refs", nargs="+", metavar="REF",
        help="视频链接、BV 号（BV…）、av 号（av12345）或含它们的字符串",
    )
    auth = sub.add_parser("auth", parents=[common], help="扫码授权并更新指定账号 Cookie")
    auth.add_argument("--account", choices=("read", "write"), default="read",
                      help="要授权的账号角色（默认 read）")
    auth.add_argument("--no-browser", action="store_true", help="不自动打开二维码图片（仍在终端打印 ASCII 二维码）")
    auth.add_argument("--timeout", type=int, default=180, metavar="SEC",
                      help="等待扫码的超时时间（默认 180 秒）")
    auth_check = sub.add_parser("auth-check", parents=[common], help="检查账号登录态")
    auth_check.add_argument("--account", choices=("read", "write", "all"), default="read",
                            help="要检查的账号角色（默认 read）")
    run = sub.add_parser("run", parents=[common], help="按配置周期常驻运行发现与快照")
    run.add_argument(
        "--once", action="store_true", help="只执行一次到期任务后退出"
    )
    serve = sub.add_parser("serve", parents=[common], help="启动只读本地看板（不运行调度）")
    serve.add_argument(
        "--preview", action="store_true",
        help="以预览模式启动（要求数据库含 preview:seeded=1 标记）",
    )
    backfill = sub.add_parser(
        "backfill-followers", parents=[common],
        help="为历史视频回填 UP 粉丝数快照（只补空值）",
    )
    backfill.add_argument("--limit", type=int, default=None, help="最多处理多少个 UP")
    backfill.add_argument("--dry-run", action="store_true", help="只统计待回填数量，不发请求")
    sub.add_parser("status", parents=[common], help="查看数据库状态（离线）")
    return parser


def _open_db(cfg, logger: JsonLogger) -> Database:
    db = Database(
        cfg.storage.database_path,
        backup_dir=cfg.storage.backup_dir,
        backup_keep_days=cfg.storage.backup_keep_days,
        logger=logger,
    )
    db.initialize()
    return db


def _print(payload) -> None:
    print(json.dumps(payload, ensure_ascii=False, indent=2), flush=True)


def _task_running_key(name: str) -> str:
    return f"task:running:{name}"


def _mark_task_running(
    repo: Repository,
    name: str,
    *,
    started_at: int | None = None,
    stat_date: str | None = None,
) -> None:
    """任务开始：running 标记 + 持久化 status=running（跨重启可续跑）。"""
    now = int(started_at if started_at is not None else time.time())
    record = {
        "name": name,
        "started_at": now,
    }
    try:
        repo.kv_set(_task_running_key(name), json.dumps(record, ensure_ascii=False))
    except Exception:  # noqa: BLE001
        pass
    try:
        task_state_mod.mark_running(
            repo, name, stat_date=stat_date, now=float(now),
        )
    except Exception:  # noqa: BLE001
        pass


def _clear_task_running(repo: Repository, name: str) -> None:
    try:
        repo.kv_delete(_task_running_key(name))
    except Exception:  # noqa: BLE001
        pass


def _is_task_running(repo: Repository, name: str) -> bool:
    """供 seed 等任务检测全站快照是否在跑（避让，而非互斥跳过成功）。"""
    try:
        return bool(repo.kv_get(_task_running_key(name)))
    except Exception:  # noqa: BLE001
        return False


def _record_task(
    repo: Repository,
    name: str,
    *,
    ok: bool,
    started_at: int | None = None,
    finished_at: int | None = None,
    payload=None,
    error: str | None = None,
    skipped: bool = False,
    next_due_at: float | None = None,
    stat_date: str | None = None,
    progress=None,
) -> None:
    """把最近一次任务结果写入 kv，并清除 running 标记。

    ``skipped=True`` 表示主动延后/避让，不得记成成功，也不推进「当日已完成」。
    同步写入 ``task:state``：有 ``next_due_at`` 用其排程；否则结束残留
    ``running`` 并保留原 ``due_at``（CLI 一锤子路径）。
    """
    now = int(time.time())
    if skipped:
        ok = False
    finished = int(finished_at if finished_at is not None else now)
    record = {
        "name": name,
        "ok": bool(ok),
        "skipped": bool(skipped),
        "started_at": int(started_at if started_at is not None else now),
        "finished_at": finished,
        "error": error,
        "summary": payload,
    }
    try:
        repo.kv_set(f"task:last:{name}", json.dumps(record, ensure_ascii=False))
    except Exception:  # noqa: BLE001  任务状态记录失败不影响主流程
        pass
    _clear_task_running(repo, name)
    try:
        if next_due_at is not None:
            due = float(next_due_at)
        else:
            # CLI 等非调度路径：结束 running，保留原 due_at，避免残留 running。
            prev = task_state_mod.load_state(repo, name)
            if prev is None or prev.status != task_state_mod.STATUS_RUNNING:
                return
            due = (
                float(prev.due_at)
                if prev.due_at and float(prev.due_at) > 0
                else float(finished)
            )
            if stat_date is None:
                stat_date = prev.stat_date
        if skipped:
            task_state_mod.mark_deferred(
                repo,
                name,
                due_at=due,
                now=float(finished),
                stat_date=stat_date,
                progress=progress,
                error=error,
            )
        elif ok:
            task_state_mod.mark_done(
                repo,
                name,
                due_at=due,
                now=float(finished),
                stat_date=stat_date,
                progress=progress,
            )
        else:
            task_state_mod.mark_failed(
                repo,
                name,
                due_at=due,
                now=float(finished),
                stat_date=stat_date,
                progress=progress,
                error=error,
            )
    except Exception:  # noqa: BLE001
        pass


# 全站快照 / 粉丝失败后的补跑间隔（秒）
_TASK_RETRY_SECONDS = 300.0
# snapshot_seed：全站快照占用 / 尚无 last_completed 时的补跑间隔（秒）
_SEED_RETRY_SECONDS = 600.0
_QUIET_SKIP_REASON = "snapshot_quiet_window"
_SEED_SKIP_SNAPSHOT_RUNNING = "snapshot_running"
_SEED_SKIP_NO_SOURCE = "no_source"


def _quiet_window_defer(
    repo: Repository,
    logger: JsonLogger,
    name: str,
    *,
    started_at: int,
) -> TaskControl:
    """静默窗内不执行：记 skip（非成功），并排到 01:00 之后。"""
    finished = int(time.time())
    logger.info(f"{name}.skip", reason=_QUIET_SKIP_REASON)
    next_run = next_after_snapshot_quiet_window(float(finished))
    _record_task(
        repo,
        name,
        ok=False,
        skipped=True,
        started_at=started_at,
        finished_at=finished,
        payload={"skipped": True, "reason": _QUIET_SKIP_REASON},
        error=_QUIET_SKIP_REASON,
        next_due_at=next_run,
    )
    return TaskControl(
        ok=False,
        next_run=next_run,
        error=_QUIET_SKIP_REASON,
    )


def _retry_soon(now: float | None = None) -> float:
    return (time.time() if now is None else now) + _TASK_RETRY_SECONDS


def _seed_retry_soon(now: float | None = None) -> float:
    return (time.time() if now is None else now) + _SEED_RETRY_SECONDS


def _defer_snapshot_seed(
    repo: Repository,
    logger: JsonLogger,
    *,
    started_at: int,
    reason: str,
    finished_at: int | None = None,
) -> TaskControl:
    """seed 因 snapshot 占用 / no_source 延后：记 skip（非成功），10 分钟后再试。"""
    finished = int(finished_at if finished_at is not None else time.time())
    logger.info("snapshot_seed.skip", reason=reason)
    next_run = _seed_retry_soon(float(finished))
    _record_task(
        repo,
        "snapshot_seed",
        ok=False,
        skipped=True,
        started_at=started_at,
        finished_at=finished,
        payload={"skipped": True, "reason": reason},
        error=reason,
        next_due_at=next_run,
    )
    return TaskControl(
        ok=False,
        next_run=next_run,
        error=reason,
    )


def _initial_daily_next_run(
    repo: Repository,
    name: str,
    at: str,
    *,
    now: float | None = None,
    retry_seconds: float | None = None,
) -> float:
    """进程启动时日更任务的下次运行：优先 ``task:state.due_at``。

    无状态时从 ``task:last`` / seed 标记冷启动回填；不得返回 ``0`` 误伤。
    失败且 ``due_at`` 已过：立刻补跑，不退回次日墙钟。
    """
    current = time.time() if now is None else float(now)
    wall = next_daily_run(current, at)
    retry = float(retry_seconds) if retry_seconds is not None else _TASK_RETRY_SECONDS
    return task_state_mod.startup_next_run(
        repo,
        name,
        wall_due_at=wall,
        now=current,
        slot_hhmm=at,
        retry_seconds=retry,
    )


def _reached_daily_slot(at: str, now: float | None = None) -> bool:
    """当前上海时间是否已到达（含）今日 ``at`` 墙钟点。"""
    current = shanghai_datetime(now)
    hour, minute = parse_hhmm(at)
    slot = current.replace(hour=hour, minute=minute, second=0, microsecond=0)
    return current >= slot


def _today_slot_ts(at: str, now: float | None = None) -> float:
    """今日 ``at`` 墙钟点的时间戳（不论是否已过）。"""
    current = shanghai_datetime(now)
    hour, minute = parse_hhmm(at)
    return current.replace(hour=hour, minute=minute, second=0, microsecond=0).timestamp()


def _last_success_started_at(repo: Repository, name: str) -> float | None:
    """最近一次成功完成记录的 ``started_at``（不用 finished_at）。"""
    try:
        raw = repo.kv_get(f"task:last:{name}")
    except Exception:  # noqa: BLE001
        return None
    if not raw:
        return None
    try:
        data = json.loads(raw)
    except (TypeError, ValueError, json.JSONDecodeError):
        return None
    if not isinstance(data, dict):
        return None
    if not data.get("ok") or data.get("skipped"):
        return None
    try:
        return float(data["started_at"])
    except (KeyError, TypeError, ValueError):
        return None


def _off_plan_completion(
    repo: Repository,
    name: str,
    at: str,
    now: float | None = None,
) -> bool:
    """成功完成的 started_at 早于当日计划点 → 非计划内，到点仍应再跑。

    产品规则：只用完成记录的 **started_at** 与今日 ``at`` 比较，禁止 finished_at。
    """
    started = _last_success_started_at(repo, name)
    if started is None:
        return False
    return float(started) < _today_slot_ts(at, now)


def _should_force_daily_seal(
    *,
    seeded: bool,
    stat_date: str,
    today: str,
    slot_hhmm: str,
    now: float,
) -> bool:
    """seed 壳是否 force 打穿：跨天未完成立即 force；同日须已到计划点。"""
    if not seeded:
        return False
    if stat_date < today:
        return True
    return _reached_daily_slot(slot_hhmm, now)


def _evening_due_at(now: float, snapshot_at: str) -> float:
    """尚未到当日 snapshot_at 则挂今晚，否则下一档墙钟。"""
    hour, minute = parse_hhmm(snapshot_at)
    dt = shanghai_datetime(now)
    slot = dt.replace(hour=hour, minute=minute, second=0, microsecond=0).timestamp()
    if slot > now:
        return float(slot)
    return float(next_daily_run(now, snapshot_at))


def _enqueue_evening_seal(
    repo: Repository,
    *,
    stat_date: str,
    snapshot_at: str,
    now: float,
) -> None:
    """seed 成功后：当晚 snapshot / fans 记 pending，due_at=当日 snapshot_at。"""
    current = shanghai_datetime(now)
    hour, minute = parse_hhmm(snapshot_at)
    slot_ts = current.replace(
        hour=hour, minute=minute, second=0, microsecond=0,
    ).timestamp()
    due = slot_ts if slot_ts > now else now
    task_state_mod.ensure_pending(
        repo, "snapshot", due_at=due, stat_date=stat_date, now=now,
    )
    task_state_mod.ensure_pending(
        repo, "fans", due_at=due, stat_date=stat_date, now=now,
    )


def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient:
    return BilibiliClient(
        cookie=credentials.cookie,
        csrf=credentials.csrf,
        mid=credentials.mid,
        pacer=pacer,
        retries=retries,
        logger=logger,
    )


def cmd_init(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        for folder in cfg.folders:
            repo.ensure_folder(folder)
        seeded = repo.seed_blacklist(cfg.up_blacklist_seed)
        _print({
            "ok": True,
            "schema_version": db.schema_version,
            "database": str(cfg.storage.database_path),
            "blacklist_seeded": seeded,
            **repo.counts(),
        })
    finally:
        db.close()
    return 0


def cmd_ingest(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        client = _client(cfg.bilibili.read, logger)
        up_profile = UpProfileService(repo, lambda: client, logger=logger)
        service = LibraryService(
            store, repo, lambda: client,
            up_profile_service=up_profile, logger=logger,
        )
        text = " ".join(args.refs)
        result = service.ingest_text(text)
        payload = result.as_dict()
        payload["database"] = str(cfg.storage.database_path)
        _record_task(
            repo, "ingest", ok=bool(result.results) and not any(
                item.state == "error" for item in result.results
            ),
            payload=payload,
        )
        _print(payload)
        return 0 if result.results and not any(
            item.state == "error" for item in result.results
        ) else 1
    finally:
        db.close()


def cmd_backfill_followers(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        client = _client(
            cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=2
        )
        service = UpProfileService(repo, lambda: client, logger=logger)
        started = int(time.time())
        result = service.backfill(limit=args.limit, dry_run=args.dry_run)
        payload = result.as_dict()
        payload["dry_run"] = bool(args.dry_run)
        payload["coverage"] = repo.up_follower_coverage()
        payload["database"] = str(cfg.storage.database_path)
        _record_task(
            repo, "backfill_followers", ok=result.failed == 0,
            started_at=started, finished_at=int(time.time()), payload=payload,
            error="；".join(result.errors) if result.errors else None,
        )
        _print(payload)
        return 0 if result.failed == 0 else 1
    finally:
        db.close()


def cmd_snapshot(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        client = _client(
            cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=1
        )
        service = SnapshotService(store, repo, client, logger=logger)
        started = int(time.time())
        hh, mm = parse_hhmm(cfg.schedule.snapshot_at)
        snapshot_at = f"{hh:02d}:{mm:02d}"
        _mark_task_running(
            repo, "snapshot", started_at=started,
            stat_date=shanghai_date(float(started)),
        )
        try:
            # CLI --force：全量重采（keep_scraped=False）；非 force 走常规门闩
            video_summary = service.run_if_needed(
                force=args.force,
                keep_scraped=False if args.force else None,
            )
        except Exception as exc:  # noqa: BLE001
            finished = int(time.time())
            _record_task(
                repo, "snapshot", ok=False, started_at=started,
                finished_at=finished, error=str(exc),
                next_due_at=_retry_soon(float(finished)),
                stat_date=shanghai_date(float(started)),
            )
            raise
        video_payload = video_summary.as_dict()
        video_payload["database"] = str(cfg.storage.database_path)
        video_payload["snapshot_counts"] = repo.snapshot_counts(video_summary.stat_date)
        video_payload["counts"] = repo.counts()
        video_finished = int(video_summary.finished_at or time.time())
        video_ok = bool(video_summary.completed and not video_summary.backup_error)
        if video_summary.reason == "seeded_pending_seal":
            evening = _evening_due_at(float(started), snapshot_at)
            _enqueue_evening_seal(
                repo,
                stat_date=video_summary.stat_date,
                snapshot_at=snapshot_at,
                now=float(video_finished),
            )
            _record_task(
                repo, "snapshot", ok=False, skipped=True,
                started_at=int(video_summary.started_at or started),
                finished_at=video_finished,
                payload=video_payload,
                error="seeded_pending_seal",
                next_due_at=evening,
                stat_date=video_summary.stat_date,
            )
        elif video_ok:
            _record_task(
                repo, "snapshot", ok=True,
                started_at=int(video_summary.started_at or started),
                finished_at=video_finished,
                payload=video_payload,
                error=video_summary.backup_error,
                next_due_at=next_daily_run(float(video_finished), snapshot_at),
                stat_date=video_summary.stat_date,
            )
        else:
            _record_task(
                repo, "snapshot", ok=False,
                started_at=int(video_summary.started_at or started),
                finished_at=video_finished,
                payload=video_payload,
                error=video_summary.backup_error or "snapshot_incomplete",
                next_due_at=_retry_soon(float(video_finished)),
                stat_date=video_summary.stat_date,
            )

        fans_started = int(time.time())
        _mark_task_running(
            repo, "fans", started_at=fans_started,
            stat_date=video_summary.stat_date,
        )
        try:
            fans_summary = service.run_fans_if_needed(
                date=video_summary.stat_date, force=args.force,
            )
        except Exception as exc:  # noqa: BLE001
            finished = int(time.time())
            _record_task(
                repo, "fans", ok=False, started_at=fans_started,
                finished_at=finished, error=str(exc),
                next_due_at=_retry_soon(float(finished)),
                stat_date=video_summary.stat_date,
            )
            raise
        fans_payload = fans_summary.as_dict()
        fans_finished = int(fans_summary.finished_at or time.time())
        fans_ok = fans_summary.fans_failed == 0
        if fans_summary.reason == "seeded_pending_seal":
            evening = _evening_due_at(float(fans_started), snapshot_at)
            _record_task(
                repo, "fans", ok=False, skipped=True,
                started_at=int(fans_summary.started_at or fans_started),
                finished_at=fans_finished,
                payload=fans_payload,
                error="seeded_pending_seal",
                next_due_at=evening,
                stat_date=video_summary.stat_date,
            )
        elif fans_ok:
            _record_task(
                repo, "fans", ok=True,
                started_at=int(fans_summary.started_at or fans_started),
                finished_at=fans_finished,
                payload=fans_payload,
                next_due_at=next_daily_run(float(fans_finished), snapshot_at),
                stat_date=video_summary.stat_date,
            )
        else:
            _record_task(
                repo, "fans", ok=False,
                started_at=int(fans_summary.started_at or fans_started),
                finished_at=fans_finished,
                payload=fans_payload,
                error="；".join(
                    err for err in fans_summary.errors if str(err).startswith("fans ")
                ) or None,
                next_due_at=_retry_soon(float(fans_finished)),
                stat_date=video_summary.stat_date,
            )
        payload = {
            **video_payload,
            "fans_ok": fans_summary.fans_ok,
            "fans_failed": fans_summary.fans_failed,
            "fans_skipped": fans_summary.fans_skipped,
        }
        _print(payload)
        return 0 if video_ok and fans_ok else 1
    finally:
        db.close()


def cmd_snapshot_seed(args, logger: JsonLogger) -> int:
    """离线克隆：``python app.py snapshot-seed [--date YYYY-MM-DD]``。

    与调度器同一套规则：全站 ``snapshot`` 运行中或 ``no_source`` 时不记成功，
    返回码非 0，便于 ``docker compose exec re-xxm python app.py snapshot-seed`` 排查。
    """
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        target = args.date
        if target is not None and not is_date(target):
            raise ConfigError(f"日期格式必须是 YYYY-MM-DD: {target!r}")
        started = int(time.time())
        if _is_task_running(repo, "snapshot"):
            payload = {
                "ok": False,
                "skipped": True,
                "reason": _SEED_SKIP_SNAPSHOT_RUNNING,
                "stat_date": target or shanghai_date(),
                "database": str(cfg.storage.database_path),
            }
            finished = int(time.time())
            _record_task(
                repo,
                "snapshot_seed",
                ok=False,
                skipped=True,
                started_at=started,
                finished_at=finished,
                payload=payload,
                error=_SEED_SKIP_SNAPSHOT_RUNNING,
                next_due_at=_seed_retry_soon(float(finished)),
                stat_date=payload["stat_date"],
            )
            _print(payload)
            return 1
        # seed 只读本地库，不调 B 站；占位 client 满足构造约束。
        service = SnapshotService(store, repo, object(), logger=logger)
        seed_date = target or shanghai_date()
        _mark_task_running(
            repo, "snapshot_seed", started_at=started, stat_date=seed_date,
        )
        try:
            summary = service.run_seed_from_last_completed(date=target)
        except Exception as exc:  # noqa: BLE001
            finished = int(time.time())
            _record_task(
                repo, "snapshot_seed", ok=False, started_at=started,
                finished_at=finished, error=str(exc),
                next_due_at=_seed_retry_soon(float(finished)),
                stat_date=seed_date,
            )
            raise
        payload = summary.as_dict()
        payload["database"] = str(cfg.storage.database_path)
        finished = int(summary.finished_at or time.time())
        hh, mm = parse_hhmm(cfg.schedule.seed_at)
        seed_at = f"{hh:02d}:{mm:02d}"
        sh, sm = parse_hhmm(cfg.schedule.snapshot_at)
        snapshot_at = f"{sh:02d}:{sm:02d}"
        if summary.reason == _SEED_SKIP_NO_SOURCE:
            _record_task(
                repo,
                "snapshot_seed",
                ok=False,
                skipped=True,
                started_at=int(summary.started_at or started),
                finished_at=finished,
                payload={**payload, "skipped": True, "reason": _SEED_SKIP_NO_SOURCE},
                error=_SEED_SKIP_NO_SOURCE,
                next_due_at=_seed_retry_soon(float(finished)),
                stat_date=summary.stat_date or seed_date,
            )
            _print(payload)
            return 1
        next_due = next_daily_run(float(finished), seed_at)
        _record_task(
            repo,
            "snapshot_seed",
            ok=True,
            started_at=int(summary.started_at or started),
            finished_at=finished,
            payload=payload,
            next_due_at=next_due,
            stat_date=summary.stat_date or seed_date,
        )
        if summary.reason == "seeded":
            _enqueue_evening_seal(
                repo,
                stat_date=summary.stat_date or seed_date,
                snapshot_at=snapshot_at,
                now=float(finished),
            )
        _print(payload)
        return 0
    finally:
        db.close()


def _telegram(cfg, logger: JsonLogger):
    if not cfg.telegram.enabled:
        return None
    if not cfg.telegram.bot_token or not cfg.telegram.chat_id:
        raise ConfigError(
            "telegram.enabled=true 时必须配置 bot_token 和 chat_id"
        )
    return TelegramClient(
        cfg.telegram.bot_token,
        cfg.telegram.chat_id,
        logger=logger,
    )


def cmd_discover(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=True)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        provider = ClientProvider(
            store, "read", lambda credentials: _client(credentials, logger)
        )
        service = DiscoveryService(
            store,
            repo,
            SearchClient(provider.get),
            notifier=None if args.no_notify else _telegram(cfg, logger),
            logger=logger,
        )
        started = int(time.time())
        interval = float(cfg.schedule.discover_minutes * 60)
        _mark_task_running(repo, "discovery", started_at=started)
        try:
            summary = service.run()
        except Exception as exc:  # noqa: BLE001
            finished = int(time.time())
            _record_task(
                repo, "discovery", ok=False, started_at=started,
                finished_at=finished, error=str(exc),
                next_due_at=_retry_soon(float(finished)),
            )
            raise
        payload = summary.as_dict()
        payload["candidate_total"] = len(repo.list_candidates())
        payload["database"] = str(cfg.storage.database_path)
        finished = int(summary.finished_at or time.time())
        ok = not summary.errors
        next_due = (
            interval_next_run(float(finished), interval) if ok
            else _retry_soon(float(finished))
        )
        _record_task(
            repo, "discovery", ok=ok,
            started_at=int(summary.started_at or started),
            finished_at=finished,
            payload=payload,
            error="；".join(summary.errors) if summary.errors else None,
            next_due_at=next_due,
        )
        _print(payload)
        return 0 if not summary.errors else 1
    finally:
        db.close()


def cmd_candidates(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        rows = repo.list_candidates(args.decision)
        _print({
            "ok": True,
            "count": len(rows),
            "candidates": rows,
            "database": str(cfg.storage.database_path),
        })
        return 0
    finally:
        db.close()


def cmd_auth(args, logger: JsonLogger) -> int:
    if args.timeout < 1:
        raise ConfigError("auth --timeout 必须至少为 1 秒")
    return AuthService(Path(args.config), account=args.account, logger=logger).run(
        timeout=args.timeout, no_browser=args.no_browser
    )


def _auth_check(cfg, account: str, logger: JsonLogger) -> dict[str, object]:
    credentials = getattr(cfg.bilibili, account)
    if not credentials.cookie:
        return {"account": account, "ok": False, "is_login": False, "message": "未配置 cookie"}
    state = _client(credentials, logger, retries=1).check_login()
    return {
        "account": account,
        "ok": state.is_login,
        "is_login": state.is_login,
        "mid": state.mid,
        "uname": state.uname,
        "code": state.code,
        "message": state.message,
    }


def cmd_auth_check(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    if args.account != "all":
        result = _auth_check(cfg, args.account, logger)
        _print(result)
        return 0 if result["ok"] else 1
    results = [_auth_check(cfg, account, logger) for account in ("read", "write")]
    _print({"ok": all(result["ok"] for result in results), "accounts": results})
    return 0 if all(result["ok"] for result in results) else 1


def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
    store = ConfigStore(args.config, require_auth=True)
    cfg = store.get()
    db = _open_db(cfg, logger)
    last: dict[str, object] = {}
    web_server = None
    try:
        repo = Repository(db)
        if cfg.up_blacklist_seed:
            repo.seed_blacklist(cfg.up_blacklist_seed)
        read_provider = ClientProvider(
            store, "read", lambda credentials: _client(credentials, logger)
        )
        snapshot_provider = ClientProvider(
            store,
            "read",
            lambda credentials: _client(
                credentials, logger, pacer=Pacer(1.0, 1.6), retries=1
            ),
        )
        snapshot_service = SnapshotService(
            store, repo, client_provider=snapshot_provider, logger=logger
        )
        up_profile_service = UpProfileService(repo, read_provider.get, logger=logger)
        library_service = LibraryService(
            store,
            repo,
            read_provider,
            up_profile_service=up_profile_service,
            logger=logger,
        )
        notifier = _telegram(cfg, logger)
        discovery_service = DiscoveryService(
            store,
            repo,
            SearchClient(read_provider.get),
            notifier=notifier,
            logger=logger,
        )

        hh, mm = parse_hhmm(cfg.schedule.snapshot_at)
        snapshot_at = f"{hh:02d}:{mm:02d}"
        sh, sm = parse_hhmm(cfg.schedule.seed_at)
        seed_at = f"{sh:02d}:{sm:02d}"
        recent_minutes = int(store.get().schedule.recent_snapshot_minutes)
        discover_minutes = int(store.get().schedule.discover_minutes)

        def run_discover():
            started = int(time.time())
            if in_snapshot_quiet_window(float(started)):
                return _quiet_window_defer(
                    repo, logger, "discovery", started_at=started,
                )
            _mark_task_running(repo, "discovery", started_at=started)
            interval = float(store.get().schedule.discover_minutes * 60)
            try:
                summary = discovery_service.run()
            except Exception as exc:  # noqa: BLE001  记录后补跑，不抛给次日
                finished = int(time.time())
                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "discovery", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
                    next_due_at=next_run,
                )
                return TaskControl(
                    ok=False, next_run=next_run, error=str(exc),
                )
            payload = summary.as_dict()
            ok = not summary.errors
            finished = int(summary.finished_at or time.time())
            next_run = (
                interval_next_run(float(finished), interval) if ok
                else _retry_soon(float(finished))
            )
            _record_task(
                repo, "discovery", ok=ok,
                started_at=summary.started_at,
                finished_at=finished,
                payload=payload,
                error="；".join(summary.errors) if summary.errors else None,
                next_due_at=next_run,
            )
            last["discover"] = summary
            logger.info("discovery.summary", **payload)
            return TaskControl(ok=ok, next_run=next_run)

        def run_cleanup():
            started = int(time.time())
            if in_snapshot_quiet_window(float(started)):
                return _quiet_window_defer(
                    repo, logger, "candidate_cleanup", started_at=started,
                )
            try:
                removed = repo.delete_expired_candidates(
                    ttl_days=cfg.schedule.candidate_ttl_days
                )
                expired = repo.expire_fav_actions()
            except Exception as exc:  # noqa: BLE001  记录后约 5 分钟补跑
                finished = int(time.time())
                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "candidate_cleanup", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
                    next_due_at=next_run,
                )
                return TaskControl(
                    ok=False, next_run=next_run, error=str(exc),
                )
            if removed or expired:
                repo.bump_cache_version("videos")
            finished = int(time.time())
            logger.info(
                "cleanup.done",
                candidates_removed=removed, fav_actions_expired=expired,
            )
            next_run = interval_next_run(float(finished), 86400.0)
            _record_task(
                repo, "candidate_cleanup", ok=True, started_at=started,
                finished_at=finished,
                payload={
                    "candidates_removed": removed,
                    "fav_actions_expired": expired,
                },
                next_due_at=next_run,
            )
            return TaskControl(ok=True, next_run=next_run)

        def run_snapshot():
            """全站快照：不因其它任务 running 而 skip；失败/未完成则短间隔补跑。"""
            started = int(time.time())
            today = shanghai_date(float(started))
            stat_date = task_state_mod.resume_stat_date(
                repo, "snapshot", fallback=today,
            )
            _mark_task_running(
                repo, "snapshot", started_at=started, stat_date=stat_date,
            )
            try:
                # seed 壳 force：跨天未完成（stat_date < today）立刻打穿；
                # 同日仅到达 snapshot_at 后才 force。due_at 管「会不会跑」。
                seeded = repo.kv_get(SnapshotService.seeded_key(stat_date)) == "1"
                force_seed = _should_force_daily_seal(
                    seeded=seeded,
                    stat_date=stat_date,
                    today=today,
                    slot_hhmm=snapshot_at,
                    now=float(started),
                )
                summary = snapshot_service.run_if_needed(
                    date=stat_date,
                    force=force_seed,
                    keep_scraped=True if force_seed else None,
                )
                # 日更规则：already_completed 但完成 started_at 早于今日计划点
                # → 非计划内，到点仍按计划再跑（全量，不用 finished_at）。
                if (
                    summary.already_completed
                    and _reached_daily_slot(snapshot_at, float(started))
                    and _off_plan_completion(
                        repo, "snapshot", snapshot_at, float(started),
                    )
                ):
                    summary = snapshot_service.run_if_needed(
                        date=stat_date, force=True, keep_scraped=False,
                    )
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "snapshot", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
                    next_due_at=next_run, stat_date=stat_date,
                )
                return TaskControl(
                    ok=False, next_run=next_run, error=str(exc),
                )
            payload = summary.as_dict()
            ok = bool(summary.completed and not summary.backup_error)
            finished = int(summary.finished_at or time.time())
            # 白天误碰到 seed 壳：延到当晚 seal，禁止 5 分钟空转。
            if (
                not ok
                and getattr(summary, "reason", None) == "seeded_pending_seal"
            ):
                current = shanghai_datetime(float(finished))
                hour, minute = parse_hhmm(snapshot_at)
                slot_ts = current.replace(
                    hour=hour, minute=minute, second=0, microsecond=0,
                ).timestamp()
                next_run = slot_ts if slot_ts > float(finished) else _retry_soon(
                    float(finished),
                )
                _record_task(
                    repo, "snapshot", ok=False, skipped=True,
                    started_at=summary.started_at, finished_at=finished,
                    payload=payload, error="seeded_pending_seal",
                    next_due_at=next_run,
                    stat_date=summary.stat_date or stat_date,
                )
                last["snapshot"] = summary
                logger.info("snapshot.summary", **payload)
                return TaskControl(
                    ok=False, next_run=next_run, error="seeded_pending_seal",
                )
            next_run = (
                next_daily_run(float(finished), snapshot_at) if ok
                else _retry_soon(float(finished))
            )
            progress = {
                "total": summary.total,
                "success": summary.success,
                "failed": summary.failed,
                "skipped": summary.skipped,
            }
            _record_task(
                repo, "snapshot", ok=ok, started_at=summary.started_at,
                finished_at=finished,
                payload=payload, error=summary.backup_error,
                next_due_at=next_run,
                stat_date=summary.stat_date or stat_date,
                progress=progress,
            )
            last["snapshot"] = summary
            logger.info("snapshot.summary", **payload)
            return TaskControl(ok=ok, next_run=next_run)

        def run_snapshot_recent():
            started = int(time.time())
            if in_snapshot_quiet_window(float(started)):
                return _quiet_window_defer(
                    repo, logger, "snapshot_recent", started_at=started,
                )
            _mark_task_running(repo, "snapshot_recent", started_at=started)
            interval = float(store.get().schedule.recent_snapshot_minutes * 60)
            try:
                summary = snapshot_service.run_recent_videos()
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "snapshot_recent", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
                    next_due_at=next_run,
                )
                return TaskControl(
                    ok=False, next_run=next_run, error=str(exc),
                )
            payload = summary.as_dict()
            ok = summary.failed == 0
            error = "；".join(summary.errors) if summary.errors else None
            finished = int(summary.finished_at or time.time())
            next_run = (
                interval_next_run(float(finished), interval) if ok
                else _retry_soon(float(finished))
            )
            _record_task(
                repo, "snapshot_recent", ok=ok, started_at=summary.started_at,
                finished_at=finished,
                payload=payload, error=error,
                next_due_at=next_run,
            )
            last["snapshot_recent"] = summary
            logger.info("snapshot.recent_summary", **payload)
            return TaskControl(ok=ok, next_run=next_run)

        def run_fans():
            """粉丝更新短耗时：不避开任何任务，可与全量并行。"""
            started = int(time.time())
            today = shanghai_date(float(started))
            stat_date = task_state_mod.resume_stat_date(
                repo, "fans", fallback=today,
            )
            _mark_task_running(
                repo, "fans", started_at=started, stat_date=stat_date,
            )
            try:
                seeded = (
                    repo.kv_get(SnapshotService.fans_seeded_key(stat_date)) == "1"
                )
                force_seed = _should_force_daily_seal(
                    seeded=seeded,
                    stat_date=stat_date,
                    today=today,
                    slot_hhmm=snapshot_at,
                    now=float(started),
                )
                summary = snapshot_service.run_fans_if_needed(
                    date=stat_date, force=force_seed,
                )
                if (
                    summary.already_completed
                    and _reached_daily_slot(snapshot_at, float(started))
                    and _off_plan_completion(
                        repo, "fans", snapshot_at, float(started),
                    )
                ):
                    summary = snapshot_service.run_fans_if_needed(
                        date=stat_date, force=True,
                    )
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
                next_run = _retry_soon(float(finished))
                _record_task(
                    repo, "fans", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
                    next_due_at=next_run, stat_date=stat_date,
                )
                return TaskControl(
                    ok=False, next_run=next_run, error=str(exc),
                )
            payload = summary.as_dict()
            finished = int(summary.finished_at or time.time())
            if getattr(summary, "reason", None) == "seeded_pending_seal":
                current = shanghai_datetime(float(finished))
                hour, minute = parse_hhmm(snapshot_at)
                slot_ts = current.replace(
                    hour=hour, minute=minute, second=0, microsecond=0,
                ).timestamp()
                next_run = slot_ts if slot_ts > float(finished) else _retry_soon(
                    float(finished),
                )
                _record_task(
                    repo, "fans", ok=False, skipped=True,
                    started_at=summary.started_at, finished_at=finished,
                    payload=payload, error="seeded_pending_seal",
                    next_due_at=next_run,
                    stat_date=summary.stat_date or stat_date,
                )
                last["fans"] = summary
                logger.info("fans.summary", **payload)
                return TaskControl(
                    ok=False, next_run=next_run, error="seeded_pending_seal",
                )
            ok = summary.fans_failed == 0
            error = "；".join(
                err for err in summary.errors if str(err).startswith("fans ")
            ) or None
            next_run = (
                next_daily_run(float(finished), snapshot_at) if ok
                else _retry_soon(float(finished))
            )
            _record_task(
                repo, "fans", ok=ok, started_at=summary.started_at,
                finished_at=finished,
                payload=payload, error=error,
                next_due_at=next_run,
                stat_date=summary.stat_date or stat_date,
            )
            last["fans"] = summary
            logger.info("fans.summary", **payload)
            return TaskControl(ok=ok, next_run=next_run)

        def run_snapshot_seed():
            started = int(time.time())
            today = shanghai_date(float(started))
            # 全站快照占用时不执行、不记成功；10 分钟后重试直到真正 seed 成功。
            if _is_task_running(repo, "snapshot"):
                return _defer_snapshot_seed(
                    repo,
                    logger,
                    started_at=started,
                    reason=_SEED_SKIP_SNAPSHOT_RUNNING,
                )
            _mark_task_running(
                repo, "snapshot_seed", started_at=started, stat_date=today,
            )
            try:
                summary = snapshot_service.run_seed_from_last_completed(date=today)
            except Exception as exc:  # noqa: BLE001
                finished = int(time.time())
                next_run = _seed_retry_soon(float(finished))
                _record_task(
                    repo, "snapshot_seed", ok=False, started_at=started,
                    finished_at=finished, error=str(exc),
                    next_due_at=next_run, stat_date=today,
                )
                return TaskControl(
                    ok=False,
                    next_run=next_run,
                    error=str(exc),
                )
            payload = summary.as_dict()
            finished = int(summary.finished_at or time.time())
            last["snapshot_seed"] = summary
            logger.info("snapshot.seed_summary", **payload)
            if summary.reason == _SEED_SKIP_NO_SOURCE:
                return _defer_snapshot_seed(
                    repo,
                    logger,
                    started_at=int(summary.started_at or started),
                    reason=_SEED_SKIP_NO_SOURCE,
                    finished_at=finished,
                )
            # already_completed / already_seeded / seeded → 成功，推到次日 seed_at
            next_run = next_daily_run(float(finished), seed_at)
            _record_task(
                repo, "snapshot_seed", ok=True, started_at=summary.started_at,
                finished_at=finished,
                payload=payload,
                next_due_at=next_run,
                stat_date=summary.stat_date or today,
            )
            if summary.reason == "seeded":
                _enqueue_evening_seal(
                    repo,
                    stat_date=summary.stat_date or today,
                    snapshot_at=snapshot_at,
                    now=float(finished),
                )
            return TaskControl(ok=True, next_run=next_run)

        def run_config_watch():
            before = store.revision
            store.get()
            if store.revision != before:
                logger.info("config.reloaded", revision=store.revision)
            return {"revision": store.revision}

        # --once：next_run=0 立刻跑；常驻：一律从 task:state.due_at 恢复，禁止日更 next_run=0。
        now_ts = time.time()
        if args.once:
            seed_next = snapshot_next = fans_next = 0.0
            recent_next = discover_next = cleanup_next = 0.0
        else:
            seed_next = _initial_daily_next_run(
                repo, "snapshot_seed", seed_at, now=now_ts,
            )
            snapshot_next = _initial_daily_next_run(
                repo, "snapshot", snapshot_at, now=now_ts,
            )
            fans_next = _initial_daily_next_run(
                repo, "fans", snapshot_at, now=now_ts,
            )
            recent_next = task_state_mod.startup_next_run(
                repo, "snapshot_recent", wall_due_at=now_ts, now=now_ts,
            )
            discover_next = task_state_mod.startup_next_run(
                repo, "discovery", wall_due_at=now_ts, now=now_ts,
            )
            cleanup_next = task_state_mod.startup_next_run(
                repo, "candidate_cleanup", wall_due_at=now_ts, now=now_ts,
            )
        def _due_reader(task_name: str) -> float | None:
            state = task_state_mod.load_state(repo, task_name)
            if state is None:
                return None
            try:
                due = float(state.due_at or 0)
            except (TypeError, ValueError):
                return None
            return due if due > 0 else None

        scheduler = Scheduler(
            [
                ScheduledTask(
                    "snapshot_seed",
                    86400.0,
                    run_snapshot_seed,
                    next_run=seed_next,
                ),
                ScheduledTask(
                    "snapshot",
                    86400.0,
                    run_snapshot,
                    next_run=snapshot_next,
                ),
                ScheduledTask(
                    "fans",
                    86400.0,
                    run_fans,
                    next_run=fans_next,
                ),
                ScheduledTask(
                    "snapshot_recent",
                    float(recent_minutes * 60),
                    run_snapshot_recent,
                    next_run=recent_next,
                ),
                ScheduledTask(
                    "discovery",
                    float(discover_minutes * 60),
                    run_discover,
                    next_run=discover_next,
                ),
                ScheduledTask(
                    "candidate_cleanup",
                    86400.0,
                    run_cleanup,
                    next_run=cleanup_next,
                ),
                ScheduledTask("config_watch", 60.0, run_config_watch),
            ],
            logger=logger,
            due_reader=_due_reader,
        )

        if serve_web and not args.once:
            from .web.server import create_server, serve_forever

            web_server = create_server(store, db, logger=logger)
            threading.Thread(
                target=lambda: serve_forever(web_server, logger=logger),
                name="web-server",
                daemon=True,
            ).start()
        if notifier is not None and not args.once:
            poller = TelegramPoller(notifier, library_service, logger=logger)
            threading.Thread(
                target=poller.run_forever,
                name="telegram-poller",
                daemon=True,
            ).start()
            logger.info("telegram.polling_started")
        if args.once:
            results = scheduler.run_due()
            _print([result.as_dict() for result in results])
        else:
            logger.info(
                "scheduler.start",
                snapshot_at=snapshot_at,
                seed_at=seed_at,
            )
            scheduler.run_forever()

        snapshot_summary = last.get("snapshot")
        fans_summary = last.get("fans")
        discover_summary = last.get("discover")
        snapshot_ok = snapshot_summary is None or getattr(snapshot_summary, "completed", True)
        fans_ok = fans_summary is None or getattr(fans_summary, "fans_failed", 0) == 0
        discover_ok = discover_summary is None or not getattr(discover_summary, "errors", [])
        return 0 if snapshot_ok and fans_ok and discover_ok else 1
    finally:
        if web_server is not None:
            try:
                web_server.shutdown()
            finally:
                web_server.server_close()
        db.close()


def _assert_preview_database(path: Path) -> None:
    import sqlite3

    if not path.exists():
        raise ConfigError(f"预览数据库不存在: {path}")
    uri = f"file:{path.resolve().as_posix()}?mode=ro"
    try:
        conn = sqlite3.connect(uri, uri=True)
    except sqlite3.Error as exc:
        raise ConfigError(f"无法读取预览数据库: {exc}") from exc
    try:
        row = conn.execute(
            "SELECT v FROM kv WHERE k = 'preview:seeded'"
        ).fetchone()
    except sqlite3.Error as exc:
        raise ConfigError(f"目标数据库不是预览库: {exc}") from exc
    finally:
        conn.close()
    if row is None or str(row[0]) != "1":
        raise ConfigError(
            "目标数据库缺少 preview:seeded=1 标记；请先运行 "
            "tools/seed_preview_db.py，或去掉 --preview"
        )


def cmd_serve(args, logger: JsonLogger) -> int:
    from .web.server import create_server, serve_forever

    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    if args.preview:
        _assert_preview_database(cfg.storage.database_path)
    db = _open_db(cfg, logger)
    try:
        server = create_server(store, db, preview=args.preview, logger=logger)
        serve_forever(server, logger=logger)
        return 0
    finally:
        db.close()


def cmd_status(args, logger: JsonLogger) -> int:
    store = ConfigStore(args.config, require_auth=False)
    cfg = store.get()
    db = _open_db(cfg, logger)
    try:
        repo = Repository(db)
        today = shanghai_date()
        snapshot = repo.snapshot_counts(today)
        snapshot.update({
            "date": today,
            "last_attempt": repo.kv_get("snapshot:last_attempt"),
            "last_completed": repo.kv_get("snapshot:last_completed"),
            "auth_ok": repo.kv_get("auth:ok"),
            "auth_message": repo.kv_get("auth:message"),
            "backup_ok": repo.kv_get(f"snapshot:backup:{today}"),
            "backup_error": repo.kv_get(f"snapshot:backup_error:{today}"),
        })
        _print({
            "ok": True,
            "schema_version": db.schema_version,
            "database": str(cfg.storage.database_path),
            "configured_folders": [
                {
                    "media_id": row.media_id,
                    "year": row.year,
                    "title": row.title,
                    "media_count": row.media_count,
                    "last_ids_at": row.last_ids_at,
                    "last_full_at": row.last_full_at,
                    "enabled": row.enabled,
                }
                for row in repo.list_folders()
            ],
            "snapshot": snapshot,
            **repo.counts(),
        })
    finally:
        db.close()
    return 0


def main(argv: Sequence[str] | None = None) -> int:
    args = _build_parser().parse_args(list(argv) if argv is not None else None)
    logger = JsonLogger()
    try:
        if args.command is None:
            # 无子命令时启动完整服务：只读 Web + 调度 + Telegram。
            args.command = "run"
            args.once = False
            return cmd_run(args, logger, serve_web=True)
        if args.command == "serve":
            return cmd_serve(args, logger)
        if args.command == "init":
            return cmd_init(args, logger)
        if args.command == "snapshot":
            return cmd_snapshot(args, logger)
        if args.command == "snapshot-seed":
            return cmd_snapshot_seed(args, logger)
        if args.command == "ingest":
            return cmd_ingest(args, logger)
        if args.command == "backfill-followers":
            return cmd_backfill_followers(args, logger)
        if args.command == "discover":
            return cmd_discover(args, logger)
        if args.command == "candidates":
            return cmd_candidates(args, logger)
        if args.command == "auth":
            return cmd_auth(args, logger)
        if args.command == "auth-check":
            return cmd_auth_check(args, logger)
        if args.command == "run":
            return cmd_run(args, logger)
        if args.command == "status":
            return cmd_status(args, logger)
    except ConfigError as exc:
        logger.error("config.error", error=str(exc))
        return 2
    except KeyboardInterrupt:
        logger.warning("app.interrupted")
        return 130
    except Exception as exc:  # noqa: BLE001  CLI 边界统一记录
        logger.error("app.error", error=str(exc))
        return 2
    logger.error("app.unknown_command", command=getattr(args, "command", None))
    return 2


if __name__ == "__main__":
    sys.exit(main())