kairee
e2cdf997af
Implement P0 of the reaction dashboard: config loading with hot reload, SQLite migrations with backups, single-writer queue, read-only Bilibili fav client, folder import/diff sync, invalid-resource handling, blacklist seed, scheduler, and JSON redacted logging. Includes offline unit and transport-level contract tests; all existing and new tests pass.
442 lines
17 KiB
Python
442 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P0 数据访问层：收藏夹、视频与黑名单。
|
|
|
|
业务状态迁移集中在这里，服务层只负责请求与编排。所有写操作经 ``Database``
|
|
的单写队列串行执行。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Any, Iterable
|
|
|
|
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 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
|
|
|
|
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, 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.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),
|
|
cover = COALESCE(excluded.cover, video.cover),
|
|
folder_id = excluded.folder_id,
|
|
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.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)
|
|
VALUES (?, ?, ?, 1, ?)
|
|
ON CONFLICT(media_id, aid) DO UPDATE SET
|
|
active = 1,
|
|
last_seen = excluded.last_seen
|
|
""",
|
|
(folder_id, resource.aid, now, now),
|
|
)
|
|
return UpsertOutcome(
|
|
aid=resource.aid,
|
|
status=status,
|
|
previous_folder_id=previous_folder,
|
|
moved_folder=previous_folder is not None and previous_folder != folder_id,
|
|
)
|
|
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),
|
|
)
|
|
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),
|
|
)
|
|
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 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)
|
|
|
|
# ------------------------------------------------------------ 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)
|