# -*- 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)