From d2adb65e087ef683bb898b49d4a96b8cf15e1d37 Mon Sep 17 00:00:00 2001 From: kaireewu Date: Sun, 1 Nov 2026 13:46:46 +0800 Subject: [PATCH] feat(ingest): parse video refs and fetch detail by bvid --- reaction_board/bilibili/client.py | 48 +++++++++++++++ reaction_board/bilibili/models.py | 83 ++++++++++++++++++++++++++ reaction_board/video_ref.py | 82 ++++++++++++++++++++++++++ tests/fixtures.py | 6 ++ tests/test_bilibili_client.py | 98 +++++++++++++++++++++++++++++++ tests/test_video_ref.py | 93 +++++++++++++++++++++++++++++ 6 files changed, 410 insertions(+) create mode 100644 reaction_board/video_ref.py create mode 100644 tests/test_video_ref.py diff --git a/reaction_board/bilibili/client.py b/reaction_board/bilibili/client.py index c0b597c..162aebf 100644 --- a/reaction_board/bilibili/client.py +++ b/reaction_board/bilibili/client.py @@ -27,6 +27,7 @@ from .models import ( FolderInfo, InfosResult, LoginState, + VideoDetail, ViewStat, ) @@ -463,6 +464,53 @@ class BilibiliClient: 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) diff --git a/reaction_board/bilibili/models.py b/reaction_board/bilibili/models.py index 6530b01..691c1c3 100644 --- a/reaction_board/bilibili/models.py +++ b/reaction_board/bilibili/models.py @@ -146,6 +146,89 @@ class ViewStat: ) +@dataclass(frozen=True) +class VideoDetail: + """``x/web-interface/view`` 的完整元数据 + 七项统计。 + + 比 :class:`ViewStat` 多出 UP、时长、分 P 数、封面等字段，供手动入库建库使用。 + """ + + aid: int + bvid: str | None + title: str + up_mid: int | None + up_name: str | None + pubdate: int + duration: int + pages: int | None + cover: str | None + stat: ViewStat + + def metrics(self) -> dict[str, int]: + return self.stat.metrics() + + @classmethod + def from_api(cls, item: Mapping[str, Any]) -> "VideoDetail": + stat = ViewStat.from_api(item) + owner = item.get("owner") or {} + if not isinstance(owner, Mapping): + owner = {} + raw_mid = owner.get("mid") + up_mid = ( + raw_mid + if isinstance(raw_mid, int) and not isinstance(raw_mid, bool) + else None + ) + up_name = owner.get("name") + if not isinstance(up_name, str) or not up_name: + up_name = None + + raw_duration = item.get("duration") + duration = ( + raw_duration + if isinstance(raw_duration, int) + and not isinstance(raw_duration, bool) + and raw_duration >= 0 + else 0 + ) + + pages: int | None = None + raw_pages = item.get("pages") + if isinstance(raw_pages, list) and raw_pages: + pages = len(raw_pages) + else: + raw_videos = item.get("videos") + if ( + isinstance(raw_videos, int) + and not isinstance(raw_videos, bool) + and raw_videos > 0 + ): + pages = raw_videos + + raw_pubdate = item.get("pubdate") or item.get("ctime") + pubdate = ( + raw_pubdate + if isinstance(raw_pubdate, int) and not isinstance(raw_pubdate, bool) + else 0 + ) + cover = item.get("pic") + if not isinstance(cover, str) or not cover: + cover = None + + return cls( + aid=stat.aid, + bvid=stat.bvid, + title=stat.title, + up_mid=up_mid, + up_name=up_name, + pubdate=pubdate, + duration=duration, + pages=pages, + cover=cover, + stat=stat, + ) + + @dataclass(frozen=True) class FanStat: mid: int diff --git a/reaction_board/video_ref.py b/reaction_board/video_ref.py new file mode 100644 index 0000000..e972da8 --- /dev/null +++ b/reaction_board/video_ref.py @@ -0,0 +1,82 @@ +# -*- coding: utf-8 -*- +"""从任意文本中提取 B 站视频引用（纯函数，不做网络请求）。 + +手动入库支持的消息形态： + +* 视频 URL：``https://www.bilibili.com/video/BV1x8x166EzY``（含移动站、查询串、 + 分 P 参数等尾巴）； +* 裸 BV 号：``BV1x8x166EzY``； +* 裸 av 号：``av12345``（大小写不敏感）； +* 任意包含上述特征的字符串（例如复制了一整段带链接的文本）； +* ``b23.tv`` 短链：``https://b23.tv/xxxx``——只提取短链 token，真正的跳转解析 + 由 ``bilibili.client`` 的 HTTP 边界完成，保持本模块可离线测试。 +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +# BV 号固定为 "BV" + 10 位 base58 字符；后面不能紧跟字母数字，避免截断更长的串。 +_BVID_RE = re.compile(r"BV[0-9A-Za-z]{10}(?![0-9A-Za-z])") +# av 号：前面不能是字母数字，避免命中 "have123" 之类；大小写都接受。 +_AID_RE = re.compile(r"(? tuple[str, str | int]: + return (self.kind, self.value) + + +def _matches(text: str) -> list[tuple[int, VideoRef]]: + found: list[tuple[int, VideoRef]] = [] + for match in _BVID_RE.finditer(text): + found.append((match.start(), VideoRef("bvid", match.group(0)))) + for match in _AID_RE.finditer(text): + found.append((match.start(), VideoRef("aid", int(match.group(1))))) + for match in _SHORT_RE.finditer(text): + found.append((match.start(), VideoRef("short", match.group(1)))) + found.sort(key=lambda item: item[0]) + return found + + +def extract_video_refs( + text: object, *, limit: int | None = DEFAULT_MAX_REFS +) -> list[VideoRef]: + """从文本中按出现顺序提取去重后的视频引用。 + + * 非字符串 / 空串 → 返回空列表； + * 相同 ``(kind, value)`` 只保留首次出现； + * ``limit`` 为 ``None`` 时不限制数量。 + """ + if not isinstance(text, str) or not text: + return [] + out: list[VideoRef] = [] + seen: set[tuple[str, str | int]] = set() + for _, ref in _matches(text): + key = ref.key() + if key in seen: + continue + seen.add(key) + out.append(ref) + if limit is not None and len(out) >= max(0, int(limit)): + break + return out \ No newline at end of file diff --git a/tests/fixtures.py b/tests/fixtures.py index 12cd1b0..38d8085 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -176,6 +176,12 @@ VIEW_PAYLOAD: dict[str, Any] = { "pubdate": 1793370433, "ctime": 1793370433, "duration": 9504, + "pic": "http://i1.hdslb.com/bfs/archive/e5d3d4f25777b08c1dfae8bc9dc3d56b21d548f1.jpg", + "videos": 2, + "pages": [ + {"cid": 43184885058, "page": 1, "from": "vupload", "part": "第一部分"}, + {"cid": 43184885059, "page": 2, "from": "vupload", "part": "第二部分"}, + ], "owner": {"mid": 434442072, "name": "_-抱抱-_", "face": ""}, "stat": { "aid": 117530286889638, diff --git a/tests/test_bilibili_client.py b/tests/test_bilibili_client.py index 3b293a6..0ebd195 100644 --- a/tests/test_bilibili_client.py +++ b/tests/test_bilibili_client.py @@ -340,6 +340,100 @@ def test_view_unavailable_and_malformed(): raise AssertionError("缺少 stat 字段应报协议错误") +def test_view_detail_by_aid_and_bvid(): + transport = QueueTransport([http_json(VIEW_PAYLOAD)]) + client, _ = make_client(transport) + detail = client.view_detail(aid=117530286889638) + assert detail.aid == 117530286889638 + assert detail.bvid == "BV1xvxi6LEZD" + assert detail.up_mid == 434442072 + assert detail.up_name == "_-抱抱-_" + assert detail.pubdate == 1793370433 + assert detail.duration == 9504 + assert detail.pages == 2 + assert detail.cover and detail.cover.startswith("http") + assert detail.metrics()["view"] == 819 + assert "aid=117530286889638" in transport.calls[0]["url"] + + transport = QueueTransport([http_json(VIEW_PAYLOAD)]) + client, _ = make_client(transport) + detail = client.view_detail(bvid="BV1xvxi6LEZD") + assert detail.aid == 117530286889638 + assert "bvid=BV1xvxi6LEZD" in transport.calls[0]["url"] + + +def test_view_detail_fallbacks_and_required_arg(): + payload = view_for(1) + for key in ("owner", "duration", "pages", "videos", "pic"): + payload["data"].pop(key, None) + transport = QueueTransport([http_json(payload)]) + client, _ = make_client(transport) + detail = client.view_detail(aid=1) + assert detail.up_mid is None and detail.up_name is None + assert detail.duration == 0 + assert detail.pages is None + assert detail.cover is None + + client, _ = make_client(QueueTransport([])) + for kwargs in ({}, {"aid": 1, "bvid": "BV1xvxi6LEZD"}): + try: + client.view_detail(**kwargs) + except ValueError: + pass + else: + raise AssertionError("必须且只能提供 aid 或 bvid") + + +def test_view_detail_unavailable(): + transport = QueueTransport([ + http_json(VIEW_UNAVAILABLE), http_json(VIEW_UNAVAILABLE), + ]) + client, _ = make_client(transport) + try: + client.view_detail(bvid="BV1xvxi6LEZD") + except BilibiliAPIError as exc: + assert exc.code == 62012 + else: + raise AssertionError("不可访问视频应抛业务错误") + + +def test_resolve_short_link(): + redirect = HTTPResponse( + 302, + {"location": "https://www.bilibili.com/video/BV1x8x166EzY"}, + b"", + ) + transport = QueueTransport([redirect]) + client, _ = make_client(transport) + final = client.resolve_short_link("https://b23.tv/abc") + assert final == "https://www.bilibili.com/video/BV1x8x166EzY" + assert transport.calls[0]["method"] == "GET" + assert "b23.tv" in transport.calls[0]["url"] + + # 多跳重定向，大小写不同的 Location 头 + transport = QueueTransport([ + HTTPResponse(301, {"Location": "https://b23.tv/next"}, b""), + HTTPResponse(302, {"location": "https://www.bilibili.com/video/BV1x8x166EzY"}, b""), + ]) + client, _ = make_client(transport) + assert client.resolve_short_link("https://b23.tv/start").endswith("BV1x8x166EzY") + + # 非重定向直接返回最终 URL + transport = QueueTransport([HTTPResponse(200, {}, b"")]) + client, _ = make_client(transport) + assert client.resolve_short_link("https://b23.tv/x") == "https://b23.tv/x" + + # HTTP 错误应抛出 + transport = QueueTransport([HTTPResponse(404, {}, b"not found")]) + client, _ = make_client(transport) + try: + client.resolve_short_link("https://b23.tv/x") + except BilibiliHTTPError as exc: + assert exc.status == 404 + else: + raise AssertionError("短链 HTTP 404 应抛出") + + def test_relation_stat(): transport = QueueTransport([http_json(RELATION_PAYLOAD)]) client, _ = make_client(transport) @@ -436,6 +530,10 @@ def main(): test_timeout_and_method_forwarded() test_view_seven_metrics() test_view_unavailable_and_malformed() + test_view_detail_by_aid_and_bvid() + test_view_detail_fallbacks_and_required_arg() + test_view_detail_unavailable() + test_resolve_short_link() test_relation_stat() test_check_login() test_check_login_without_cookie_makes_no_request() diff --git a/tests/test_video_ref.py b/tests/test_video_ref.py new file mode 100644 index 0000000..5db7cf2 --- /dev/null +++ b/tests/test_video_ref.py @@ -0,0 +1,93 @@ +# -*- coding: utf-8 -*- +"""视频引用解析纯函数测试: uv run python tests/test_video_ref.py""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.stdout.reconfigure(encoding="utf-8") + +from reaction_board.video_ref import VideoRef, extract_video_refs # noqa: E402 + +BV = "BV1x8x166EzY" + + +def test_bv_url_and_bare_bvid(): + assert extract_video_refs(f"https://www.bilibili.com/video/{BV}") == [ + VideoRef("bvid", BV) + ] + assert extract_video_refs(BV) == [VideoRef("bvid", BV)] + assert extract_video_refs( + f"https://m.bilibili.com/video/{BV}?p=2&spm_id_from=333.999" + ) == [VideoRef("bvid", BV)] + assert extract_video_refs(f"https://www.bilibili.com/video/{BV}/?share=1") == [ + VideoRef("bvid", BV) + ] + + +def test_arbitrary_text_containing_bvid(): + assert extract_video_refs(f"看看这个 {BV} 很有意思") == [VideoRef("bvid", BV)] + assert extract_video_refs(f"前缀{BV}后缀") == [VideoRef("bvid", BV)] + + +def test_av_id_forms(): + assert extract_video_refs("av12345") == [VideoRef("aid", 12345)] + assert extract_video_refs("AV12345") == [VideoRef("aid", 12345)] + assert extract_video_refs("https://www.bilibili.com/video/av12345") == [ + VideoRef("aid", 12345) + ] + + +def test_short_link(): + assert extract_video_refs("https://b23.tv/abcDEF12") == [ + VideoRef("short", "abcDEF12") + ] + assert extract_video_refs("b23.tv/xyz") == [VideoRef("short", "xyz")] + + +def test_multiple_and_order_and_dedup(): + text = f"{BV} 还有 av100 再看 {BV} 以及 https://b23.tv/short1" + assert extract_video_refs(text) == [ + VideoRef("bvid", BV), + VideoRef("aid", 100), + VideoRef("short", "short1"), + ] + + +def test_limit(): + text = " ".join(["av1", "av2", "av3"]) + assert extract_video_refs(text, limit=2) == [ + VideoRef("aid", 1), + VideoRef("aid", 2), + ] + assert len(extract_video_refs(text, limit=None)) == 3 + + +def test_invalid_and_boundaries(): + assert extract_video_refs("") == [] + assert extract_video_refs(None) == [] + assert extract_video_refs(12345) == [] + assert extract_video_refs("hello world") == [] + # BV 号长度不足 + assert extract_video_refs("BV1x8") == [] + # 更长的字母数字串不应被截断成假 BV 号 + assert extract_video_refs(f"{BV}EXTRA") == [] + # "have123" 不应命中 av + assert extract_video_refs("have123") == [] + # av 后紧跟字母不应命中 + assert extract_video_refs("av12x") == [] + + +def main(): + test_bv_url_and_bare_bvid() + test_arbitrary_text_containing_bvid() + test_av_id_forms() + test_short_link() + test_multiple_and_order_and_dedup() + test_limit() + test_invalid_and_boundaries() + print("OK: 视频引用（BV/av/URL/短链）解析测试全部通过") + + +if __name__ == "__main__": + main() \ No newline at end of file