561 lines
20 KiB
Python
561 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""只读 B 站 Web API 客户端（P0 范围）。
|
|
|
|
P0 只需要收藏夹读取能力：``ids`` / ``infos`` / ``folder/created/list-all``。
|
|
所有网络访问都经过 ``Transport`` 边界，测试可直接替换该边界，而不需要访问
|
|
真实 B 站。写操作（``deal`` / ``sort``）不在本模块实现。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import random
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
from typing import Any, Mapping, Protocol, Sequence
|
|
|
|
import httpx
|
|
|
|
from .auth import make_headers
|
|
from .models import (
|
|
FanStat,
|
|
FavId,
|
|
FavResource,
|
|
FolderInfo,
|
|
InfosResult,
|
|
LoginState,
|
|
VideoDetail,
|
|
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):
|
|
"""B 站访问层基础异常。"""
|
|
|
|
|
|
class TransportError(BilibiliError):
|
|
"""网络层错误，可重试。"""
|
|
|
|
retryable = True
|
|
|
|
|
|
class BilibiliHTTPError(BilibiliError):
|
|
def __init__(self, status: int, message: str = "") -> None:
|
|
self.status = status
|
|
self.message = message
|
|
self.retryable = status >= 500 or status == 429
|
|
super().__init__(f"HTTP {status}: {message}".rstrip())
|
|
|
|
def __str__(self) -> str:
|
|
return f"HTTP {self.status}: {self.message}".rstrip(": ")
|
|
|
|
|
|
class BilibiliAPIError(BilibiliError):
|
|
def __init__(self, code: int, message: str = "") -> None:
|
|
self.code = code
|
|
self.message = message
|
|
super().__init__(f"B 站接口业务错误 code={code}: {message}".rstrip(": "))
|
|
|
|
def __str__(self) -> str:
|
|
return f"B 站接口业务错误 code={self.code}: {self.message}".rstrip(": ")
|
|
|
|
|
|
class BilibiliProtocolError(BilibiliError):
|
|
"""响应结构与已知契约不一致。"""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HTTPResponse:
|
|
status: int
|
|
headers: Mapping[str, str]
|
|
body: bytes
|
|
|
|
def json(self) -> Any:
|
|
try:
|
|
return json.loads(self.body.decode("utf-8", "replace"))
|
|
except json.JSONDecodeError as exc:
|
|
raise BilibiliProtocolError(f"响应不是合法 JSON: {exc}") from exc
|
|
|
|
|
|
class Transport(Protocol):
|
|
def request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
headers: Mapping[str, str],
|
|
data: bytes | None = None,
|
|
timeout: float = 30,
|
|
) -> HTTPResponse: ...
|
|
|
|
|
|
class UrllibTransport:
|
|
"""基于 ``urllib.request`` 的默认传输实现。"""
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
headers: Mapping[str, str],
|
|
data: bytes | None = None,
|
|
timeout: float = 30,
|
|
) -> HTTPResponse:
|
|
req = urllib.request.Request(
|
|
url, data=data, headers=dict(headers), method=method
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return HTTPResponse(
|
|
status=int(resp.status),
|
|
headers=dict(resp.headers.items()),
|
|
body=resp.read(),
|
|
)
|
|
except urllib.error.HTTPError as exc:
|
|
try:
|
|
body = exc.read()
|
|
except Exception: # noqa: BLE001
|
|
body = b""
|
|
return HTTPResponse(
|
|
status=int(exc.code),
|
|
headers=dict(exc.headers.items()) if exc.headers else {},
|
|
body=body,
|
|
)
|
|
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
|
raise TransportError(str(exc)) from exc
|
|
|
|
|
|
class HttpxTransport:
|
|
"""基于 httpx 的连接池传输，默认用于 B 站请求。"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
timeout: float = 30,
|
|
client: httpx.Client | None = None,
|
|
) -> None:
|
|
self.timeout = timeout
|
|
self._client = client or httpx.Client(
|
|
timeout=httpx.Timeout(timeout, connect=10),
|
|
limits=httpx.Limits(
|
|
max_keepalive_connections=10,
|
|
keepalive_expiry=30,
|
|
),
|
|
follow_redirects=False,
|
|
trust_env=True,
|
|
)
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
headers: Mapping[str, str],
|
|
data: bytes | None = None,
|
|
timeout: float = 30,
|
|
) -> HTTPResponse:
|
|
try:
|
|
response = self._client.request(
|
|
method,
|
|
url,
|
|
headers=dict(headers),
|
|
content=data,
|
|
timeout=timeout,
|
|
)
|
|
except httpx.TransportError as exc:
|
|
raise TransportError(str(exc)) from exc
|
|
return HTTPResponse(
|
|
status=int(response.status_code),
|
|
headers=dict(response.headers),
|
|
body=response.content,
|
|
)
|
|
|
|
def close(self) -> None:
|
|
self._client.close()
|
|
|
|
|
|
class Pacer:
|
|
"""请求间随机等待；第一个请求不等待。"""
|
|
|
|
def __init__(
|
|
self,
|
|
min_delay: float = 1.5,
|
|
max_delay: float = 2.5,
|
|
*,
|
|
sleep=time.sleep,
|
|
rng: random.Random | None = None,
|
|
) -> None:
|
|
if min_delay < 0 or max_delay < min_delay:
|
|
raise ValueError("非法的请求间隔")
|
|
self.min_delay = min_delay
|
|
self.max_delay = max_delay
|
|
self._sleep = sleep
|
|
self._rng = rng or random.Random()
|
|
self._first = True
|
|
|
|
def wait(self) -> float:
|
|
if self._first:
|
|
self._first = False
|
|
return 0.0
|
|
delay = self._rng.uniform(self.min_delay, self.max_delay)
|
|
if delay > 0:
|
|
self._sleep(delay)
|
|
return delay
|
|
|
|
|
|
def _chunks(seq: Sequence[int], size: int):
|
|
for index in range(0, len(seq), size):
|
|
yield seq[index:index + size]
|
|
|
|
|
|
class BilibiliClient:
|
|
"""只读收藏夹 API 客户端。"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
cookie: str = "",
|
|
csrf: str = "",
|
|
mid: int = 0,
|
|
transport: Transport | None = None,
|
|
pacer: Pacer | None = None,
|
|
sleeper=time.sleep,
|
|
retries: int = 3,
|
|
backoff: float = 1.0,
|
|
timeout: float = 30,
|
|
logger=None,
|
|
) -> None:
|
|
if retries < 1:
|
|
raise ValueError("retries 必须 >= 1")
|
|
self.cookie = cookie or ""
|
|
self.csrf = csrf or ""
|
|
self.mid = mid
|
|
self.transport = transport or HttpxTransport()
|
|
self.pacer = pacer or Pacer(sleep=sleeper)
|
|
self._sleeper = sleeper
|
|
self.retries = retries
|
|
self.backoff = backoff
|
|
self.timeout = timeout
|
|
self.logger = logger
|
|
|
|
def close(self) -> None:
|
|
close = getattr(self.transport, "close", None)
|
|
if callable(close):
|
|
close()
|
|
|
|
# ------------------------------------------------------------ core
|
|
|
|
def _headers(self, auth: bool, referer: str | None = None) -> dict[str, str]:
|
|
return make_headers({"cookie": self.cookie}, auth=auth, referer=referer)
|
|
|
|
def _request_json(
|
|
self, url: str, *, auth: bool, referer: str | None = None
|
|
) -> dict[str, Any]:
|
|
self.pacer.wait()
|
|
resp = self.transport.request(
|
|
"GET", url, self._headers(auth, referer), data=None, timeout=self.timeout
|
|
)
|
|
if resp.status >= 400:
|
|
raise BilibiliHTTPError(resp.status, resp.body.decode("utf-8", "replace")[:200])
|
|
payload = resp.json()
|
|
if not isinstance(payload, dict):
|
|
raise BilibiliProtocolError("接口响应顶层不是 JSON 对象")
|
|
return payload
|
|
|
|
def _request_with_retry(
|
|
self, url: str, *, auth: bool, referer: str | None = None
|
|
) -> dict[str, Any]:
|
|
last: BilibiliError | None = None
|
|
for attempt in range(self.retries):
|
|
try:
|
|
return self._request_json(url, auth=auth, referer=referer)
|
|
except (TransportError, BilibiliHTTPError) as exc:
|
|
last = exc
|
|
if not exc.retryable or attempt >= self.retries - 1:
|
|
raise
|
|
delay = self.backoff * (2 ** attempt)
|
|
if self.logger is not None:
|
|
self.logger.warning(
|
|
"bili.retry",
|
|
attempt=attempt + 1,
|
|
delay=delay,
|
|
status=getattr(exc, "status", None),
|
|
)
|
|
if delay > 0:
|
|
self._sleeper(delay)
|
|
raise last or TransportError("未知网络错误")
|
|
|
|
@staticmethod
|
|
def _api_error(payload: Mapping[str, Any]) -> BilibiliAPIError:
|
|
code = payload.get("code")
|
|
if isinstance(code, bool) or not isinstance(code, int):
|
|
raise BilibiliProtocolError("接口响应缺少整数 code")
|
|
return BilibiliAPIError(code, str(payload.get("message") or ""))
|
|
|
|
def get_json(
|
|
self, url: str, *, auth: bool = False, referer: str | None = None
|
|
) -> dict[str, Any]:
|
|
"""GET JSON；匿名失败时自动带完整 Cookie 重试一次（与 favsort 一致）。"""
|
|
if auth:
|
|
payload = self._request_with_retry(url, auth=True, referer=referer)
|
|
if payload.get("code") != 0:
|
|
raise self._api_error(payload)
|
|
return payload
|
|
|
|
anon_error: BilibiliError | None = None
|
|
try:
|
|
payload = self._request_with_retry(url, auth=False, referer=referer)
|
|
if payload.get("code") == 0:
|
|
return payload
|
|
anon_error = self._api_error(payload)
|
|
except BilibiliError as exc:
|
|
anon_error = exc
|
|
|
|
if self.logger is not None and anon_error is not None:
|
|
self.logger.debug("bili.anon_fallback", error=str(anon_error))
|
|
payload = self._request_with_retry(url, auth=True, referer=referer)
|
|
if payload.get("code") != 0:
|
|
raise self._api_error(payload)
|
|
return payload
|
|
|
|
@staticmethod
|
|
def _data_list(payload: Mapping[str, Any], name: str) -> list[Any]:
|
|
"""区分“真实空数组”与“协议缺字段”。
|
|
|
|
实测空收藏夹返回 ``{"code":0,"data":[]}``；缺少 ``data`` 或为 ``null``
|
|
都不是已知契约，必须中止，避免被误判成空收藏夹后批量改状态。
|
|
"""
|
|
if "data" not in payload:
|
|
raise BilibiliProtocolError(f"{name} 响应缺少 data 字段")
|
|
data = payload["data"]
|
|
if not isinstance(data, list):
|
|
raise BilibiliProtocolError(f"{name} data 不是数组")
|
|
return data
|
|
|
|
# ------------------------------------------------------------ fav
|
|
|
|
def fav_ids(self, media_id: int) -> list[FavId]:
|
|
url = (
|
|
f"{BASE}/x/v3/fav/resource/ids?media_id={int(media_id)}&platform=web"
|
|
)
|
|
referer = (
|
|
f"https://space.bilibili.com/{self.mid}/favlist"
|
|
f"?fid={int(media_id)}&ftype=create"
|
|
)
|
|
payload = self.get_json(url, referer=referer)
|
|
data = self._data_list(payload, "fav/resource/ids")
|
|
out: list[FavId] = []
|
|
for index, item in enumerate(data):
|
|
if not isinstance(item, Mapping):
|
|
raise BilibiliProtocolError(f"fav/resource/ids data[{index}] 不是对象")
|
|
aid = item.get("id")
|
|
if isinstance(aid, bool) or not isinstance(aid, int):
|
|
raise BilibiliProtocolError(f"fav/resource/ids data[{index}].id 不是整数")
|
|
raw_type = item.get("type", 2)
|
|
if isinstance(raw_type, bool) or not isinstance(raw_type, int):
|
|
raise BilibiliProtocolError(f"fav/resource/ids data[{index}].type 不是整数")
|
|
bvid = item.get("bvid") or item.get("bv_id")
|
|
if not isinstance(bvid, str) or not bvid:
|
|
bvid = None
|
|
out.append(FavId(aid=aid, type=raw_type, bvid=bvid))
|
|
return out
|
|
|
|
def fav_infos(
|
|
self,
|
|
media_id: int,
|
|
aids: Sequence[int],
|
|
*,
|
|
folder_mid: int | None = None,
|
|
chunk_size: int = DEFAULT_CHUNK,
|
|
) -> InfosResult:
|
|
if not 1 <= chunk_size <= MAX_CHUNK:
|
|
raise ValueError(f"chunk_size 必须在 1..{MAX_CHUNK} 之间")
|
|
requested: list[int] = []
|
|
seen: set[int] = set()
|
|
for raw in aids:
|
|
aid = int(raw)
|
|
if aid not in seen:
|
|
seen.add(aid)
|
|
requested.append(aid)
|
|
if not requested:
|
|
return InfosResult(requested=(), items={}, missing=())
|
|
|
|
fmid = self.mid if folder_mid is None else int(folder_mid)
|
|
referer = (
|
|
f"https://space.bilibili.com/{fmid}/favlist"
|
|
f"?fid={int(media_id)}&ftype=create"
|
|
)
|
|
items: dict[int, FavResource] = {}
|
|
for chunk in _chunks(requested, chunk_size):
|
|
resources = ",".join(f"{aid}:2" for aid in chunk)
|
|
url = (
|
|
f"{BASE}/x/v3/fav/resource/infos"
|
|
f"?resources={urllib.parse.quote(resources)}"
|
|
f"&platform=web&folder_mid={fmid}&folder_id={int(media_id)}"
|
|
)
|
|
payload = self.get_json(url, referer=referer)
|
|
data = self._data_list(payload, "fav/resource/infos")
|
|
chunk_set = set(chunk)
|
|
for item in data:
|
|
if not isinstance(item, Mapping):
|
|
raise BilibiliProtocolError("fav/resource/infos item 不是对象")
|
|
try:
|
|
parsed = FavResource.from_api(item)
|
|
except (TypeError, ValueError) as exc:
|
|
raise BilibiliProtocolError(f"fav/resource/infos 解析失败: {exc}") from exc
|
|
if parsed.aid not in chunk_set:
|
|
raise BilibiliProtocolError(
|
|
f"fav/resource/infos 返回了未请求的 aid={parsed.aid}"
|
|
)
|
|
items[parsed.aid] = parsed
|
|
|
|
missing = tuple(aid for aid in requested if aid not in items)
|
|
return InfosResult(requested=tuple(requested), items=items, missing=missing)
|
|
|
|
def fav_folders(self, up_mid: int | None = None) -> list[FolderInfo]:
|
|
mid = self.mid if up_mid is None else int(up_mid)
|
|
url = f"{BASE}/x/v3/fav/folder/created/list-all?up_mid={mid}&jsonp=jsonp"
|
|
payload = self.get_json(
|
|
url,
|
|
auth=True,
|
|
referer=f"https://space.bilibili.com/{mid}/favlist",
|
|
)
|
|
data = payload.get("data") or {}
|
|
if not isinstance(data, Mapping):
|
|
raise BilibiliProtocolError("fav/folder/created/list-all data 不是对象")
|
|
rows = data.get("list")
|
|
if rows is None:
|
|
return []
|
|
if not isinstance(rows, list):
|
|
raise BilibiliProtocolError("fav/folder/created/list-all data.list 不是数组")
|
|
out: list[FolderInfo] = []
|
|
for index, row in enumerate(rows):
|
|
if not isinstance(row, Mapping):
|
|
raise BilibiliProtocolError(f"folder list[{index}] 不是对象")
|
|
media_id = row.get("id")
|
|
if isinstance(media_id, bool) or not isinstance(media_id, int):
|
|
raise BilibiliProtocolError(f"folder list[{index}].id 不是整数")
|
|
count = row.get("media_count", 0)
|
|
if isinstance(count, bool) or not isinstance(count, int):
|
|
count = 0
|
|
out.append(FolderInfo(
|
|
media_id=media_id,
|
|
title=str(row.get("title") or ""),
|
|
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 view_detail(
|
|
self, *, aid: int | None = None, bvid: str | None = None
|
|
) -> VideoDetail:
|
|
"""按 aid 或 bvid 取视频完整详情；两者必须且只能提供一个。"""
|
|
if (aid is None) == (bvid is None):
|
|
raise ValueError("view_detail 必须且只能提供 aid 或 bvid")
|
|
if bvid is not None:
|
|
url = f"{BASE}/x/web-interface/view?bvid={urllib.parse.quote(str(bvid))}"
|
|
else:
|
|
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 VideoDetail.from_api(data)
|
|
except (TypeError, ValueError) as exc:
|
|
raise BilibiliProtocolError(f"view 解析失败: {exc}") from exc
|
|
|
|
def resolve_short_link(self, url: str, *, max_hops: int = 5) -> str:
|
|
"""跟随 ``b23.tv`` 短链重定向，返回最终 URL。
|
|
|
|
只用于把短链还原成含 bvid/aid 的 bilibili 视频链接；不做分页或登录。
|
|
"""
|
|
current = str(url)
|
|
for _ in range(max(1, int(max_hops))):
|
|
host = (urllib.parse.urlparse(current).hostname or "").lower()
|
|
if not host.endswith("b23.tv"):
|
|
# 已经不是短链域名（例如已还原到 bilibili.com/video/BV…）直接返回。
|
|
return current
|
|
response = self.transport.request(
|
|
"GET", current, self._headers(False), data=None, timeout=self.timeout
|
|
)
|
|
headers = {
|
|
str(key).lower(): value for key, value in response.headers.items()
|
|
}
|
|
location = headers.get("location")
|
|
if response.status >= 400:
|
|
raise BilibiliHTTPError(
|
|
response.status, response.body.decode("utf-8", "replace")[:200]
|
|
)
|
|
if response.status in {301, 302, 303, 307, 308} and location:
|
|
current = urllib.parse.urljoin(current, location)
|
|
continue
|
|
return current
|
|
return current
|
|
|
|
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 ""),
|
|
)
|