feat(ingest): parse video refs and fetch detail by bvid

This commit is contained in:
kairee committed 2026-11-01 13:46:46 +08:00
1 parent e903c49921
commit d2adb65e08
6 files changed
+410

No files matched your search

+82
View File
@@ -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"(?<![0-9A-Za-z])av(\d{1,20})(?![0-9A-Za-z])", re.IGNORECASE)
# b23.tv 短链 token。
_SHORT_RE = re.compile(r"b23\.tv/([0-9A-Za-z]+)")
# 单条消息最多处理的引用数,避免误粘贴一段文本导致突发请求。
DEFAULT_MAX_REFS = 10
@dataclass(frozen=True)
class VideoRef:
"""一条待解析的视频引用。
``kind`` 取值:
* ``bvid``:``value`` 为 BV 号字符串;
* ``aid``:``value`` 为 av 号对应的整数 aid;
* ``short``:``value`` 为 b23.tv 短链 token,需要 HTTP 跳转解析。
"""
kind: str
value: str | int
def key(self) -> 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