Files
re-xxm/reaction_board/repository.py
T
bot-ci 9163f47b76 feat: home growth auto(7→1) and hourly recent-video snapshot
- Add analytics.home_growth_days (auto|1|7); API growth_days=auto tries 7 then 1
- Homepage shows active window and growth-board link; clearer empty reason
- New snapshot_recent scheduler task (schedule.recent_snapshot_minutes/days)
- run_recent_videos refreshes new aids only; never touches completed/backup
- Full 00:01 job keeps ok=1 skip (covers hourly refresh within 1h); mutex with recent
2026-11-02 10:00:02 +08:00

1214 lines
47 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""P0 数据访问层:收藏夹、视频与黑名单。
业务状态迁移集中在这里,服务层只负责请求与编排。所有写操作经 ``Database``
的单写队列串行执行。
"""
from __future__ import annotations
import sqlite3
import time
from dataclasses import dataclass
from typing import Any, Iterable, Mapping
from .config import FolderConfig
from .database import Database
@dataclass(frozen=True)
class UpsertOutcome:
aid: int
status: str
previous_folder_id: int | None
moved_folder: bool
@dataclass(frozen=True)
class IngestOutcome:
"""入库(候选确认或手动)结果。"""
aid: int
status: str
previous_status: str | None
created: bool
recovered: bool
@dataclass(frozen=True)
class FolderRow:
media_id: int
year: int
title: str
media_count: int | None
last_ids_at: int | None
last_full_at: int | None
enabled: bool
class Repository:
def __init__(self, db: Database, *, clock=time.time) -> None:
self.db = db
self._clock = clock
def _now(self) -> int:
return int(self._clock())
# ------------------------------------------------------------ folder
def ensure_folder(self, folder: FolderConfig, *, now: int | None = None) -> None:
now = self._now() if now is None else int(now)
def run(conn: sqlite3.Connection) -> None:
conn.execute(
"""
INSERT INTO folder(media_id, year, title, media_count,
last_ids_at, last_full_at, enabled)
VALUES (?, ?, ?, NULL, NULL, NULL, 1)
ON CONFLICT(media_id) DO UPDATE SET
year = excluded.year,
title = COALESCE(NULLIF(excluded.title, ''), folder.title),
enabled = 1
""",
(folder.media_id, folder.year, folder.title),
)
self.db.write(run)
def disable_folders_except(self, media_ids: Iterable[int]) -> None:
ids = [int(mid) for mid in media_ids]
def run(conn: sqlite3.Connection) -> None:
if ids:
placeholders = ",".join("?" for _ in ids)
conn.execute(
f"UPDATE folder SET enabled = 0 WHERE media_id NOT IN ({placeholders})",
ids,
)
else:
conn.execute("UPDATE folder SET enabled = 0")
self.db.write(run)
def finish_folder_sync(
self,
folder: FolderConfig,
*,
media_count: int,
full_refresh: bool,
now: int | None = None,
) -> None:
now = self._now() if now is None else int(now)
def run(conn: sqlite3.Connection) -> None:
conn.execute(
"""
INSERT INTO folder(media_id, year, title, media_count,
last_ids_at, last_full_at, enabled)
VALUES (?, ?, ?, ?, ?, ?, 1)
ON CONFLICT(media_id) DO UPDATE SET
year = excluded.year,
title = COALESCE(NULLIF(excluded.title, ''), folder.title),
media_count = excluded.media_count,
last_ids_at = excluded.last_ids_at,
last_full_at = CASE
WHEN ? THEN excluded.last_full_at
ELSE folder.last_full_at
END,
enabled = 1
""",
(
folder.media_id, folder.year, folder.title, int(media_count),
now, now if full_refresh else None, 1 if full_refresh else 0,
),
)
self.db.write(run)
def get_folder(self, media_id: int) -> FolderRow | None:
def run(conn: sqlite3.Connection) -> FolderRow | None:
row = conn.execute(
"SELECT media_id, year, title, media_count, last_ids_at, "
"last_full_at, enabled FROM folder WHERE media_id = ?",
(int(media_id),),
).fetchone()
if row is None:
return None
return FolderRow(
media_id=int(row["media_id"]),
year=int(row["year"]),
title=str(row["title"] or ""),
media_count=row["media_count"],
last_ids_at=row["last_ids_at"],
last_full_at=row["last_full_at"],
enabled=bool(row["enabled"]),
)
return self.db.read(run)
def list_folders(self) -> list[FolderRow]:
def run(conn: sqlite3.Connection) -> list[FolderRow]:
rows = conn.execute(
"SELECT media_id, year, title, media_count, last_ids_at, "
"last_full_at, enabled FROM folder ORDER BY year, media_id"
).fetchall()
return [
FolderRow(
media_id=int(row["media_id"]),
year=int(row["year"]),
title=str(row["title"] or ""),
media_count=row["media_count"],
last_ids_at=row["last_ids_at"],
last_full_at=row["last_full_at"],
enabled=bool(row["enabled"]),
)
for row in rows
]
return self.db.read(run)
def folder_active_aids(self, media_id: int) -> set[int]:
def run(conn: sqlite3.Connection) -> set[int]:
rows = conn.execute(
"SELECT aid FROM folder_resource WHERE media_id = ? AND active = 1",
(int(media_id),),
).fetchall()
return {int(row["aid"]) for row in rows}
return self.db.read(run)
def touch_folder_resources(self, media_id: int, *, now: int | None = None) -> None:
now = self._now() if now is None else int(now)
self.db.write(lambda conn: conn.execute(
"UPDATE folder_resource SET last_seen = ? WHERE media_id = ? AND active = 1",
(now, int(media_id)),
))
# ------------------------------------------------------------ video
@staticmethod
def _recompute_primary_folder(conn: sqlite3.Connection, aid: int) -> int | None:
"""按有效收藏关系重算主归属,避免依赖收藏夹遍历顺序。
优先使用逐夹 ``fav_time``(后加入者优先),其次 ``first_seen``,
最后用年度与 media_id 做确定性决胜。
"""
row = conn.execute(
"""
SELECT fr.media_id
FROM folder_resource fr
JOIN folder f ON f.media_id = fr.media_id
WHERE fr.aid = ? AND fr.active = 1 AND f.enabled = 1
ORDER BY COALESCE(fr.fav_time, 0) DESC,
fr.first_seen DESC,
f.year DESC,
fr.media_id DESC
LIMIT 1
""",
(int(aid),),
).fetchone()
if row is None:
return None
folder_id = int(row["media_id"])
conn.execute(
"UPDATE video SET folder_id = ? WHERE aid = ?", (folder_id, int(aid))
)
return folder_id
def upsert_from_fav(
self,
resource,
folder_id: int,
*,
was_added: bool,
now: int | None = None,
) -> UpsertOutcome:
now = self._now() if now is None else int(now)
folder_id = int(folder_id)
invalid = bool(getattr(resource, "invalid", False))
def run(conn: sqlite3.Connection) -> UpsertOutcome:
previous = conn.execute(
"SELECT status, folder_id FROM video WHERE aid = ?", (resource.aid,)
).fetchone()
previous_status = str(previous["status"]) if previous else None
previous_folder = (
int(previous["folder_id"])
if previous and previous["folder_id"] is not None else None
)
if previous_status is None:
status = "removed" if invalid else "active"
elif previous_status == "dropped":
status = "active" if (was_added and not invalid) else "dropped"
elif invalid:
status = "removed"
else:
status = "active"
conn.execute(
"""
INSERT INTO video(aid, bvid, title, up_mid, up_name, pubdate,
duration, pages, cover, folder_id, fav_time, status,
source, first_seen, last_sync, attr)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(aid) DO UPDATE SET
bvid = COALESCE(excluded.bvid, video.bvid),
title = CASE
WHEN (excluded.attr = 9 OR excluded.status = 'removed')
AND video.title != ''
THEN video.title
WHEN excluded.title != '' THEN excluded.title
ELSE video.title
END,
up_mid = COALESCE(excluded.up_mid, video.up_mid),
up_name = COALESCE(excluded.up_name, video.up_name),
pubdate = CASE
WHEN excluded.pubdate > 0 THEN excluded.pubdate
ELSE video.pubdate
END,
duration = COALESCE(excluded.duration, video.duration),
pages = COALESCE(excluded.pages, video.pages),
cover = COALESCE(excluded.cover, video.cover),
folder_id = CASE
WHEN video.folder_id IS NULL THEN excluded.folder_id
ELSE video.folder_id
END,
fav_time = COALESCE(excluded.fav_time, video.fav_time),
status = excluded.status,
source = COALESCE(video.source, excluded.source),
last_sync = excluded.last_sync,
attr = excluded.attr
""",
(
resource.aid,
resource.bvid,
resource.title or "(无标题)",
resource.up_mid,
resource.up_name,
int(resource.pubdate or 0),
resource.duration,
resource.pages,
resource.cover,
folder_id,
resource.fav_time,
status,
f"fav:{folder_id}",
now,
now,
resource.attr,
),
)
conn.execute(
"""
INSERT INTO folder_resource(media_id, aid, first_seen, active,
last_seen, fav_time)
VALUES (?, ?, ?, 1, ?, ?)
ON CONFLICT(media_id, aid) DO UPDATE SET
active = 1,
last_seen = excluded.last_seen,
fav_time = COALESCE(excluded.fav_time, folder_resource.fav_time)
""",
(folder_id, resource.aid, now, now, resource.fav_time),
)
primary = self._recompute_primary_folder(conn, resource.aid)
return UpsertOutcome(
aid=resource.aid,
status=status,
previous_folder_id=previous_folder,
moved_folder=previous_folder is not None and previous_folder != primary,
)
return self.db.write(run)
def mark_missing_resource(
self,
aid: int,
*,
folder_id: int,
bvid: str | None = None,
now: int | None = None,
) -> str:
"""处理 infos 静默缺失:写入/更新为 removed,保留历史。"""
now = self._now() if now is None else int(now)
aid = int(aid)
folder_id = int(folder_id)
def run(conn: sqlite3.Connection) -> str:
row = conn.execute("SELECT status FROM video WHERE aid = ?", (aid,)).fetchone()
previous_status = str(row["status"]) if row else None
status = "dropped" if previous_status == "dropped" else "removed"
conn.execute(
"""
INSERT INTO video(aid, bvid, title, up_mid, up_name, pubdate,
duration, cover, folder_id, fav_time, status,
source, first_seen, last_sync, attr)
VALUES (?, ?, '(资源已失效)', NULL, NULL, 0, NULL, NULL, ?, NULL,
?, ?, ?, ?, 9)
ON CONFLICT(aid) DO UPDATE SET
bvid = COALESCE(video.bvid, excluded.bvid),
status = ?,
last_sync = excluded.last_sync,
attr = 9
""",
(aid, bvid, folder_id, status, f"fav:{folder_id}", now, now, status),
)
conn.execute(
"""
INSERT INTO folder_resource(media_id, aid, first_seen, active, last_seen)
VALUES (?, ?, ?, 1, ?)
ON CONFLICT(media_id, aid) DO UPDATE SET
active = 1,
last_seen = excluded.last_seen
""",
(folder_id, aid, now, now),
)
self._recompute_primary_folder(conn, aid)
return status
return self.db.write(run)
def mark_resource_removed(
self, media_id: int, aid: int, *, now: int | None = None
) -> str | None:
"""资源从收藏夹移除:active -> unfaved,其他(dropped/removed)保持不变。"""
now = self._now() if now is None else int(now)
media_id, aid = int(media_id), int(aid)
def run(conn: sqlite3.Connection) -> str | None:
conn.execute(
"UPDATE folder_resource SET active = 0, last_seen = ? "
"WHERE media_id = ? AND aid = ?",
(now, media_id, aid),
)
self._recompute_primary_folder(conn, aid)
other = conn.execute(
"""
SELECT 1
FROM folder_resource fr
JOIN folder f ON f.media_id = fr.media_id
WHERE fr.aid = ? AND fr.active = 1 AND f.enabled = 1
LIMIT 1
""",
(aid,),
).fetchone()
if other is None:
conn.execute(
"UPDATE video SET status = 'unfaved', last_sync = ? "
"WHERE aid = ? AND status IN ('active', 'unfaved')",
(now, aid),
)
row = conn.execute("SELECT status FROM video WHERE aid = ?", (aid,)).fetchone()
return str(row["status"]) if row else None
return self.db.write(run)
def set_video_status(self, aid: int, status: str, *, now: int | None = None) -> None:
now = self._now() if now is None else int(now)
self.db.write(lambda conn: conn.execute(
"UPDATE video SET status = ?, last_sync = ? WHERE aid = ?",
(status, now, int(aid)),
))
def get_video(self, aid: int) -> dict[str, Any] | None:
def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
row = conn.execute("SELECT * FROM video WHERE aid = ?", (int(aid),)).fetchone()
return dict(row) if row else None
return self.db.read(run)
def list_videos(self) -> list[dict[str, Any]]:
def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
rows = conn.execute("SELECT * FROM video ORDER BY aid").fetchall()
return [dict(row) for row in rows]
return self.db.read(run)
def ingest_video(
self,
*,
aid: int,
bvid: str | None,
title: str,
up_mid: int | None,
up_name: str | None,
pubdate: int,
duration: int | None = None,
pages: int | None = None,
cover: str | None = None,
up_face: str | None = None,
source: str = "manual",
now: int | None = None,
) -> IngestOutcome:
"""把一条视频写入库并置为 ``active``(候选确认 / Telegram 手动入库共用)。
不写 B 站收藏夹、不设置 ``folder_id``;同时删除同 aid 的候选行。
既有的 ``removed`` / ``dropped`` / ``unfaved`` 状态都会被重新激活。
"""
now = self._now() if now is None else int(now)
aid = int(aid)
def run(conn: sqlite3.Connection) -> IngestOutcome:
previous = conn.execute(
"SELECT status FROM video WHERE aid = ?", (aid,)
).fetchone()
previous_status = str(previous["status"]) if previous else None
conn.execute(
"""
INSERT INTO video(aid, bvid, title, up_mid, up_name, pubdate,
duration, pages, cover, up_face, folder_id, fav_time, status,
source, first_seen, last_sync, attr)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, 'active', ?, ?, ?, 0)
ON CONFLICT(aid) DO UPDATE SET
bvid = COALESCE(excluded.bvid, video.bvid),
title = CASE
WHEN excluded.title != '' THEN excluded.title
ELSE video.title
END,
up_mid = COALESCE(excluded.up_mid, video.up_mid),
up_name = COALESCE(excluded.up_name, video.up_name),
pubdate = CASE
WHEN excluded.pubdate > 0 THEN excluded.pubdate
ELSE video.pubdate
END,
duration = COALESCE(excluded.duration, video.duration),
pages = COALESCE(excluded.pages, video.pages),
cover = COALESCE(excluded.cover, video.cover),
up_face = COALESCE(excluded.up_face, video.up_face),
status = 'active',
source = COALESCE(video.source, excluded.source),
last_sync = excluded.last_sync
""",
(
aid,
bvid,
title or "(无标题)",
up_mid,
up_name,
int(pubdate or 0),
None if duration is None else int(duration),
None if pages is None else int(pages),
cover,
up_face,
source,
now,
now,
),
)
conn.execute("DELETE FROM candidate WHERE aid = ?", (aid,))
return IngestOutcome(
aid=aid,
status="active",
previous_status=previous_status,
created=previous is None,
recovered=previous_status in {"removed", "dropped", "unfaved"},
)
return self.db.write(run)
def apply_video_up_face(self, aid: int, face: str | None) -> None:
"""用 ``view`` 返回的 UP 头像刷新视频行;仅在变化时写入。"""
if not isinstance(face, str) or not face.strip():
return
face = face.strip()
self.db.write(lambda conn: conn.execute(
"UPDATE video SET up_face = ? WHERE aid = ? AND (up_face IS NULL OR up_face != ?)",
(face, int(aid), face),
))
# ------------------------------------------------------- up profile
def apply_up_follower(
self,
*,
mid: int,
name: str | None,
follower: int,
following: int | None = None,
now: int | None = None,
backfilled: bool = False,
live_since: int | None = None,
) -> int:
"""写入 UP 粉丝快照,并补齐该 UP 尚未记录粉丝数的视频。
``follower_first`` 只在首次插入时写入;``follower_latest`` 每次刷新都会
更新。视频上的 ``up_follower`` 是入库时快照,已有值不会被覆盖。传入
``live_since`` 时,只有 ``first_seen >= live_since`` 的视频视为实时采集,
更早的视频标记为回填。返回本次补齐的视频数。
"""
now = self._now() if now is None else int(now)
mid = int(mid)
follower = max(0, int(follower))
following_value = None if following is None else max(0, int(following))
backfill_flag = 1 if backfilled else 0
def run(conn: sqlite3.Connection) -> int:
conn.execute(
"""
INSERT INTO up_profile(mid, name, follower_first, following_first,
first_captured_at, follower_latest, last_refreshed)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(mid) DO UPDATE SET
name = COALESCE(NULLIF(excluded.name, ''), up_profile.name),
follower_latest = excluded.follower_latest,
last_refreshed = excluded.last_refreshed
""",
(mid, name, follower, following_value, now, follower, now),
)
if live_since is None:
cursor = conn.execute(
"""
UPDATE video
SET up_follower = ?, up_follower_at = ?, up_follower_backfilled = ?
WHERE up_mid = ? AND up_follower IS NULL
AND status IN ('active', 'unfaved')
""",
(follower, now, backfill_flag, mid),
)
else:
cursor = conn.execute(
"""
UPDATE video
SET up_follower = ?, up_follower_at = ?,
up_follower_backfilled = CASE
WHEN first_seen >= ? THEN 0 ELSE 1
END
WHERE up_mid = ? AND up_follower IS NULL
AND status IN ('active', 'unfaved')
""",
(follower, now, int(live_since), mid),
)
return max(cursor.rowcount, 0)
return self.db.write(run)
def get_up_profile(self, mid: int) -> dict[str, Any] | None:
def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
row = conn.execute(
"SELECT * FROM up_profile WHERE mid = ?", (int(mid),)
).fetchone()
return dict(row) if row else None
return self.db.read(run)
def up_mids_missing_follower(self) -> list[dict[str, Any]]:
"""追踪中但仍有视频缺少入库粉丝数的 UP。"""
def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
rows = conn.execute(
"""
SELECT up_mid, COALESCE(MAX(up_name), '') AS up_name,
COUNT(*) AS missing
FROM video
WHERE status IN ('active', 'unfaved')
AND up_mid IS NOT NULL
AND up_follower IS NULL
GROUP BY up_mid
ORDER BY up_mid
"""
).fetchall()
return [dict(row) for row in rows]
return self.db.read(run)
def up_follower_coverage(self) -> dict[str, int]:
"""返回追踪视频的粉丝快照覆盖情况。"""
def run(conn: sqlite3.Connection) -> dict[str, int]:
row = conn.execute(
"""
SELECT COUNT(*) AS total,
SUM(CASE WHEN up_follower IS NOT NULL THEN 1 ELSE 0 END) AS covered,
COUNT(DISTINCT CASE WHEN up_follower IS NOT NULL
THEN up_mid END) AS ups_covered
FROM video
WHERE status IN ('active', 'unfaved')
AND up_mid IS NOT NULL
"""
).fetchone()
total = int(row["total"] or 0)
covered = int(row["covered"] or 0)
return {
"total": total,
"covered": covered,
"missing": max(0, total - covered),
"ups_covered": int(row["ups_covered"] or 0),
}
return self.db.read(run)
def folder_resources(self, media_id: int) -> list[dict[str, Any]]:
def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
rows = conn.execute(
"SELECT * FROM folder_resource WHERE media_id = ? ORDER BY aid",
(int(media_id),),
).fetchall()
return [dict(row) for row in rows]
return self.db.read(run)
# ------------------------------------------------------------ snapshot
def video_aids_for_snapshot(self) -> list[int]:
def run(conn: sqlite3.Connection) -> list[int]:
rows = conn.execute(
"SELECT aid FROM video WHERE status IN ('active', 'unfaved') "
"ORDER BY aid"
).fetchall()
return [int(row["aid"]) for row in rows]
return self.db.read(run)
def video_aids_recent_for_snapshot(self, *, since_pubdate: int) -> list[int]:
"""近 N 天新视频:按 pubdate 墙钟窗口(since_pubdate = now - days*86400)。"""
def run(conn: sqlite3.Connection) -> list[int]:
rows = conn.execute(
"SELECT aid FROM video "
"WHERE status IN ('active', 'unfaved') "
"AND pubdate IS NOT NULL AND pubdate >= ? "
"ORDER BY aid",
(int(since_pubdate),),
).fetchall()
return [int(row["aid"]) for row in rows]
return self.db.read(run)
def pending_snapshot_aids(self, stat_date: str) -> list[int]:
"""当天仍缺成功快照的追踪视频;用于新入库/恢复后的同日补采。"""
def run(conn: sqlite3.Connection) -> list[int]:
rows = conn.execute(
"""
SELECT v.aid
FROM video v
WHERE v.status IN ('active', 'unfaved')
AND NOT EXISTS (
SELECT 1 FROM snapshot s
WHERE s.stat_date = ? AND s.aid = v.aid AND s.ok = 1
)
ORDER BY v.aid
""",
(stat_date,),
).fetchall()
return [int(row["aid"]) for row in rows]
return self.db.read(run)
def get_snapshot(self, stat_date: str, aid: int) -> dict[str, Any] | None:
def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
row = conn.execute(
"SELECT * FROM snapshot WHERE stat_date = ? AND aid = ?",
(stat_date, int(aid)),
).fetchone()
return dict(row) if row else None
return self.db.read(run)
def upsert_snapshot(
self,
stat_date: str,
aid: int,
*,
metrics: Mapping[str, int] | None,
ok: bool,
err: str | None,
captured_at: int,
src: str = "view",
) -> None:
values = metrics or {}
def run(conn: sqlite3.Connection) -> None:
conn.execute(
"""
INSERT INTO snapshot(stat_date, aid, "view", "like", favorite, coin,
share, reply, danmaku, captured_at, ok, err, src)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(stat_date, aid) DO UPDATE SET
"view" = excluded."view",
"like" = excluded."like",
favorite = excluded.favorite,
coin = excluded.coin,
share = excluded.share,
reply = excluded.reply,
danmaku = excluded.danmaku,
captured_at = excluded.captured_at,
ok = excluded.ok,
err = excluded.err,
src = excluded.src
""",
(
stat_date,
int(aid),
values.get("view"),
values.get("like"),
values.get("favorite"),
values.get("coin"),
values.get("share"),
values.get("reply"),
values.get("danmaku"),
int(captured_at),
1 if ok else 0,
err,
src,
),
)
self.db.write(run)
def snapshot_counts(self, stat_date: str) -> dict[str, int]:
def run(conn: sqlite3.Connection) -> dict[str, int]:
rows = conn.execute(
"SELECT ok, COUNT(*) AS n FROM snapshot WHERE stat_date = ? GROUP BY ok",
(stat_date,),
).fetchall()
by_ok = {int(row["ok"]): int(row["n"]) for row in rows}
return {
"total": sum(by_ok.values()),
"ok": by_ok.get(1, 0),
"failed": by_ok.get(0, 0),
}
return self.db.read(run)
def upsert_up_stat(
self,
stat_date: str,
mid: int,
*,
follower: int | None,
following: int | None,
ok: bool,
captured_at: int,
) -> None:
self.db.write(lambda conn: conn.execute(
"""
INSERT INTO up_stat(stat_date, mid, follower, following, captured_at, ok)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(stat_date, mid) DO UPDATE SET
follower = excluded.follower,
following = excluded.following,
captured_at = excluded.captured_at,
ok = excluded.ok
""",
(stat_date, int(mid), follower, following, int(captured_at), 1 if ok else 0),
))
def get_up_stat(self, stat_date: str, mid: int) -> dict[str, Any] | None:
def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
row = conn.execute(
"SELECT * FROM up_stat WHERE stat_date = ? AND mid = ?",
(stat_date, int(mid)),
).fetchone()
return dict(row) if row else None
return self.db.read(run)
# ------------------------------------------------------------ kv
def kv_get(self, key: str) -> str | None:
def run(conn: sqlite3.Connection) -> str | None:
row = conn.execute("SELECT v FROM kv WHERE k = ?", (key,)).fetchone()
return None if row is None else str(row["v"])
return self.db.read(run)
def kv_set(self, key: str, value: str) -> None:
self.db.write(lambda conn: conn.execute(
"INSERT INTO kv(k, v) VALUES (?, ?) "
"ON CONFLICT(k) DO UPDATE SET v = excluded.v",
(key, str(value)),
))
def kv_delete(self, key: str) -> None:
self.db.write(lambda conn: conn.execute(
"DELETE FROM kv WHERE k = ?", (key,),
))
# ------------------------------------------------------------ cache version
@staticmethod
def cache_version_key(group: str) -> str:
return f"cache:v:{group}"
def cache_version(self, group: str) -> int:
def run(conn: sqlite3.Connection) -> int:
row = conn.execute(
"SELECT v FROM kv WHERE k = ?", (self.cache_version_key(group),)
).fetchone()
if row is None:
return 0
try:
return int(str(row["v"]))
except ValueError:
return 0
return self.db.read(run)
def bump_cache_version(self, group: str, *, now: int | None = None) -> int:
now = self._now() if now is None else int(now)
key = self.cache_version_key(group)
def run(conn: sqlite3.Connection) -> int:
row = conn.execute("SELECT v FROM kv WHERE k = ?", (key,)).fetchone()
try:
value = int(str(row["v"])) + 1 if row is not None else 1
except ValueError:
value = 1
conn.execute(
"INSERT INTO kv(k, v) VALUES (?, ?) "
"ON CONFLICT(k) DO UPDATE SET v = excluded.v",
(key, str(value)),
)
return value
return self.db.write(run)
# ------------------------------------------------------------ candidate
def known_aid(self, aid: int) -> bool:
def run(conn: sqlite3.Connection) -> bool:
row = conn.execute(
"SELECT 1 FROM video WHERE aid = ? "
"UNION ALL SELECT 1 FROM seen_aid WHERE aid = ? LIMIT 1",
(int(aid), int(aid)),
).fetchone()
return row is not None
return self.db.read(run)
def record_seen(
self, aid: int, *, source: str, now: int | None = None
) -> None:
now = self._now() if now is None else int(now)
self.db.write(lambda conn: conn.execute(
"INSERT INTO seen_aid(aid, first_seen, source) VALUES (?, ?, ?) "
"ON CONFLICT(aid) DO NOTHING",
(int(aid), now, source),
))
def upsert_candidate(
self,
*,
aid: int,
bvid: str | None,
title: str,
up_mid: int | None,
up_name: str | None,
pubdate: int,
cover: str | None,
score: int,
channel: str,
duration: int = 0,
play: int = 0,
like: int = 0,
favorite: int = 0,
danmaku: int = 0,
review: int = 0,
now: int | None = None,
) -> dict[str, Any]:
now = self._now() if now is None else int(now)
def run(conn: sqlite3.Connection) -> dict[str, Any]:
conn.execute(
"""
INSERT INTO candidate(aid, bvid, title, up_mid, up_name, pubdate,
cover, score, channel, first_seen,
notified_at, decision, decided_at,
duration, play, like, favorite, danmaku, review)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, NULL,
?, ?, ?, ?, ?, ?)
ON CONFLICT(aid) DO UPDATE SET
bvid = COALESCE(excluded.bvid, candidate.bvid),
title = COALESCE(NULLIF(excluded.title, ''), candidate.title),
up_mid = COALESCE(excluded.up_mid, candidate.up_mid),
up_name = COALESCE(excluded.up_name, candidate.up_name),
pubdate = CASE WHEN excluded.pubdate > 0
THEN excluded.pubdate ELSE candidate.pubdate END,
cover = COALESCE(excluded.cover, candidate.cover),
score = excluded.score,
channel = excluded.channel,
duration = excluded.duration,
play = excluded.play,
"like" = excluded."like",
favorite = excluded.favorite,
danmaku = excluded.danmaku,
review = excluded.review
""",
(
int(aid), bvid, title, up_mid, up_name, int(pubdate),
cover, float(score), channel, now,
int(duration or 0), int(play or 0), int(like or 0),
int(favorite or 0), int(danmaku or 0), int(review or 0),
),
)
row = conn.execute(
"SELECT * FROM candidate WHERE aid = ?", (int(aid),)
).fetchone()
return dict(row) if row else {}
return self.db.write(run)
def get_candidate(self, aid: int) -> dict[str, Any] | None:
def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
row = conn.execute(
"SELECT * FROM candidate WHERE aid = ?", (int(aid),)
).fetchone()
return dict(row) if row else None
return self.db.read(run)
def list_candidates(self, decision: str | None = None) -> list[dict[str, Any]]:
def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
if decision is None:
rows = conn.execute(
"SELECT * FROM candidate ORDER BY COALESCE(score, 0) DESC, "
"first_seen DESC, aid"
).fetchall()
else:
rows = conn.execute(
"SELECT * FROM candidate WHERE decision = ? "
"ORDER BY COALESCE(score, 0) DESC, first_seen DESC, aid",
(decision,),
).fetchall()
return [dict(row) for row in rows]
return self.db.read(run)
def list_pending_candidates(self) -> list[dict[str, Any]]:
"""待决策候选(``decision IS NULL``),按分数倒序。"""
def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
rows = conn.execute(
"SELECT * FROM candidate WHERE decision IS NULL "
"ORDER BY COALESCE(score, 0) DESC, first_seen DESC, aid"
).fetchall()
return [dict(row) for row in rows]
return self.db.read(run)
def mark_candidate_decision(
self, aid: int, decision: str, *, now: int | None = None
) -> None:
now = self._now() if now is None else int(now)
self.db.write(lambda conn: conn.execute(
"UPDATE candidate SET decision = ?, decided_at = ? WHERE aid = ?",
(decision, now, int(aid)),
))
def reject_pending_for_up(self, up_mid: int, *, now: int | None = None) -> int:
now = self._now() if now is None else int(now)
def run(conn: sqlite3.Connection) -> int:
cur = conn.execute(
"UPDATE candidate SET decision = 'rejected', decided_at = ? "
"WHERE up_mid = ? AND decision IS NULL",
(now, int(up_mid)),
)
return max(cur.rowcount, 0)
return self.db.write(run)
def delete_candidate(self, aid: int) -> None:
self.db.write(lambda conn: conn.execute(
"DELETE FROM candidate WHERE aid = ?", (int(aid),)
))
def delete_all_candidates(self, *, pending_only: bool = False) -> int:
"""删除候选:``pending_only=True`` 只删待决策,否则清空整表;返回删除数。"""
def run(conn: sqlite3.Connection) -> int:
if pending_only:
cur = conn.execute("DELETE FROM candidate WHERE decision IS NULL")
else:
cur = conn.execute("DELETE FROM candidate")
return max(cur.rowcount, 0)
return self.db.write(run)
def delete_expired_candidates(
self, *, ttl_days: int, now: int | None = None
) -> int:
now = self._now() if now is None else int(now)
cutoff = now - int(ttl_days) * 86400
def run(conn: sqlite3.Connection) -> int:
cur = conn.execute(
"DELETE FROM candidate WHERE decision IS NULL AND first_seen < ?",
(cutoff,),
)
return max(cur.rowcount, 0)
return self.db.write(run)
def mark_notified(
self,
aid: int,
*,
channel: str,
batch_id: str | None,
ok: bool,
err: str | None = None,
now: int | None = None,
) -> None:
now = self._now() if now is None else int(now)
def run(conn: sqlite3.Connection) -> None:
conn.execute(
"""
INSERT INTO notify(aid, channel, batch_id, sent_at, ok, err)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(aid, channel) DO UPDATE SET
batch_id = excluded.batch_id,
sent_at = excluded.sent_at,
ok = excluded.ok,
err = excluded.err
""",
(int(aid), channel, batch_id, now, 1 if ok else 0, err),
)
if ok:
conn.execute(
"UPDATE candidate SET notified_at = ? WHERE aid = ?",
(now, int(aid)),
)
self.db.write(run)
# ------------------------------------------------------------ fav action
def video_metadata(self, aids: Iterable[int]) -> dict[int, dict[str, Any]]:
ids = [int(aid) for aid in aids]
if not ids:
return {}
def run(conn: sqlite3.Connection) -> dict[int, dict[str, Any]]:
placeholders = ",".join("?" for _ in ids)
rows = conn.execute(
f"SELECT aid, up_mid, pubdate FROM video WHERE aid IN ({placeholders})",
ids,
).fetchall()
return {
int(row["aid"]): {
"up_mid": row["up_mid"],
"pubdate": int(row["pubdate"] or 0),
}
for row in rows
}
return self.db.read(run)
def folder_for_up(self, up_mid: int) -> int | None:
def run(conn: sqlite3.Connection) -> int | None:
row = conn.execute(
"SELECT folder_id FROM video "
"WHERE up_mid = ? AND folder_id IS NOT NULL "
"AND status IN ('active', 'unfaved') "
"ORDER BY pubdate ASC, aid ASC LIMIT 1",
(int(up_mid),),
).fetchone()
return int(row["folder_id"]) if row else None
return self.db.read(run)
def begin_fav_action(
self,
aid: int,
folder_id: int,
*,
now: int | None = None,
ttl: int = 86400,
) -> tuple[str, str]:
now = self._now() if now is None else int(now)
token = f"fav:{int(aid)}"
def run(conn: sqlite3.Connection) -> tuple[str, str]:
row = conn.execute(
"SELECT state, expires_at FROM fav_action WHERE token = ?",
(token,),
).fetchone()
if row is not None:
state = str(row["state"])
if state == "done":
return token, "done"
if state == "processing" and int(row["expires_at"]) > now:
return token, "busy"
conn.execute(
"""
INSERT INTO fav_action(token, aid, folder_id, state, created_at,
expires_at, done_at, position, err)
VALUES (?, ?, ?, 'processing', ?, ?, NULL, NULL, NULL)
ON CONFLICT(token) DO UPDATE SET
folder_id = excluded.folder_id,
state = 'processing',
expires_at = excluded.expires_at,
done_at = NULL,
position = NULL,
err = NULL
""",
(token, int(aid), int(folder_id), now, now + int(ttl)),
)
return token, "processing"
return self.db.write(run)
def finish_fav_action(
self, token: str, *, position: int, now: int | None = None
) -> None:
now = self._now() if now is None else int(now)
self.db.write(lambda conn: conn.execute(
"UPDATE fav_action SET state = 'done', done_at = ?, position = ?, err = NULL "
"WHERE token = ?",
(now, int(position), token),
))
def fail_fav_action(
self, token: str, *, error: str, now: int | None = None
) -> None:
now = self._now() if now is None else int(now)
self.db.write(lambda conn: conn.execute(
"UPDATE fav_action SET state = 'failed', done_at = ?, err = ? "
"WHERE token = ?",
(now, error, token),
))
def get_fav_action(self, token: str) -> dict[str, Any] | None:
def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
row = conn.execute(
"SELECT * FROM fav_action WHERE token = ?", (token,)
).fetchone()
return dict(row) if row else None
return self.db.read(run)
def expire_fav_actions(self, *, now: int | None = None) -> int:
now = self._now() if now is None else int(now)
def run(conn: sqlite3.Connection) -> int:
cur = conn.execute(
"UPDATE fav_action SET state = 'failed', done_at = ?, "
"err = '处理超时' WHERE state = 'processing' AND expires_at < ?",
(now, now),
)
return max(cur.rowcount, 0)
return self.db.write(run)
# ------------------------------------------------------------ blacklist
def seed_blacklist(self, mids: Iterable[int], *, now: int | None = None) -> int:
now = self._now() if now is None else int(now)
def run(conn: sqlite3.Connection) -> int:
inserted = 0
for mid in mids:
cur = conn.execute(
"INSERT OR IGNORE INTO up_blacklist(mid, name, reason, created_at) "
"VALUES (?, NULL, 'seed', ?)",
(int(mid), now),
)
inserted += max(cur.rowcount, 0)
return inserted
return self.db.write(run)
def add_blacklist(
self,
mid: int,
*,
name: str | None = None,
reason: str | None = None,
now: int | None = None,
) -> None:
now = self._now() if now is None else int(now)
self.db.write(lambda conn: conn.execute(
"""
INSERT INTO up_blacklist(mid, name, reason, created_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(mid) DO UPDATE SET
name = COALESCE(excluded.name, up_blacklist.name),
reason = COALESCE(excluded.reason, up_blacklist.reason)
""",
(int(mid), name, reason, now),
))
def is_blacklisted(self, mid: int) -> bool:
def run(conn: sqlite3.Connection) -> bool:
row = conn.execute(
"SELECT 1 FROM up_blacklist WHERE mid = ?", (int(mid),)
).fetchone()
return row is not None
return self.db.read(run)
def blacklist(self) -> list[dict[str, Any]]:
def run(conn: sqlite3.Connection) -> list[dict[str, Any]]:
rows = conn.execute(
"SELECT mid, name, reason, created_at FROM up_blacklist ORDER BY created_at, mid"
).fetchall()
return [dict(row) for row in rows]
return self.db.read(run)
# ------------------------------------------------------------ summary
def counts(self) -> dict[str, Any]:
def run(conn: sqlite3.Connection) -> dict[str, Any]:
status_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 status_rows}
folder_count = conn.execute("SELECT COUNT(*) FROM folder").fetchone()[0]
blacklist_count = conn.execute("SELECT COUNT(*) FROM up_blacklist").fetchone()[0]
candidate_count = conn.execute("SELECT COUNT(*) FROM candidate").fetchone()[0]
return {
"video_total": sum(by_status.values()),
"video_by_status": by_status,
"folders": int(folder_count),
"blacklist": int(blacklist_count),
"candidates": int(candidate_count),
}
return self.db.read(run)