Files
Agent-Reach/agent_reach/channels/bilibili.py
T
Pnant 762824c590 feat(routing): ordered backend candidates + real-probing doctor
- backends is now an ordered candidate list (first = preferred); channels
  report the backend actually serving via active_backend, surfaced in the
  doctor text report and --json
- new agent_reach/probe.py really executes upstream commands and tells
  apart missing / broken (stale venv shebang after a system Python
  upgrade) / timeout, with a reinstall prescription for broken installs
- all 13 channels migrated off which()-only checks: fixes bilibili
  false-positive "bili-cli 可用" on broken shims, misleading xiaohongshu
  "连接失败", rdt OSError crashing doctor, mcporter breakage masquerading
  as "未配置"
- twitter: 15s probe + 1 retry (flaky 10s timeout), broken twitter-cli
  now falls back to bird instead of aborting the check
- doctor survives per-channel exceptions; config supports per-channel
  backend override (<channel>_backend / <CHANNEL>_BACKEND env)
- fix skill install/uninstall crash on symlinked skill dirs (the
  "[Errno None] None" warning from shutil.rmtree on a symlink)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-06-11 15:48:33 +08:00

86 lines
3.2 KiB
Python

# -*- coding: utf-8 -*-
"""Bilibili — video via yt-dlp, search/browse via bili-cli or API."""
import json
import os
import urllib.request
from agent_reach.probe import probe_command
from .base import Channel
_UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
_TIMEOUT = 10
_SEARCH_API = "https://api.bilibili.com/x/web-interface/search/all/v2?keyword=test&page=1"
def _search_api_ok() -> bool:
"""Return True if Bilibili search API responds with code 0."""
req = urllib.request.Request(_SEARCH_API, headers={"User-Agent": _UA})
try:
with urllib.request.urlopen(req, timeout=_TIMEOUT) as resp:
data = json.loads(resp.read())
return data.get("code") == 0
except Exception:
return False
class BilibiliChannel(Channel):
name = "bilibili"
description = "B站视频、字幕和搜索"
backends = ["yt-dlp", "bili-cli (可选)", "B站搜索 API"]
tier = 1
def can_handle(self, url: str) -> bool:
from urllib.parse import urlparse
d = urlparse(url).netloc.lower()
return "bilibili.com" in d or "b23.tv" in d
def check(self, config=None):
self.active_backend = None
# 真跑 yt-dlp --version,区分 未装 / 断链 / 异常(which 命中不等于能用)
yt = probe_command("yt-dlp", ["--version"], timeout=10, package="yt-dlp")
if yt.status == "missing":
return "off", "yt-dlp 未安装。安装:pip install yt-dlp"
if yt.status == "broken":
return "error", "yt-dlp 已安装但无法执行\n" + yt.hint
if not yt.ok:
detail = yt.hint or yt.output or yt.status
return "error", f"yt-dlp 探测失败({yt.status}):{detail}"
self.active_backend = "yt-dlp"
proxy = (config.get("bilibili_proxy") if config else None) or os.environ.get("BILIBILI_PROXY")
# 真跑 bili --version——断链时旧的 which 检测会误报"bili-cli 可用"
bili = probe_command("bili", ["--version"], timeout=10, package="bilibili-cli")
parts = []
# 视频读取状态
if proxy:
parts.append("视频读取:yt-dlp(代理已配置)")
else:
parts.append("视频读取:yt-dlp")
# bili-cli 增强
if bili.ok:
parts.append("搜索/热门/排行:bili-cli 可用")
status = "ok"
else:
if bili.status == "broken":
parts.append("bili-cli 已安装但无法执行,不计为可用\n" + bili.hint)
elif bili.status in ("timeout", "error"):
parts.append(f"bili-cli 探测失败({bili.status}),不计为可用")
# 降级走搜索 API;只探测一次,message 和 status 共用结果
api_ok = _search_api_ok()
if api_ok:
parts.append("搜索:B站 API 可用")
else:
parts.append("搜索:B站 API 不可达")
if bili.status == "missing":
parts.append("提示:安装 bili-cli 可解锁热门/排行/动态:pipx install bilibili-cli")
status = "ok" if api_ok else "warn"
return status, "".join(parts)