feat(p1): add daily snapshots and fan tracking

Implement P1: seven-metric view snapshots, relation/stat fan snapshots, failure placeholders, resume by daily completion markers, Cookie login warnings, daily SQLite backups, Shanghai daily scheduling, and CLI snapshot/auth-check commands. Real read-only run captured 795/795 videos successfully over 1124s with a verified daily backup.
This commit is contained in:
kairee committed 2026-10-31 19:19:24 +08:00
1 parent ee289bfa56
commit 5c2ab7b187
20 files changed
+1680 -54

No files matched your search

+72 -1
View File
@@ -18,11 +18,21 @@ from dataclasses import dataclass
from typing import Any, Mapping, Protocol, Sequence
from .auth import make_headers
from .models import FavId, FavResource, FolderInfo, InfosResult
from .models import (
FanStat,
FavId,
FavResource,
FolderInfo,
InfosResult,
LoginState,
ViewStat,
)
BASE = "https://api.bilibili.com"
DEFAULT_CHUNK = 200
MAX_CHUNK = 400
# 视频不可访问/已删除:-404 为旧口径,62002/62012 为当前实测返回。
VIDEO_UNAVAILABLE_CODES = frozenset({-404, 62002, 62012})
class BilibiliError(RuntimeError):
@@ -360,3 +370,64 @@ class BilibiliClient:
media_count=count,
))
return out
# ------------------------------------------------------------ view / fan
def view(self, aid: int) -> ViewStat:
url = f"{BASE}/x/web-interface/view?aid={int(aid)}"
payload = self.get_json(url)
data = payload.get("data")
if not isinstance(data, Mapping):
raise BilibiliProtocolError("view data 不是对象")
try:
return ViewStat.from_api(data)
except (TypeError, ValueError) as exc:
raise BilibiliProtocolError(f"view 解析失败: {exc}") from exc
def relation_stat(self, mid: int) -> FanStat:
url = f"{BASE}/x/relation/stat?vmid={int(mid)}"
payload = self.get_json(url)
data = payload.get("data")
if not isinstance(data, Mapping):
raise BilibiliProtocolError("relation/stat data 不是对象")
try:
return FanStat.from_api(data)
except (TypeError, ValueError) as exc:
raise BilibiliProtocolError(f"relation/stat 解析失败: {exc}") from exc
def check_login(self) -> LoginState:
"""检查 Cookie 登录态;未配置 Cookie 时不发请求。"""
if not self.cookie:
return LoginState(
is_login=False,
mid=None,
uname=None,
code=0,
message="未配置 cookie",
)
url = f"{BASE}/x/web-interface/nav"
try:
payload = self.get_json(url, auth=True)
except BilibiliAPIError as exc:
return LoginState(
is_login=False, mid=None, uname=None,
code=exc.code, message=exc.message,
)
except BilibiliError as exc:
return LoginState(
is_login=False, mid=None, uname=None,
code=-1, message=str(exc),
)
data = payload.get("data") or {}
if not isinstance(data, Mapping):
raise BilibiliProtocolError("nav data 不是对象")
raw_mid = data.get("mid")
mid = raw_mid if isinstance(raw_mid, int) and not isinstance(raw_mid, bool) else None
uname = data.get("uname")
return LoginState(
is_login=bool(data.get("isLogin")),
mid=mid,
uname=str(uname) if isinstance(uname, str) else None,
code=0,
message=str(payload.get("message") or ""),
)
+80
View File
@@ -85,3 +85,83 @@ class InfosResult:
requested: tuple[int, ...]
items: dict[int, FavResource]
missing: tuple[int, ...]
@dataclass(frozen=True)
class ViewStat:
"""``x/web-interface/view`` 的七项统计。"""
aid: int
bvid: str | None
title: str
view: int
like: int
favorite: int
coin: int
share: int
reply: int
danmaku: int
def metrics(self) -> dict[str, int]:
return {
"view": self.view,
"like": self.like,
"favorite": self.favorite,
"coin": self.coin,
"share": self.share,
"reply": self.reply,
"danmaku": self.danmaku,
}
@classmethod
def from_api(cls, item: Mapping[str, Any]) -> "ViewStat":
aid = item.get("aid")
if isinstance(aid, bool) or not isinstance(aid, int):
raise ValueError("view data.aid 不是整数")
stat = item.get("stat")
if not isinstance(stat, Mapping):
raise ValueError("view data.stat 不是对象")
metrics: dict[str, int] = {}
for key in ("view", "like", "favorite", "coin", "share", "reply", "danmaku"):
value = stat.get(key)
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise ValueError(f"view stat.{key} 不是非负整数")
metrics[key] = value
bvid = item.get("bvid")
if not isinstance(bvid, str) or not bvid:
bvid = None
return cls(
aid=aid,
bvid=bvid,
title=str(item.get("title") or ""),
**metrics,
)
@dataclass(frozen=True)
class FanStat:
mid: int
follower: int
following: int
@classmethod
def from_api(cls, item: Mapping[str, Any]) -> "FanStat":
mid = item.get("mid")
follower = item.get("follower")
following = item.get("following", 0)
if isinstance(mid, bool) or not isinstance(mid, int):
raise ValueError("relation/stat data.mid 不是整数")
if isinstance(follower, bool) or not isinstance(follower, int) or follower < 0:
raise ValueError("relation/stat data.follower 不是非负整数")
if isinstance(following, bool) or not isinstance(following, int) or following < 0:
raise ValueError("relation/stat data.following 不是非负整数")
return cls(mid=mid, follower=follower, following=following)
@dataclass(frozen=True)
class LoginState:
is_login: bool
mid: int | None
uname: str | None
code: int
message: str