1163 lines
44 KiB
Python
1163 lines
44 KiB
Python
# -*- 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,
|
||
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, 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),
|
||
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,
|
||
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)
|
||
|
||
# ------------------------------------------------------- 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 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)),
|
||
))
|
||
|
||
# ------------------------------------------------------------ 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 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_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)
|