re-xxm/reaction_board/web/queries.py

# -*- coding: utf-8 -*-
"""只读看板查询层。

所有 SQL 都在这里集中，Web 层只负责参数解析与 JSON 序列化。查询不会写入
数据库，也不会访问 B 站。
"""

from __future__ import annotations

import calendar
import json
import math
import sqlite3
import time
from datetime import date, timedelta
from typing import Any, Iterable, Mapping, Sequence

from ..analytics import (
    DEFAULT_HEAT_WEIGHTS,
    METRICS,
    MIN_VIEW_FOR_GROWTH_RATE,
    compute_growth,
    smoothed_growth_rate,
    content_hours,
    eb_rate,
    effective_shrink_minutes,
    heat_scores,
    interactions_per_hour,
    interaction_rate,
    is_mature,
    normalize_cover,
    score_percentiles,
    views_per_hour,
)
from ..config import ConfigStore
from ..database import Database
from ..timeutil import day_bounds, next_daily_run, shanghai_date, shanghai_datetime
from .cache import DASHBOARD_CACHE, TTL_FANS, TTL_VIDEOS, build_cache_key

TRACKED_STATUSES = ("active", "unfaved")
VIDEO_SORTS = {
    "heat", "view", "like", "coin", "favorite", "share", "reply", "danmaku",
    "interaction", "growth", "growth_rate", "added", "published",
    "efficiency", "interaction_efficiency",
}
UP_SORTS = {
    "views", "count", "avg_views", "interaction", "avg_interaction", "latest",
    "efficiency", "interaction_efficiency", "duration",
}
# 搜索字段白名单：视频搜索匹配标题或 UP 名，UP 搜索只匹配 UP 名。
_Q_FIELDS = {"title": "v.title", "up_name": "v.up_name"}
CAPACITY = 1000
CAPACITY_WARNING = 950
TRACK_FANS_NAMES = {37754047: "咻咻满"}


def _chunks(values: Sequence[int], size: int = 400) -> Iterable[Sequence[int]]:
    for index in range(0, len(values), size):
        yield values[index:index + size]


def _growth_baseline_view(item: Mapping[str, Any]) -> int | None:
    """增速分母：当前播放减去区间增量。zero 基线的增量等于当前播放，结果为 0。"""
    growth = item.get("growth") or {}
    current = (item.get("metrics") or {}).get("view")
    delta = growth.get("delta")
    if current is None or delta is None:
        return None
    try:
        return int(current) - int(delta)
    except (TypeError, ValueError):
        return None


def _date_minus(day: str, days: int) -> str:
    return (date.fromisoformat(day) - timedelta(days=int(days))).isoformat()


def _date_range(start: str, end: str) -> list[str]:
    first = date.fromisoformat(start)
    last = date.fromisoformat(end)
    if last < first:
        return []
    return [
        (first + timedelta(days=offset)).isoformat()
        for offset in range((last - first).days + 1)
    ]


def _ts_date(timestamp: int | float | None) -> str | None:
    if timestamp is None:
        return None
    return shanghai_date(float(timestamp))


def _like_pattern(value: str) -> str:
    escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    return f"%{escaped}%"


def _metric_map(row: Mapping[str, Any] | None) -> dict[str, int] | None:
    if row is None:
        return None
    result: dict[str, int] = {}
    for metric in METRICS:
        value = row.get(metric)
        result[metric] = 0 if value is None else int(value)
    return result


def _safe_float(value: Any) -> float | None:
    if value is None or isinstance(value, bool):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _with_fan_deltas(
    rows: Sequence[Mapping[str, Any]], recent_start: str
) -> list[dict[str, Any]]:
    """给最近记录附加“相较上一成功日期”的增量。"""
    result: list[dict[str, Any]] = []
    last_ok: int | None = None
    for row in rows:
        day = str(row["stat_date"])
        ok = int(row.get("ok") or 0) == 1
        delta: int | None = None
        if ok:
            follower = int(row.get("follower") or 0)
            if last_ok is not None:
                delta = follower - last_ok
            last_ok = follower
        if day >= recent_start:
            result.append({**dict(row), "delta": delta})
    return result


def _daily_fan_series(
    rows: Sequence[Mapping[str, Any]], start: str, end: str
) -> tuple[list[str], list[int | None]]:
    by_date = {str(row["stat_date"]): row for row in rows}
    dates = _date_range(start, end)
    series: list[int | None] = []
    for day in dates:
        row = by_date.get(day)
        if row is None or int(row.get("ok") or 0) != 1:
            series.append(None)
        else:
            series.append(int(row["follower"] or 0))
    return dates, series


def _monthly_peak_series(
    rows: Sequence[Mapping[str, Any]], start: str, end: str
) -> tuple[list[str], list[int | None]]:
    """超过一年时按自然月聚合，每月取粉丝峰值所在的日期。"""
    start_day = date.fromisoformat(start)
    end_day = date.fromisoformat(end)
    peaks: dict[tuple[int, int], tuple[str, int]] = {}
    for row in rows:
        if int(row.get("ok") or 0) != 1:
            continue
        day = date.fromisoformat(str(row["stat_date"]))
        key = (day.year, day.month)
        value = int(row["follower"] or 0)
        current = peaks.get(key)
        if current is None or value > current[1] or (
            value == current[1] and str(row["stat_date"]) > current[0]
        ):
            peaks[key] = (str(row["stat_date"]), value)

    dates: list[str] = []
    series: list[int | None] = []
    year, month = start_day.year, start_day.month
    while (year, month) <= (end_day.year, end_day.month):
        peak = peaks.get((year, month))
        if peak is not None:
            dates.append(peak[0])
            series.append(peak[1])
        else:
            last_day = calendar.monthrange(year, month)[1]
            placeholder = date(year, month, last_day)
            if placeholder < start_day:
                placeholder = start_day
            if placeholder > end_day:
                placeholder = end_day
            dates.append(placeholder.isoformat())
            series.append(None)
        month += 1
        if month > 12:
            month = 1
            year += 1
    return dates, series


class DashboardQueries:
    def __init__(
        self,
        db: Database,
        config_store: ConfigStore,
        *,
        clock=time.time,
    ) -> None:
        self.db = db
        self.config_store = config_store
        self._clock = clock
        self._cache_scope = str(getattr(db, "path", "db"))

    def _cache_version(self, group: str) -> int:
        def run(conn: sqlite3.Connection) -> int:
            row = conn.execute(
                "SELECT v FROM kv WHERE k = ?", (f"cache:v:{group}",)
            ).fetchone()
            if row is None:
                return 0
            try:
                return int(str(row["v"]))
            except ValueError:
                return 0
        return self.db.read(run)

    def _cached(self, group: str, params: dict[str, Any], ttl: float, builder):
        version = self._cache_version(group)
        key = build_cache_key(self._cache_scope, group, version, params)
        hit = DASHBOARD_CACHE.get(key)
        if hit is not None:
            return hit
        value = builder()
        DASHBOARD_CACHE.set(key, ttl, value)
        return value

    # ------------------------------------------------------------ helpers

    def _now(self) -> float:
        return float(self._clock())

    def _today(self) -> str:
        return shanghai_date(self._now())

    @staticmethod
    def _latest_date(conn: sqlite3.Connection) -> str | None:
        """数据日期：仅取视频快照任务完整成功后的完成日，避免采集中途抢跑。"""
        row = conn.execute(
            "SELECT v FROM kv WHERE k = 'snapshot:last_completed'"
        ).fetchone()
        if row is not None and row["v"]:
            return str(row["v"])
        return None

    @staticmethod
    def _snapshot_days(conn: sqlite3.Connection) -> int:
        """快照天数：统计已完整完成的采集日（completed=1），与数据日期口径一致。"""
        row = conn.execute(
            "SELECT COUNT(*) AS n FROM kv "
            "WHERE k LIKE 'snapshot:completed:%' AND v = '1'"
        ).fetchone()
        return int(row["n"] or 0)

    @staticmethod
    def _snapshot_map(
        conn: sqlite3.Connection,
        aids: Sequence[int],
        *,
        before: str | None = None,
        before_inclusive: bool = False,
    ) -> dict[int, dict[str, Any]]:
        if not aids:
            return {}
        result: dict[int, dict[str, Any]] = {}
        for chunk in _chunks(list(aids)):
            placeholders = ",".join("?" for _ in chunk)
            if before is None:
                sql = f"""
                    SELECT s.* FROM snapshot s
                    JOIN (
                      SELECT aid, MAX(stat_date) AS stat_date
                      FROM snapshot WHERE ok = 1 AND aid IN ({placeholders})
                      GROUP BY aid
                    ) m ON m.aid = s.aid AND m.stat_date = s.stat_date
                """
                params: list[Any] = list(chunk)
            else:
                operator = "<=" if before_inclusive else "<"
                sql = f"""
                    SELECT s.* FROM snapshot s
                    JOIN (
                      SELECT aid, MAX(stat_date) AS stat_date
                      FROM snapshot
                      WHERE ok = 1 AND aid IN ({placeholders}) AND stat_date {operator} ?
                      GROUP BY aid
                    ) m ON m.aid = s.aid AND m.stat_date = s.stat_date
                """
                params = [*chunk, before]
            for row in conn.execute(sql, params).fetchall():
                data = dict(row)
                result[int(data["aid"])] = data
        return result

    @classmethod
    def _snapshot_previous_relative(
        cls,
        conn: sqlite3.Connection,
        current: Mapping[int, Mapping[str, Any]],
        *,
        days_back: int,
    ) -> dict[int, dict[str, Any]]:
        """相对各片当前快照日取 previous：days_back=1 为相邻成功日，否则 <= current-N。"""
        if not current or days_back < 1:
            return {}
        by_date: dict[str, list[int]] = {}
        for aid, snap in current.items():
            day = snap.get("stat_date")
            if not day:
                continue
            by_date.setdefault(str(day), []).append(int(aid))
        result: dict[int, dict[str, Any]] = {}
        for day, aids in by_date.items():
            if days_back <= 1:
                prev = cls._snapshot_map(
                    conn, aids, before=day, before_inclusive=False,
                )
            else:
                baseline = _date_minus(day, days_back)
                prev = cls._snapshot_map(
                    conn, aids, before=baseline, before_inclusive=True,
                )
            result.update(prev)
        return result

    @staticmethod
    def _videos(
        conn: sqlite3.Connection,
        *,
        statuses: Sequence[str] = TRACKED_STATUSES,
        folder_id: int | None = None,
        q: str | None = None,
        q_fields: Sequence[str] | None = None,
        within_field: str | None = None,
        within_start: str | None = None,
        within_end: str | None = None,
        up_mid: int | None = None,
    ) -> list[dict[str, Any]]:
        clauses = [f"v.status IN ({','.join('?' for _ in statuses)})"]
        params: list[Any] = list(statuses)
        if folder_id is not None:
            clauses.append("v.folder_id = ?")
            params.append(int(folder_id))
        if up_mid is not None:
            clauses.append("v.up_mid = ?")
            params.append(int(up_mid))
        if q:
            pattern = _like_pattern(q)
            fields = tuple(q_fields) if q_fields else ("title", "up_name")
            columns = []
            for field in fields:
                column = _Q_FIELDS.get(field)
                if column is None:
                    raise ValueError(f"不支持的搜索字段: {field}")
                columns.append(column)
            clauses.append(
                "(" + " OR ".join(f"{col} LIKE ? ESCAPE '\\'" for col in columns) + ")"
            )
            params.extend([pattern] * len(columns))
        if within_field:
            if within_field not in {"pubdate", "first_seen"}:
                raise ValueError("within_field 必须是 pubdate 或 first_seen")
            if within_start:
                start_ts, _ = day_bounds(within_start)
                clauses.append(f"v.{within_field} >= ?")
                params.append(start_ts)
            if within_end:
                end_ts, _ = day_bounds(within_end)
                clauses.append(f"v.{within_field} < ?")
                params.append(end_ts)
        sql = (
            "SELECT v.*, f.title AS folder_title, f.year AS folder_year "
            "FROM video v LEFT JOIN folder f ON f.media_id = v.folder_id "
            f"WHERE {' AND '.join(clauses)} ORDER BY v.aid"
        )
        return [dict(row) for row in conn.execute(sql, params).fetchall()]

    def _global_heat(
        self, conn: sqlite3.Connection
    ) -> tuple[list[dict[str, Any]], dict[int, dict[str, Any]], dict[int, float], dict[int, float]]:
        videos = self._videos(conn)
        snapshots = self._snapshot_map(conn, [int(v["aid"]) for v in videos])
        rows = []
        for video in videos:
            snapshot = snapshots.get(int(video["aid"]))
            if snapshot is None:
                continue
            item = {"aid": int(video["aid"])}
            item.update(_metric_map(snapshot) or {})
            rows.append(item)
        weights = self.config_store.get().heat_weights
        scores = heat_scores(rows, weights or DEFAULT_HEAT_WEIGHTS)
        percentiles = score_percentiles(scores)
        return videos, snapshots, scores, percentiles

    @staticmethod
    def _publish_date(pubdate: object) -> str | None:
        if pubdate is None or isinstance(pubdate, bool):
            return None
        try:
            ts = int(pubdate)
        except (TypeError, ValueError):
            return None
        if ts <= 0:
            return None
        return shanghai_date(ts)

    @staticmethod
    def _growth_for(
        snapshot: Mapping[str, Any] | None,
        previous: Mapping[str, Any] | None,
        metric: str = "view",
    ):
        if snapshot is None:
            return compute_growth(None, None, None)
        current = snapshot.get(metric)
        before = None if previous is None else previous.get(metric)
        gap = None
        if previous is not None and snapshot.get("stat_date") and previous.get("stat_date"):
            try:
                gap = (
                    date.fromisoformat(str(snapshot["stat_date"]))
                    - date.fromisoformat(str(previous["stat_date"]))
                ).days
            except ValueError:
                gap = None
        return compute_growth(current, before, gap)

    @classmethod
    def _same_day_growth(
        cls,
        snapshot: Mapping[str, Any] | None,
        video: Mapping[str, Any],
        metric: str = "view",
    ) -> dict[str, Any] | None:
        """近 1 日、没有更早快照时：当日发布相对 0；否则用 view_open。"""
        if snapshot is None:
            return None
        current = snapshot.get(metric)
        if current is None:
            return None
        stat_date = str(snapshot.get("stat_date") or "")
        published_on = cls._publish_date(video.get("pubdate"))
        if published_on is not None and published_on == stat_date:
            growth = compute_growth(current, 0, 0, allow_zero_gap=True)
            payload = growth.as_dict()
            payload["baseline"] = "zero"
            # previous=0 时 rate = view/1，会按原始播放碾压真日环比。
            payload["rate"] = None
            return payload
        try:
            open_at = snapshot.get("open_captured_at")
            captured_at = snapshot.get("captured_at")
            view_open = snapshot.get("view_open")
            open_ready = (
                open_at is not None
                and captured_at is not None
                and int(open_at) < int(captured_at)
                and view_open is not None
            )
        except (TypeError, ValueError):
            open_ready = False
        if not open_ready:
            return None
        growth = compute_growth(current, view_open, 0, allow_zero_gap=True)
        payload = growth.as_dict()
        payload["baseline"] = "open"
        return payload

    @classmethod
    def _growth_result(
        cls,
        video: Mapping[str, Any],
        snapshot: Mapping[str, Any] | None,
        previous: Mapping[str, Any] | None,
        *,
        days_back: int,
        metric: str = "view",
    ) -> dict[str, Any]:
        if previous is not None:
            growth = cls._growth_for(snapshot, previous, metric=metric)
            payload = growth.as_dict()
            payload["baseline"] = "snapshot" if growth.available else None
            return payload
        if days_back == 1:
            fallback = cls._same_day_growth(snapshot, video, metric=metric)
            if fallback is not None:
                return fallback
        payload = compute_growth(None, None, None).as_dict()
        payload["baseline"] = None
        return payload

    def _item(
        self,
        video: Mapping[str, Any],
        snapshot: Mapping[str, Any] | None,
        *,
        heat: float | None = None,
        percentile: float | None = None,
        previous: Mapping[str, Any] | None = None,
        include_growth: bool = False,
        growth_days: int | None = None,
    ) -> dict[str, Any]:
        metrics = _metric_map(snapshot)
        duration = video.get("duration")
        hours = content_hours(duration)
        item: dict[str, Any] = {
            "aid": int(video["aid"]),
            "bvid": video.get("bvid"),
            "title": video.get("title") or "(无标题)",
            "up_mid": video.get("up_mid"),
            "up_name": video.get("up_name"),
            "pubdate": int(video.get("pubdate") or 0),
            "duration": duration,
            "pages": video.get("pages"),
            "content_hours": None if hours is None else round(hours, 3),
            "up_follower": video.get("up_follower"),
            "up_follower_at": video.get("up_follower_at"),
            "up_follower_backfilled": bool(video.get("up_follower_backfilled") or 0),
            "cover": normalize_cover(video.get("cover")),
            "up_face": normalize_cover(video.get("up_face")),
            "folder_id": video.get("folder_id"),
            "folder_title": video.get("folder_title"),
            "folder_year": video.get("folder_year"),
            "status": video.get("status"),
            "first_seen": int(video.get("first_seen") or 0),
            "attr": video.get("attr"),
            "metrics": metrics,
            "snapshot_date": None if snapshot is None else snapshot.get("stat_date"),
            "heat": None if heat is None else round(float(heat), 3),
            "heat_percentile": percentile,
            "interaction_rate": None if metrics is None else interaction_rate(metrics),
            "efficiency": (
                None if metrics is None
                else views_per_hour(metrics.get("view"), duration)
            ),
            "interaction_efficiency": (
                None if metrics is None
                else interactions_per_hour(metrics, duration)
            ),
        }
        if include_growth:
            item["growth"] = self._growth_result(
                video, snapshot, previous, days_back=int(growth_days or 1),
            )
        return item

    # ------------------------------------------------------------ overview

    def overview(self) -> dict[str, Any]:
        return self._cached("videos", {}, TTL_VIDEOS, self._overview_uncached)

    def _overview_uncached(self) -> dict[str, Any]:
        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            latest = self._latest_date(conn)
            snapshot_days = self._snapshot_days(conn)
            counts_rows = conn.execute(
                "SELECT status, COUNT(*) AS n FROM video GROUP BY status"
            ).fetchall()
            by_status = {str(row["status"]): int(row["n"]) for row in counts_rows}
            videos = self._videos(conn)
            aids = [int(video["aid"]) for video in videos]
            snapshots = self._snapshot_map(conn, aids)
            # 播放增量：近 1 日。有更早日用相邻成功日；当日发布相对 0；否则用 view_open。
            previous = self._snapshot_previous_relative(conn, snapshots, days_back=1)
            totals = {metric: 0 for metric in METRICS}
            for video in videos:
                metrics = _metric_map(snapshots.get(int(video["aid"])))
                if metrics:
                    for metric in METRICS:
                        totals[metric] += metrics[metric]

            view_delta: int | None = None
            available_deltas = 0
            view_delta_sum = 0
            for video in videos:
                aid = int(video["aid"])
                snapshot = snapshots.get(aid)
                if snapshot is None:
                    continue
                growth = self._growth_result(
                    video, snapshot, previous.get(aid), days_back=1,
                )
                if growth.get("available") and growth.get("delta") is not None:
                    view_delta_sum += int(growth["delta"])
                    available_deltas += 1
            if available_deltas:
                view_delta = view_delta_sum

            first_seen_row = conn.execute(
                "SELECT MIN(first_seen) AS ts FROM video"
            ).fetchone()
            first_seen_ts = first_seen_row["ts"] if first_seen_row else None
            first_seen_date = _ts_date(first_seen_ts)
            initial_import_count = 0
            if first_seen_ts is not None:
                initial_import_count = int(conn.execute(
                    "SELECT COUNT(*) FROM video WHERE first_seen = ? "
                    "AND status IN ('active','unfaved')",
                    (int(first_seen_ts),),
                ).fetchone()[0])

            start_ts, end_ts = day_bounds(self._today())
            added_today = int(conn.execute(
                "SELECT COUNT(*) FROM video WHERE first_seen >= ? AND first_seen < ?",
                (start_ts, end_ts),
            ).fetchone()[0])
            is_first_day = snapshot_days <= 1
            if is_first_day:
                added_today = 0

            snapshot_health = {"total": 0, "ok": 0, "failed": 0}
            if latest:
                rows = conn.execute(
                    "SELECT ok, COUNT(*) AS n FROM snapshot WHERE stat_date = ? GROUP BY ok",
                    (latest,),
                ).fetchall()
                by_ok = {int(row["ok"]): int(row["n"]) for row in rows}
                snapshot_health = {
                    "total": sum(by_ok.values()),
                    "ok": by_ok.get(1, 0),
                    "failed": by_ok.get(0, 0),
                }

            candidate_pending = int(conn.execute(
                "SELECT COUNT(*) FROM candidate WHERE decision IS NULL"
            ).fetchone()[0])
            candidate_total = int(conn.execute(
                "SELECT COUNT(*) FROM candidate"
            ).fetchone()[0])

            cfg = self.config_store.get()
            fans = self._fans_overview(conn, cfg.track_fans, latest, snapshot_days)
            folders = self._folders_overview(conn)
            pub_years = sorted(
                {
                    int(shanghai_date(int(video["pubdate"]))[:4])
                    for video in videos
                    if int(video.get("pubdate") or 0) > 0
                },
                reverse=True,
            )
            tasks = self._task_records(conn)
            updated_at = self._latest_short_update_at(tasks)
            auth = {
                "ok": self._kv(conn, "auth:ok"),
                "message": self._kv(conn, "auth:message"),
                "last_check": self._kv(conn, "auth:last_check"),
                "mid": self._kv(conn, "auth:mid"),
            }
            preview = self._kv(conn, "preview:seeded") == "1"
            snapshot_completed_today = self._kv(
                conn, f"snapshot:completed:{self._today()}"
            ) == "1"
            return {
                "generated_at": int(self._now()),
                "today": self._today(),
                "data_date": latest,
                "updated_at": updated_at,
                "snapshot_days": snapshot_days,
                "first_seen_date": first_seen_date,
                "is_first_day": is_first_day,
                "initial_import_count": initial_import_count,
                "added_today": added_today,
                "counts": {
                    **by_status,
                    "total": sum(by_status.values()),
                    "active_ups": int(conn.execute(
                        "SELECT COUNT(DISTINCT up_mid) FROM video "
                        "WHERE status IN ('active','unfaved') AND up_mid IS NOT NULL"
                    ).fetchone()[0]),
                    "candidates_pending": candidate_pending,
                    "candidates_total": candidate_total,
                },
                "totals": totals,
                "interactions": totals["like"] + totals["coin"] + totals["favorite"],
                "view_delta": view_delta,
                "view_delta_available": view_delta is not None,
                "snapshot_health": snapshot_health,
                "snapshot_completed_today": snapshot_completed_today,
                "fans": fans,
                "folders": folders,
                "years": pub_years,
                "tasks": tasks,
                "auth": auth,
                "preview": preview,
                "capacity": CAPACITY,
                "capacity_warning": CAPACITY_WARNING,
            }
        return self.db.read(run)

    @staticmethod
    def _kv(conn: sqlite3.Connection, key: str) -> str | None:
        row = conn.execute("SELECT v FROM kv WHERE k = ?", (key,)).fetchone()
        return None if row is None else str(row["v"])

    @staticmethod
    def _latest_short_update_at(tasks: Mapping[str, Any]) -> int | None:
        """首页「更新」时刻：优先短更新任务，其次全日更 / 粉丝。"""
        for name in ("snapshot_recent", "snapshot", "fans"):
            raw = tasks.get(name)
            if not isinstance(raw, Mapping):
                continue
            finished = raw.get("finished_at")
            if finished is None:
                continue
            try:
                return int(finished)
            except (TypeError, ValueError):
                continue
        return None

    @staticmethod
    def _task_started_sort_key(value: Any) -> tuple[int, int]:
        """缺 started_at 的排末尾；有值的按开始时间倒序（含进行中）。"""
        if isinstance(value, Mapping):
            raw = value.get("started_at")
            try:
                return (0, -int(raw))
            except (TypeError, ValueError):
                pass
        return (1, 0)

    def _task_records(self, conn: sqlite3.Connection) -> dict[str, Any]:
        rows = conn.execute(
            "SELECT k, v FROM kv WHERE k LIKE 'task:last:%'"
        ).fetchall()
        items: list[tuple[str, Any]] = []
        for row in rows:
            name = str(row["k"]).removeprefix("task:last:")
            try:
                value = json.loads(str(row["v"]))
            except (TypeError, json.JSONDecodeError):
                value = {"raw": row["v"]}
            items.append((name, value))
        items.sort(key=lambda pair: self._task_started_sort_key(pair[1]))
        # 插入序即开始时间倒序；前端 Object.entries 保留该序，勿再排序。
        return dict(items)

    def _task_running(self, conn: sqlite3.Connection) -> dict[str, Any]:
        rows = conn.execute(
            "SELECT k, v FROM kv WHERE k LIKE 'task:running:%' ORDER BY k"
        ).fetchall()
        result: dict[str, Any] = {}
        for row in rows:
            name = str(row["k"]).removeprefix("task:running:")
            try:
                value = json.loads(str(row["v"]))
            except (TypeError, json.JSONDecodeError):
                value = {"raw": row["v"]}
            if not isinstance(value, dict):
                value = {"raw": value}
            value["running"] = True
            result[name] = value
        return result

    def _folders_overview(self, conn: sqlite3.Connection) -> list[dict[str, Any]]:
        rows = conn.execute(
            """
            SELECT f.*,
              (SELECT COUNT(*) FROM folder_resource fr
               WHERE fr.media_id = f.media_id AND fr.active = 1) AS local_count
            FROM folder f ORDER BY f.year DESC, f.media_id DESC
            """
        ).fetchall()
        result = []
        for row in rows:
            media_count = row["media_count"]
            count = int(media_count) if media_count is not None else int(row["local_count"] or 0)
            result.append({
                "media_id": int(row["media_id"]),
                "year": int(row["year"]),
                "title": row["title"] or "",
                "media_count": count,
                "local_count": int(row["local_count"] or 0),
                "last_ids_at": row["last_ids_at"],
                "last_full_at": row["last_full_at"],
                "enabled": bool(row["enabled"]),
                "capacity": CAPACITY,
                "ratio": min(1.0, count / CAPACITY),
                "warning": count >= CAPACITY_WARNING,
            })
        return result

    def _fans_overview(
        self,
        conn: sqlite3.Connection,
        mid_list: Sequence[int],
        latest: str | None,
        snapshot_days: int,
    ) -> dict[str, Any] | None:
        mids = list(mid_list) or [
            int(row["mid"])
            for row in conn.execute("SELECT DISTINCT mid FROM up_stat ORDER BY mid").fetchall()
        ]
        if not mids:
            return None
        mid = int(mids[0])
        rows = conn.execute(
            "SELECT * FROM up_stat WHERE mid = ? ORDER BY stat_date", (mid,)
        ).fetchall()
        if not rows:
            return None
        ok_rows = [dict(row) for row in rows if int(row["ok"]) == 1]
        if not ok_rows:
            return None
        current = ok_rows[-1]
        previous = ok_rows[-2] if len(ok_rows) >= 2 else None
        if previous is not None:
            try:
                gap = (
                    date.fromisoformat(str(current["stat_date"]))
                    - date.fromisoformat(str(previous["stat_date"]))
                ).days
            except ValueError:
                gap = 0
            if gap != 1:
                previous = None
        baseline = None
        if latest:
            cutoff = _date_minus(latest, 7)
            by_date = {str(row["stat_date"]): row for row in ok_rows}
            baseline = by_date.get(cutoff)
        first = ok_rows[0]
        return {
            "mid": mid,
            "name": TRACK_FANS_NAMES.get(mid, f"UP {mid}"),
            "stat_date": current["stat_date"],
            "follower": current["follower"],
            "following": current["following"],
            "delta_1d": None if previous is None else int(current["follower"] or 0) - int(previous["follower"] or 0),
            "delta_7d": None if baseline is None else int(current["follower"] or 0) - int(baseline["follower"] or 0),
            "delta_cum": int(current["follower"] or 0) - int(first["follower"] or 0),
            "first_date": first["stat_date"],
            "snapshot_days": snapshot_days,
            "points": [
                {
                    "date": row["stat_date"],
                    "follower": row["follower"] if int(row["ok"]) == 1 else None,
                    "ok": bool(row["ok"]),
                }
                for row in rows[-30:]
            ],
        }

    # ------------------------------------------------------------ videos

    def list_videos(
        self,
        *,
        sort: str = "heat",
        within_days: int | None = None,
        year: int | None = None,
        growth_days: int | str = "auto",
        folder_id: int | None = None,
        q: str | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> dict[str, Any]:
        params = {
            "sort": sort,
            "within_days": within_days,
            "year": year,
            "growth_days": growth_days,
            "folder_id": folder_id,
            "q": q or "",
            "page": page,
            "page_size": page_size,
        }
        return self._cached(
            "videos", params, TTL_VIDEOS, lambda: self._list_videos_uncached(
                sort=sort,
                within_days=within_days,
                year=year,
                growth_days=growth_days,
                folder_id=folder_id,
                q=q,
                page=page,
                page_size=page_size,
            ),
        )

    def _list_videos_uncached(
        self,
        *,
        sort: str = "heat",
        within_days: int | None = None,
        year: int | None = None,
        growth_days: int | str = "auto",
        folder_id: int | None = None,
        q: str | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> dict[str, Any]:
        if sort not in VIDEO_SORTS:
            raise ValueError(f"不支持的排序: {sort}")
        # auto：仅增长类排序由服务端选窗（配置 home_growth_days；默认先 7 再降 1）。
        # 首页应省略 growth_days，勿自行探查。其它排序把 auto 当作 7（兼容旧默认）。
        if growth_days == "auto":
            if sort in {"growth", "growth_rate"}:
                preferred = self.config_store.get().analytics.home_growth_days
                if preferred in (1, 7):
                    growth_days = int(preferred)
                else:
                    trial = self._list_videos_uncached(
                        sort=sort,
                        within_days=within_days,
                        year=year,
                        growth_days=7,
                        folder_id=folder_id,
                        q=q,
                        page=page,
                        page_size=page_size,
                    )
                    if trial.get("available") and int(trial.get("total") or 0) > 0:
                        return trial
                    growth_days = 1
            else:
                growth_days = 7
        growth_days = int(growth_days)
        if growth_days < 1 or growth_days > 365:
            raise ValueError("growth_days 必须在 1~365 之间")
        page = max(1, int(page))
        page_size = max(1, min(int(page_size), 200))

        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            latest = self._latest_date(conn) or self._today()
            analytics = self.config_store.get().analytics
            duration_sorts = {"efficiency", "interaction_efficiency"}
            min_view_for_ranking = max(0, int(analytics.min_view_for_ranking))
            within_field = None
            within_start: str | None = None
            within_end: str | None = None
            if year is not None:
                within_field = "pubdate"
                within_start = f"{int(year)}-01-01"
                within_end = f"{int(year) + 1}-01-01"
            elif within_days is not None:
                if within_days < 1:
                    raise ValueError("within_days 必须大于 0")
                within_field = "first_seen" if sort == "added" else "pubdate"
                within_start = _date_minus(latest, int(within_days) - 1)
            videos = self._videos(
                conn,
                folder_id=folder_id,
                q=q,
                within_field=within_field,
                within_start=within_start,
                within_end=within_end,
            )
            aids = [int(video["aid"]) for video in videos]
            snapshots = self._snapshot_map(conn, aids)
            videos_all, snapshots_all, heat, percentiles = self._global_heat(conn)

            previous: dict[int, dict[str, Any]] = {}
            if sort in {"growth", "growth_rate", "interaction"}:
                # 相对各片当前快照日回退 growth_days，避免 seed/recent 相对 last_completed 拉长窗口。
                previous = self._snapshot_previous_relative(
                    conn, snapshots, days_back=growth_days,
                )
            elif sort in {"heat", "view", "like", "coin", "favorite", "share", "reply", "danmaku"}:
                previous = {}

            items = [
                self._item(
                    video,
                    snapshots.get(int(video["aid"])),
                    heat=heat.get(int(video["aid"])),
                    percentile=percentiles.get(int(video["aid"])),
                    previous=previous.get(int(video["aid"])),
                    include_growth=sort in {"growth", "growth_rate"},
                    growth_days=growth_days,
                )
                for video in videos
            ]

            if sort in duration_sorts:
                latest_start, _ = day_bounds(latest)
                minimum_seconds = max(0, int(analytics.min_duration_seconds))
                items = [
                    item for item in items
                    if item["metrics"] is not None
                    and int(item["duration"] or 0) >= minimum_seconds
                    and int((item["metrics"] or {}).get("view", 0)) >= min_view_for_ranking
                    and is_mature(
                        item["pubdate"], latest_start, analytics.maturity_days
                    )
                ]
                kappa_hours = max(0, int(analytics.video_shrink_minutes)) / 60.0
                total_hours = 0.0
                total_views = 0.0
                total_interactions = 0.0
                for video in videos_all:
                    hours = content_hours(video.get("duration"))
                    if hours is None:
                        continue
                    metrics = _metric_map(snapshots_all.get(int(video["aid"])))
                    if metrics is None:
                        continue
                    total_hours += hours
                    total_views += metrics["view"]
                    total_interactions += (
                        metrics["like"] + metrics["coin"] + metrics["favorite"]
                    )
                prior_views = total_views / total_hours if total_hours > 0 else 0.0
                prior_interactions = (
                    total_interactions / total_hours if total_hours > 0 else 0.0
                )
                for item in items:
                    hours = content_hours(item.get("duration")) or 0.0
                    metrics = item["metrics"] or {}
                    denominator = hours + kappa_hours
                    if hours <= 0 or denominator <= 0:
                        item["efficiency_score"] = None
                        item["interaction_efficiency_score"] = None
                        continue
                    view = int(metrics.get("view", 0))
                    interactions = (
                        int(metrics.get("like", 0))
                        + int(metrics.get("coin", 0))
                        + int(metrics.get("favorite", 0))
                    )
                    item["efficiency_score"] = (
                        view + kappa_hours * prior_views
                    ) / denominator
                    item["interaction_efficiency_score"] = (
                        interactions + kappa_hours * prior_interactions
                    ) / denominator
            if sort == "heat":
                items = [
                    item for item in items
                    if int((item["metrics"] or {}).get("view", 0)) >= min_view_for_ranking
                ]
            if sort == "interaction":
                items = [item for item in items if item["interaction_rate"] is not None]
            if sort in {"growth", "growth_rate"}:
                items = [
                    item for item in items
                    if item.get("growth", {}).get("available")
                ]
                if sort == "growth_rate":
                    items = [
                        item for item in items
                        if item.get("growth", {}).get("baseline") != "zero"
                        and item.get("growth", {}).get("rate") is not None
                        and (_growth_baseline_view(item) or 0) >= MIN_VIEW_FOR_GROWTH_RATE
                    ]

            rate_prior = max(0, int(analytics.growth_rate_prior))

            def sort_key(item: Mapping[str, Any]):
                if sort == "added":
                    return (int(item["first_seen"]), int(item["pubdate"]))
                if sort == "published":
                    return (int(item["pubdate"]), int(item["first_seen"]))
                if sort == "heat":
                    return item["heat"] if item["heat"] is not None else float("-inf")
                if sort == "interaction":
                    return item["interaction_rate"] if item["interaction_rate"] is not None else float("-inf")
                if sort in {"efficiency", "interaction_efficiency"}:
                    field = (
                        "efficiency_score" if sort == "efficiency"
                        else "interaction_efficiency_score"
                    )
                    value = item.get(field)
                    return value if value is not None else float("-inf")
                if sort == "growth":
                    value = item.get("growth", {}).get("delta")
                    return value if value is not None else float("-inf")
                if sort == "growth_rate":
                    growth = item.get("growth") or {}
                    delta = growth.get("delta")
                    current = (item.get("metrics") or {}).get("view")
                    if delta is None or current is None:
                        return float("-inf")
                    value = smoothed_growth_rate(
                        delta, int(current) - int(delta), rate_prior,
                    )
                    return value if value is not None else float("-inf")
                value = (item["metrics"] or {}).get(sort)
                return value if value is not None else -1

            items.sort(key=lambda item: item["aid"])
            items.sort(key=sort_key, reverse=True)
            total = len(items)
            growth_available = any(
                item.get("growth", {}).get("available") for item in items
            )
            start = (page - 1) * page_size
            items = items[start:start + page_size]
            available = True
            reason = None
            if sort in {"growth", "growth_rate"} and not growth_available:
                available = False
                if growth_days == 1 and sort == "growth_rate":
                    reason = (
                        "近 1 日增速需要基线播放至少 "
                        f"{MIN_VIEW_FOR_GROWTH_RATE}；当日发布不计入增速"
                    )
                elif growth_days == 1:
                    reason = (
                        "近 1 日增长需要前一日快照；当日发布有一次成功采样即可；"
                        "更早发布的新入库需要当天至少两次成功采样"
                    )
                else:
                    reason = (
                        f"近 {growth_days} 日增长需要至少 {growth_days + 1} 天快照"
                    )
            duration_filter = None
            if sort in duration_sorts:
                duration_filter = {
                    "min_duration_seconds": max(0, int(analytics.min_duration_seconds)),
                    "maturity_days": max(0, int(analytics.maturity_days)),
                    "min_view": min_view_for_ranking,
                    "shrink_minutes": max(0, int(analytics.video_shrink_minutes)),
                }
                if not items:
                    available = False
                    reason = (
                        f"仅统计时长 ≥ {duration_filter['min_duration_seconds']} 秒、"
                        f"发布满 {duration_filter['maturity_days']} 天、"
                        f"播放 ≥ {duration_filter['min_view']} 的视频"
                    )
            return {
                "sort": sort,
                "within_days": within_days,
                "year": year,
                "growth_days": growth_days,
                # days：与 growth_days 同步；首页 auto 时为 1|7，供前端链接/展示
                "days": growth_days,
                "page": page,
                "page_size": page_size,
                "total": total,
                "pages": max(1, math.ceil(total / page_size)) if total else 0,
                "items": items,
                "available": available,
                "reason": reason,
                "duration_filter": duration_filter,
                "growth_rate_prior": rate_prior if sort == "growth_rate" else None,
            }
        return self.db.read(run)

    def video_detail(self, aid: int) -> dict[str, Any] | None:
        aid = int(aid)

        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            video_rows = self._videos(conn)
            video = next((item for item in video_rows if int(item["aid"]) == aid), None)
            if video is None:
                row = conn.execute(
                    "SELECT v.*, f.title AS folder_title, f.year AS folder_year "
                    "FROM video v LEFT JOIN folder f ON f.media_id = v.folder_id "
                    "WHERE v.aid = ?",
                    (aid,),
                ).fetchone()
                if row is None:
                    return None
                video = dict(row)
            snapshots = self._snapshot_map(conn, [aid])
            snapshot = snapshots.get(aid)
            previous = self._snapshot_map(
                conn, [aid], before=snapshot["stat_date"] if snapshot else None,
            ).get(aid)
            _, _, heat, percentiles = self._global_heat(conn)
            item = self._item(
                video,
                snapshot,
                heat=heat.get(aid),
                percentile=percentiles.get(aid),
            )
            item["growth"] = self._growth_result(
                video, snapshot, previous, days_back=1,
            )
            item["snapshot_count"] = int(conn.execute(
                "SELECT COUNT(*) FROM snapshot WHERE aid = ?", (aid,)
            ).fetchone()[0])
            item["blacklisted"] = bool(conn.execute(
                "SELECT 1 FROM up_blacklist WHERE mid = ?", (video.get("up_mid"),)
            ).fetchone()) if video.get("up_mid") is not None else False
            item["url"] = (
                f"https://www.bilibili.com/video/{video.get('bvid')}"
                if video.get("bvid") else None
            )
            return item
        return self.db.read(run)

    def video_trend(self, aid: int, *, days: int | None = 30) -> dict[str, Any]:
        aid = int(aid)
        if days is not None and days < 1:
            raise ValueError("days 必须大于 0")

        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            latest = self._latest_date(conn)
            rows = conn.execute(
                "SELECT * FROM snapshot WHERE aid = ? ORDER BY stat_date", (aid,)
            ).fetchall()
            points = [dict(row) for row in rows]
            if latest is None or not points:
                return {
                    "aid": aid,
                    "days": days,
                    "dates": [],
                    "series": {metric: [] for metric in METRICS},
                    "points": [],
                    "available": False,
                }
            requested_start = (
                _date_minus(latest, days - 1) if days is not None
                else str(points[0]["stat_date"])
            )
            first_point = str(points[0]["stat_date"])
            start = max(requested_start, first_point)
            dates = _date_range(start, latest)
            by_date = {str(point["stat_date"]): point for point in points}
            series: dict[str, list[int | None]] = {}
            for metric in METRICS:
                values: list[int | None] = []
                for day in dates:
                    point = by_date.get(day)
                    if point is None or int(point["ok"]) != 1 or point.get(metric) is None:
                        values.append(None)
                    else:
                        values.append(int(point[metric]))
                series[metric] = values
            filtered_points = [
                point for point in points if str(point["stat_date"]) in set(dates)
            ]
            return {
                "aid": aid,
                "days": days,
                "dates": dates,
                "series": series,
                "points": filtered_points,
                "available": any(point["ok"] == 1 for point in filtered_points),
            }
        return self.db.read(run)

    # ------------------------------------------------------------ ups

    def list_ups(
        self,
        *,
        sort: str = "avg_views",
        within_days: int | None = None,
        year: int | None = None,
        q: str | None = None,
        min_videos: int = 1,
        page: int = 1,
        page_size: int = 20,
    ) -> dict[str, Any]:
        params = {
            "sort": sort,
            "within_days": within_days,
            "year": year,
            "q": q or "",
            "min_videos": min_videos,
            "page": page,
            "page_size": page_size,
        }
        return self._cached(
            "videos", params, TTL_VIDEOS, lambda: self._list_ups_uncached(
                sort=sort,
                within_days=within_days,
                year=year,
                q=q,
                min_videos=min_videos,
                page=page,
                page_size=page_size,
            ),
        )

    def _list_ups_uncached(
        self,
        *,
        sort: str = "avg_views",
        within_days: int | None = None,
        year: int | None = None,
        q: str | None = None,
        min_videos: int = 1,
        page: int = 1,
        page_size: int = 20,
    ) -> dict[str, Any]:
        if sort not in UP_SORTS:
            raise ValueError(f"不支持的 UP 排序: {sort}")
        page = max(1, int(page))
        page_size = max(1, min(int(page_size), 200))
        min_videos = max(1, int(min_videos))

        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            latest = self._latest_date(conn) or self._today()
            if year is not None:
                within_field = "pubdate"
                within_start: str | None = f"{int(year)}-01-01"
                within_end: str | None = f"{int(year) + 1}-01-01"
            elif within_days is not None:
                within_field = "pubdate"
                within_start = _date_minus(latest, int(within_days) - 1)
                within_end = None
            else:
                within_field = None
                within_start = None
                within_end = None
            videos = self._videos(
                conn,
                q=q,
                q_fields=("up_name",) if q else None,
                within_field=within_field,
                within_start=within_start,
                within_end=within_end,
            )
            snapshots = self._snapshot_map(conn, [int(video["aid"]) for video in videos])
            groups: dict[int, dict[str, Any]] = {}
            for video in videos:
                mid = video.get("up_mid")
                if mid is None:
                    continue
                mid = int(mid)
                group = groups.setdefault(mid, {
                    "up_mid": mid,
                    "up_name": video.get("up_name") or f"UP {mid}",
                    "videos": [],
                    "face": None,
                })
                if group["face"] is None and video.get("up_face"):
                    group["face"] = video.get("up_face")
                metrics = _metric_map(snapshots.get(int(video["aid"])))
                group["videos"].append((video, metrics))
            items = []
            for group in groups.values():
                group_videos = group["videos"]
                metrics_list = [metrics for _, metrics in group_videos if metrics]
                total_views = sum(metrics["view"] for metrics in metrics_list)
                total_interactions = sum(
                    metrics["like"] + metrics["coin"] + metrics["favorite"]
                    for metrics in metrics_list
                )
                rates = [
                    rate for rate in (
                        interaction_rate(metrics) for metrics in metrics_list
                    ) if rate is not None
                ]
                exposure_seconds = sum(
                    int(video.get("duration") or 0)
                    for video, metrics in group_videos
                    if metrics is not None and int(video.get("duration") or 0) > 0
                )
                exposure_minutes = exposure_seconds / 60.0
                best = max(
                    group_videos,
                    key=lambda pair: (pair[1] or {}).get("view", 0),
                    default=None,
                )
                items.append({
                    "up_mid": group["up_mid"],
                    "up_name": group["up_name"],
                    "up_face": normalize_cover(group["face"]),
                    "follower_latest": None,
                    "video_count": len(group_videos),
                    "total_views": total_views,
                    "avg_views": round(total_views / len(metrics_list), 1) if metrics_list else None,
                    "total_interactions": total_interactions,
                    "avg_interaction": round(sum(rates) / len(rates), 6) if rates else None,
                    "latest_pubdate": max(int(video["pubdate"]) for video, _ in group_videos),
                    "first_pubdate": min(int(video["pubdate"]) for video, _ in group_videos),
                    "_exposure_minutes": exposure_minutes,
                    "best_video": None if best is None else {
                        "aid": int(best[0]["aid"]),
                        "title": best[0].get("title"),
                        "cover": normalize_cover(best[0].get("cover")),
                        "view": (best[1] or {}).get("view"),
                    },
                })

            total_minutes = sum(item["_exposure_minutes"] for item in items)
            total_views_all = sum(item["total_views"] for item in items)
            total_interactions_all = sum(item["total_interactions"] for item in items)
            prior_views = (
                total_views_all / total_minutes if total_minutes > 0 else 0.0
            )
            prior_interactions = (
                total_interactions_all / total_minutes if total_minutes > 0 else 0.0
            )
            shrink_minutes = effective_shrink_minutes(
                item["_exposure_minutes"] for item in items
            )
            for item in items:
                minutes = item.pop("_exposure_minutes")
                item["total_duration"] = int(round(minutes * 60))
                item["content_hours"] = round(minutes / 60.0, 2)
                item["views_per_hour"] = (
                    round(item["total_views"] / minutes * 60.0, 1)
                    if minutes > 0 else None
                )
                item["interactions_per_hour"] = (
                    round(item["total_interactions"] / minutes * 60.0, 1)
                    if minutes > 0 else None
                )
                item["efficiency"] = (
                    round(
                        eb_rate(
                            item["total_views"], minutes,
                            prior_views, shrink_minutes,
                        ) * 60.0,
                        2,
                    )
                    if minutes > 0 else None
                )
                item["interaction_efficiency"] = (
                    round(
                        eb_rate(
                            item["total_interactions"], minutes,
                            prior_interactions, shrink_minutes,
                        ) * 60.0,
                        2,
                    )
                    if minutes > 0 else None
                )

            mids = [item["up_mid"] for item in items]
            if mids:
                placeholders = ",".join("?" for _ in mids)
                profile_rows = conn.execute(
                    f"SELECT mid, follower_latest FROM up_profile "
                    f"WHERE mid IN ({placeholders})",
                    mids,
                ).fetchall()
                followers = {
                    int(row["mid"]): row["follower_latest"] for row in profile_rows
                }
                for item in items:
                    item["follower_latest"] = followers.get(item["up_mid"])

            items = [item for item in items if item["video_count"] >= min_videos]
            sort_map = {
                "views": lambda item: item["total_views"],
                "count": lambda item: item["video_count"],
                "avg_views": lambda item: item["avg_views"] or 0,
                "interaction": lambda item: item["total_interactions"],
                "avg_interaction": lambda item: item["avg_interaction"] or 0,
                "latest": lambda item: item["latest_pubdate"],
                "duration": lambda item: item["total_duration"],
                "efficiency": lambda item: item["efficiency"] or 0,
                "interaction_efficiency": lambda item: item["interaction_efficiency"] or 0,
            }
            key = sort_map[sort]
            items.sort(key=lambda item: (key(item), item["up_mid"]), reverse=True)
            total = len(items)
            start = (page - 1) * page_size
            items = items[start:start + page_size]
            return {
                "sort": sort,
                "within_days": within_days,
                "year": year,
                "min_videos": min_videos,
                "page": page,
                "page_size": page_size,
                "total": total,
                "pages": max(1, math.ceil(total / page_size)) if total else 0,
                "items": items,
            }
        return self.db.read(run)

    def up_detail(self, mid: int) -> dict[str, Any] | None:
        mid = int(mid)

        def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
            latest = self._latest_date(conn) or self._today()
            videos = self._videos(conn, up_mid=mid)
            if not videos:
                row = conn.execute(
                    "SELECT 1 FROM up_blacklist WHERE mid = ?", (mid,)
                ).fetchone()
                if row is None:
                    return None
                return {
                    "up_mid": mid,
                    "up_name": f"UP {mid}",
                    "videos": [],
                    "video_count": 0,
                    "blacklisted": True,
                    "series": {"dates": [], "views": []},
                }
            snapshots = self._snapshot_map(conn, [int(video["aid"]) for video in videos])
            items = [
                self._item(video, snapshots.get(int(video["aid"])))
                for video in videos
            ]
            metrics_list = [item["metrics"] for item in items if item["metrics"]]
            rates = [item["interaction_rate"] for item in items if item["interaction_rate"] is not None]
            total_views = sum(metrics["view"] for metrics in metrics_list)
            total_interactions = sum(
                metrics["like"] + metrics["coin"] + metrics["favorite"]
                for metrics in metrics_list
            )
            exposure_seconds = sum(
                int(item["duration"] or 0)
                for item in items
                if item["metrics"] is not None and int(item["duration"] or 0) > 0
            )
            exposure_minutes = exposure_seconds / 60.0
            profile = conn.execute(
                "SELECT * FROM up_profile WHERE mid = ?", (mid,)
            ).fetchone()
            follower_item = next(
                (item for item in items if item.get("up_follower") is not None), None
            )
            aids = [int(video["aid"]) for video in videos]
            rows = conn.execute(
                f"SELECT stat_date, SUM(view) AS view FROM snapshot "
                f"WHERE ok = 1 AND aid IN ({','.join('?' for _ in aids)}) "
                "GROUP BY stat_date ORDER BY stat_date",
                aids,
            ).fetchall()
            view_by_date = {str(row["stat_date"]): int(row["view"] or 0) for row in rows}
            dates = _date_range(min(view_by_date), latest) if view_by_date else []
            items.sort(key=lambda item: ((item["metrics"] or {}).get("view", 0), item["pubdate"]), reverse=True)
            return {
                "up_mid": mid,
                "up_name": items[0].get("up_name") or f"UP {mid}",
                "up_face": next(
                    (item["up_face"] for item in items if item.get("up_face")), None
                ),
                "video_count": len(items),
                "total_views": total_views,
                "avg_views": round(total_views / len(metrics_list), 1) if metrics_list else None,
                "total_interactions": total_interactions,
                "avg_interaction": round(sum(rates) / len(rates), 6) if rates else None,
                "total_duration": int(round(exposure_minutes * 60)),
                "content_hours": round(exposure_minutes / 60.0, 2),
                "views_per_hour": (
                    round(total_views / exposure_minutes * 60.0, 1)
                    if exposure_minutes > 0 else None
                ),
                "interactions_per_hour": (
                    round(total_interactions / exposure_minutes * 60.0, 1)
                    if exposure_minutes > 0 else None
                ),
                "up_follower": None if follower_item is None else follower_item["up_follower"],
                "up_follower_at": None if follower_item is None else follower_item["up_follower_at"],
                "up_follower_backfilled": bool(
                    follower_item is not None
                    and follower_item.get("up_follower_backfilled")
                ),
                "follower_first": None if profile is None else profile["follower_first"],
                "follower_latest": None if profile is None else profile["follower_latest"],
                "first_pubdate": min(item["pubdate"] for item in items),
                "latest_pubdate": max(item["pubdate"] for item in items),
                "blacklisted": bool(conn.execute(
                    "SELECT 1 FROM up_blacklist WHERE mid = ?", (mid,)
                ).fetchone()),
                "space_url": f"https://space.bilibili.com/{mid}",
                "videos": items,
                "series": {
                    "dates": dates,
                    "views": [view_by_date.get(day) for day in dates],
                },
            }
        return self.db.read(run)

    # ------------------------------------------------------------ fans

    def fans(
        self,
        mid: int | None = None,
        *,
        days: int = 90,
        start: str | None = None,
        end: str | None = None,
        recent_days: int = 14,
    ) -> dict[str, Any]:
        params = {
            "mid": mid if mid is not None else "",
            "days": days,
            "start": start or "",
            "end": end or "",
            "recent_days": recent_days,
        }
        return self._cached(
            "fans", params, TTL_FANS, lambda: self._fans_uncached(
                mid=mid,
                days=days,
                start=start,
                end=end,
                recent_days=recent_days,
            ),
        )

    def _fans_uncached(
        self,
        mid: int | None = None,
        *,
        days: int = 90,
        start: str | None = None,
        end: str | None = None,
        recent_days: int = 14,
    ) -> dict[str, Any]:
        if days < 1:
            raise ValueError("days 必须大于 0")
        if recent_days < 1 or recent_days > 3650:
            raise ValueError("recent_days 必须在 1~3650 之间")
        for label, value in (("start", start), ("end", end)):
            if value:
                try:
                    date.fromisoformat(value)
                except ValueError as exc:
                    raise ValueError(f"{label} 必须是 YYYY-MM-DD") from exc
        if start and end and start > end:
            raise ValueError("start 不能晚于 end")

        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            cfg = self.config_store.get()
            mids = list(cfg.track_fans)
            if not mids:
                mids = [
                    int(row["mid"]) for row in conn.execute(
                        "SELECT DISTINCT mid FROM up_stat ORDER BY mid"
                    ).fetchall()
                ]
            selected = int(mid) if mid is not None else (mids[0] if mids else None)
            options = [
                {"mid": value, "name": TRACK_FANS_NAMES.get(value, f"UP {value}")}
                for value in mids
            ]
            if selected is None:
                return {
                    "mid": None, "name": None, "available": False,
                    "options": options, "dates": [], "series": [],
                    "points": [], "recent_points": [], "days": days,
                    "mode": "daily", "range_start": None, "range_end": None,
                    "span_days": 0, "recent_days": recent_days,
                }
            rows = [
                dict(row) for row in conn.execute(
                    "SELECT * FROM up_stat WHERE mid = ? ORDER BY stat_date",
                    (selected,),
                ).fetchall()
            ]
            if not rows:
                return {
                    "mid": selected,
                    "name": TRACK_FANS_NAMES.get(selected, f"UP {selected}"),
                    "available": False,
                    "options": options,
                    "dates": [],
                    "series": [],
                    "points": [],
                    "recent_points": [],
                    "days": days,
                    "mode": "daily", "range_start": None, "range_end": None,
                    "span_days": 0, "recent_days": recent_days,
                }
            latest = max(str(row["stat_date"]) for row in rows if int(row["ok"]) == 1) if any(int(row["ok"]) == 1 for row in rows) else str(rows[-1]["stat_date"])
            first_date = str(rows[0]["stat_date"])
            recent_start = _date_minus(latest, recent_days - 1)
            recent_points = _with_fan_deltas(rows, recent_start)
            if start or end:
                range_start = max(str(start or first_date), first_date)
                range_end = min(str(end or latest), latest)
                if range_start > range_end:
                    raise ValueError("所选日期范围内没有粉丝数据")
            else:
                range_start = max(_date_minus(latest, days - 1), first_date)
                range_end = latest
            span_days = (
                date.fromisoformat(range_end) - date.fromisoformat(range_start)
            ).days + 1
            mode = "daily" if span_days <= 366 else "monthly"
            points = [
                row for row in rows
                if range_start <= str(row["stat_date"]) <= range_end
            ]
            if mode == "daily":
                dates, series = _daily_fan_series(points, range_start, range_end)
            else:
                dates, series = _monthly_peak_series(
                    points, range_start, range_end
                )
            ok_rows = [row for row in rows if int(row["ok"]) == 1]
            current = ok_rows[-1] if ok_rows else rows[-1]
            previous = ok_rows[-2] if len(ok_rows) >= 2 else None
            if previous is not None:
                try:
                    gap = (
                        date.fromisoformat(str(current["stat_date"]))
                        - date.fromisoformat(str(previous["stat_date"]))
                    ).days
                except ValueError:
                    gap = 0
                if gap != 1:
                    previous = None
            cutoff = _date_minus(latest, 7)
            by_date = {str(row["stat_date"]): row for row in ok_rows}
            baseline = by_date.get(cutoff)
            first = ok_rows[0] if ok_rows else rows[0]
            return {
                "mid": selected,
                "name": TRACK_FANS_NAMES.get(selected, f"UP {selected}"),
                "available": bool(ok_rows),
                "options": options,
                "days": days,
                "mode": mode,
                "range_start": range_start,
                "range_end": range_end,
                "span_days": span_days,
                "dates": dates,
                "series": series,
                "points": points,
                "recent_points": recent_points,
                "recent_days": recent_days,
                "stat_date": current["stat_date"],
                "follower": current.get("follower"),
                "following": current.get("following"),
                "delta_1d": None if previous is None else int(current.get("follower") or 0) - int(previous.get("follower") or 0),
                "delta_7d": None if baseline is None else int(current.get("follower") or 0) - int(baseline.get("follower") or 0),
                "delta_cum": int(current.get("follower") or 0) - int(first.get("follower") or 0),
                "first_date": first["stat_date"],
            }
        return self.db.read(run)

    # ------------------------------------------------------------ candidates

    def about(self) -> dict[str, Any]:
        """数据说明页所需的动态口径；不包含任何凭据。"""
        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            cfg = self.config_store.get()
            latest = self._latest_date(conn)
            first_row = conn.execute(
                "SELECT MIN(stat_date) AS d FROM snapshot WHERE ok = 1"
            ).fetchone()
            first_date = None if first_row is None else first_row["d"]
            counts = {
                str(row["status"]): int(row["n"])
                for row in conn.execute(
                    "SELECT status, COUNT(*) AS n FROM video GROUP BY status"
                ).fetchall()
            }
            follower_row = conn.execute(
                """
                SELECT COUNT(*) AS total,
                       SUM(CASE WHEN up_follower IS NOT NULL THEN 1 ELSE 0 END) AS covered
                FROM video
                WHERE status IN ('active', 'unfaved') AND up_mid IS NOT NULL
                """
            ).fetchone()
            follower_total = int(follower_row["total"] or 0)
            follower_covered = int(follower_row["covered"] or 0)
            profile_count = int(conn.execute(
                "SELECT COUNT(*) FROM up_profile"
            ).fetchone()[0])
            preview_row = conn.execute(
                "SELECT v FROM kv WHERE k = 'preview:seeded'"
            ).fetchone()
            fans_start_date = None
            track_fans = [
                int(mid) for mid in cfg.track_fans if not isinstance(mid, bool)
            ]
            if track_fans:
                placeholders = ",".join("?" for _ in track_fans)
                row = conn.execute(
                    f"SELECT MIN(stat_date) AS d FROM up_stat "
                    f"WHERE ok = 1 AND mid IN ({placeholders})",
                    track_fans,
                ).fetchone()
                fans_start_date = None if row is None else row["d"]
            return {
                "data_date": latest,
                "snapshot_days": self._snapshot_days(conn),
                "snapshot_first_date": first_date,
                "fans_start_date": fans_start_date,
                "track_fans": track_fans,
                "schema_version": int(self.db.schema_version or 0),
                "preview": bool(preview_row and str(preview_row["v"]) == "1"),
                "counts": counts,
                "library": {
                    "active_videos": counts.get("active", 0),
                    "unfaved_videos": counts.get("unfaved", 0),
                    "removed_videos": counts.get("removed", 0),
                },
                "analytics": {
                    "min_duration_seconds": cfg.analytics.min_duration_seconds,
                    "maturity_days": cfg.analytics.maturity_days,
                    "min_view_for_ranking": cfg.analytics.min_view_for_ranking,
                    "video_shrink_minutes": cfg.analytics.video_shrink_minutes,
                    "shrink_minutes": cfg.analytics.shrink_minutes,
                    "home_growth_days": cfg.analytics.home_growth_days,
                    "growth_rate_prior": cfg.analytics.growth_rate_prior,
                },
                "heat_weights": {
                    metric: float(value)
                    for metric, value in (cfg.heat_weights or {}).items()
                },
                "followers": {
                    "videos_total": follower_total,
                    "videos_covered": follower_covered,
                    "videos_missing": max(0, follower_total - follower_covered),
                    "ups_tracked": profile_count,
                },
                "folders": [
                    {
                        "year": folder.year,
                        "media_id": folder.media_id,
                        "title": folder.title,
                    }
                    for folder in cfg.folders
                ],
            }
        return self.db.read(run)

    # ------------------------------------------------------------ stats

    def stats(
        self,
        *,
        days: int = 30,
        pending: Sequence[Mapping[str, Any]] | None = None,
    ) -> dict[str, Any]:
        """访问统计：区间内每日 PV/UV 与 Top 页面。

        ``pending`` 为内存缓冲中尚未落库的上报行；与已落库数据按
        (日期, IP, 路由) 合并去重后统一聚合，因此页面基本是实时可见的。
        """
        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            cfg = self.config_store.get()
            today = self._today()
            start = _date_minus(today, int(days) - 1)
            rows: list[Mapping[str, Any]] = [
                dict(row) for row in conn.execute(
                    "SELECT stat_date, ip, path, hits, first_at, last_at FROM visit "
                    "WHERE stat_date >= ? AND stat_date <= ?",
                    (start, today),
                ).fetchall()
            ]
            if pending:
                rows.extend(dict(row) for row in pending)

            buckets: dict[tuple[str, str, str], list[int]] = {}
            for row in rows:
                day = str(row["stat_date"])
                ip = str(row["ip"])
                path = str(row["path"])
                hits = int(row.get("hits") or 0)
                first = int(row.get("first_at") or 0)
                last = int(row.get("last_at") or 0)
                bucket = buckets.get((day, ip, path))
                if bucket is None:
                    buckets[(day, ip, path)] = [hits, first, last]
                else:
                    bucket[0] += hits
                    bucket[1] = min(bucket[1], first) if first else bucket[1]
                    bucket[2] = max(bucket[2], last)

            dates = _date_range(start, today)
            daily_pv = {day: 0 for day in dates}
            daily_uv: dict[str, set[str]] = {day: set() for day in dates}
            page_pv: dict[str, int] = {}
            page_uv: dict[str, set[str]] = {}
            total_pv = 0
            total_ips: set[str] = set()
            for (day, ip, path), bucket in buckets.items():
                if day not in daily_pv:
                    continue
                daily_pv[day] += bucket[0]
                daily_uv[day].add(ip)
                total_pv += bucket[0]
                total_ips.add(ip)
                page_pv[path] = page_pv.get(path, 0) + bucket[0]
                page_uv.setdefault(path, set()).add(ip)

            top_pages = [
                {"path": path, "pv": hits, "uv": len(page_uv[path])}
                for path, hits in sorted(
                    page_pv.items(), key=lambda kv: (-kv[1], kv[0])
                )[:10]
            ]
            return {
                "enabled": bool(cfg.web.track.enabled),
                "days": int(days),
                "range": {"start": start, "end": today},
                "today": today,
                "today_pv": daily_pv.get(today, 0),
                "today_uv": len(daily_uv.get(today, set())),
                "total_pv": total_pv,
                "total_uv": len(total_ips),
                "series": {
                    "dates": dates,
                    "pv": [daily_pv[day] for day in dates],
                    "uv": [len(daily_uv[day]) for day in dates],
                },
                "top_pages": top_pages,
                "pending": bool(pending),
            }
        return self.db.read(run)

    def candidates(self) -> dict[str, Any]:
        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            cfg = self.config_store.get()
            now = int(self._now())
            ttl = int(cfg.schedule.candidate_ttl_days)
            rows = conn.execute(
                "SELECT * FROM candidate ORDER BY "
                "CASE WHEN decision IS NULL THEN 0 ELSE 1 END, "
                "COALESCE(score, 0) DESC, first_seen DESC, aid"
            ).fetchall()
            items = []
            for row in rows:
                data = dict(row)
                first_seen = int(data.get("first_seen") or 0)
                expires_at = first_seen + ttl * 86400
                items.append({
                    **data,
                    "cover": normalize_cover(data.get("cover")),
                    "expires_at": expires_at,
                    "days_left": max(0, math.ceil((expires_at - now) / 86400)),
                    "url": (
                        f"https://www.bilibili.com/video/{data.get('bvid')}"
                        if data.get("bvid") else None
                    ),
                })
            pending = sum(1 for item in items if item.get("decision") is None)
            return {
                "items": items,
                "total": len(items),
                "pending": pending,
                "ttl_days": ttl,
                "telegram_enabled": bool(cfg.telegram.enabled),
                "now": now,
            }
        return self.db.read(run)

    # ------------------------------------------------------------ system

    def system(self) -> dict[str, Any]:
        def run(conn: sqlite3.Connection) -> dict[str, Any]:
            cfg = self.config_store.get()
            latest = self._latest_date(conn)
            snapshot_days = self._snapshot_days(conn)
            health = {"total": 0, "ok": 0, "failed": 0}
            if latest:
                rows = conn.execute(
                    "SELECT ok, COUNT(*) AS n FROM snapshot WHERE stat_date = ? GROUP BY ok",
                    (latest,),
                ).fetchall()
                by_ok = {int(row["ok"]): int(row["n"]) for row in rows}
                health = {
                    "total": sum(by_ok.values()),
                    "ok": by_ok.get(1, 0),
                    "failed": by_ok.get(0, 0),
                }
            blacklist = [
                dict(row) for row in conn.execute(
                    """
                    SELECT b.*,
                      (SELECT COUNT(*) FROM video v WHERE v.up_mid = b.mid
                       AND v.status IN ('active','unfaved')) AS video_count
                    FROM up_blacklist b ORDER BY b.created_at, b.mid
                    """
                ).fetchall()
            ]
            tasks = self._task_records(conn)
            running = self._task_running(conn)
            now = self._now()
            discover_interval = int(cfg.schedule.discover_minutes) * 60
            recent_interval = int(cfg.schedule.recent_snapshot_minutes) * 60
            next_runs = {
                "fans": int(next_daily_run(now, cfg.schedule.snapshot_at)),
                "snapshot": int(next_daily_run(now, cfg.schedule.snapshot_at)),
                "snapshot_seed": int(next_daily_run(now, cfg.schedule.seed_at)),
                "discover": int(now + discover_interval),
                "snapshot_recent": int(now + recent_interval),
            }
            for name, key, at in (
                ("fans", "fans", cfg.schedule.snapshot_at),
                ("snapshot", "snapshot", cfg.schedule.snapshot_at),
                ("snapshot_seed", "snapshot_seed", cfg.schedule.seed_at),
                ("discover", "discovery", None),
                ("snapshot_recent", "snapshot_recent", None),
            ):
                record = tasks.get(key)
                if not (isinstance(record, dict) and record.get("finished_at")):
                    continue
                finished_at = int(record["finished_at"])
                if name in {"discover", "snapshot_recent"}:
                    # 只按上次完成推下一档；进行中由 running 展示，不按墙钟抢跑。
                    interval = (
                        recent_interval if name == "snapshot_recent"
                        else discover_interval
                    )
                    next_runs[name] = finished_at + interval
                else:
                    next_runs[name] = int(next_daily_run(finished_at, at))
            errors = [
                {"task": name, **record}
                for name, record in tasks.items()
                if isinstance(record, dict) and record.get("ok") is False
            ]
            counts = {str(row["status"]): int(row["n"]) for row in conn.execute(
                "SELECT status, COUNT(*) AS n FROM video GROUP BY status"
            ).fetchall()}
            return {
                "generated_at": int(now),
                "today": self._today(),
                "data_date": latest,
                "snapshot_days": snapshot_days,
                "snapshot_health": health,
                "counts": counts,
                "blacklist": blacklist,
                "tasks": tasks,
                "running": running,
                "errors": errors,
                "next_runs": next_runs,
                "auth": {
                    "ok": self._kv(conn, "auth:ok"),
                    "message": self._kv(conn, "auth:message"),
                    "last_check": self._kv(conn, "auth:last_check"),
                    "mid": self._kv(conn, "auth:mid"),
                },
                "backup": {
                    "last_completed": self._kv(conn, "snapshot:last_completed"),
                    "last_attempt": self._kv(conn, "snapshot:last_attempt"),
                    "ok": self._kv(conn, f"snapshot:backup:{latest}" if latest else "snapshot:backup"),
                    "error": self._kv(conn, f"snapshot:backup_error:{latest}" if latest else "snapshot:backup_error"),
                },
                "preview": self._kv(conn, "preview:seeded") == "1",
                "schedule": {
                    "discover_minutes": cfg.schedule.discover_minutes,
                    "snapshot_at": cfg.schedule.snapshot_at,
                    "seed_at": cfg.schedule.seed_at,
                    "candidate_ttl_days": cfg.schedule.candidate_ttl_days,
                    "recent_snapshot_minutes": cfg.schedule.recent_snapshot_minutes,
                    "recent_video_days": cfg.schedule.recent_video_days,
                },
                "telegram_enabled": bool(cfg.telegram.enabled),
            }
        return self.db.read(run)