Panniantong
5c62a21f32
全局重命名： - 包名: agent_eyes → agent_reach - CLI: agent-eyes → agent-reach - 类名: AgentEyes → AgentReach - 显示名: Agent Eyes → Agent Reach - GitHub: Panniantong/agent-eyes → Panniantong/Agent-Reach 所有 36 个测试通过，CLI/doctor/read/search 全部正常。
122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Bilibili — via public API (free, no config needed).
|
|
|
|
Backend: Bilibili public API
|
|
Swap to: any Bilibili access method
|
|
"""
|
|
|
|
import requests
|
|
from urllib.parse import urlparse, parse_qs
|
|
from .base import Channel, ReadResult
|
|
|
|
|
|
class BilibiliChannel(Channel):
|
|
name = "bilibili"
|
|
description = "B站视频信息和字幕"
|
|
backends = ["Bilibili API"]
|
|
tier = 0
|
|
|
|
def can_handle(self, url: str) -> bool:
|
|
domain = urlparse(url).netloc.lower()
|
|
return "bilibili.com" in domain or "b23.tv" in domain
|
|
|
|
def check(self, config=None):
|
|
proxy = config.get("bilibili_proxy") if config else None
|
|
if proxy:
|
|
return "ok", "已配置代理，完整可用"
|
|
# Detect if we're on a server (same logic as cli._detect_environment)
|
|
import os
|
|
indicators = [
|
|
os.path.exists("/var/run/docker.sock"),
|
|
os.path.exists("/etc/cloud"),
|
|
"SSH_CONNECTION" in os.environ,
|
|
"container" in os.environ.get("container", ""),
|
|
]
|
|
is_server = any(indicators)
|
|
if is_server:
|
|
return "warn", "服务器 IP 可能被封，配置代理即可解决：agent-reach configure proxy URL"
|
|
return "ok", "本地直连可用"
|
|
|
|
async def read(self, url: str, config=None) -> ReadResult:
|
|
# Proxy support (Bilibili blocks server IPs)
|
|
proxy = config.get("bilibili_proxy") if config else None
|
|
proxies = {"http": proxy, "https": proxy} if proxy else None
|
|
|
|
# Extract BV id from URL
|
|
path = urlparse(url).path
|
|
bv_id = ""
|
|
for part in path.split("/"):
|
|
if part.startswith("BV"):
|
|
bv_id = part
|
|
break
|
|
|
|
if not bv_id:
|
|
# Fallback to Jina Reader
|
|
from agent_reach.channels.web import WebChannel
|
|
return await WebChannel().read(url, config)
|
|
|
|
# Get video info
|
|
resp = requests.get(
|
|
"https://api.bilibili.com/x/web-interface/view",
|
|
params={"bvid": bv_id},
|
|
headers={"User-Agent": "Mozilla/5.0"},
|
|
proxies=proxies,
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
api_data = resp.json()
|
|
|
|
# Check for API errors (IP blocked, video not found, etc.)
|
|
if api_data.get("code") != 0:
|
|
msg = api_data.get("message", "Unknown error")
|
|
# Bilibili returns -404 when server IP is blocked
|
|
if api_data.get("code") in (-404, -403, -412):
|
|
return ReadResult(
|
|
title=f"Bilibili: {bv_id}",
|
|
content=f"⚠️ Bilibili blocked this request ({msg}). "
|
|
f"This usually means the server IP is blocked. "
|
|
f"Try: agent-reach configure proxy http://user:pass@ip:port",
|
|
url=url,
|
|
platform="bilibili",
|
|
)
|
|
return ReadResult(
|
|
title=f"Bilibili: {bv_id}",
|
|
content=f"Bilibili API error: {msg} (code: {api_data.get('code')})",
|
|
url=url,
|
|
platform="bilibili",
|
|
)
|
|
|
|
data = api_data.get("data", {})
|
|
|
|
title = data.get("title", "")
|
|
desc = data.get("desc", "")
|
|
author = data.get("owner", {}).get("name", "")
|
|
|
|
# Try to get subtitles
|
|
subtitle_text = ""
|
|
subtitle_list = data.get("subtitle", {}).get("list", [])
|
|
if subtitle_list:
|
|
sub_url = subtitle_list[0].get("subtitle_url", "")
|
|
if sub_url:
|
|
if sub_url.startswith("//"):
|
|
sub_url = "https:" + sub_url
|
|
sr = requests.get(sub_url, timeout=10)
|
|
if sr.ok:
|
|
sub_data = sr.json()
|
|
lines = [item.get("content", "") for item in sub_data.get("body", [])]
|
|
subtitle_text = "\n".join(lines)
|
|
|
|
content = desc
|
|
if subtitle_text:
|
|
content += f"\n\n## Transcript\n{subtitle_text}"
|
|
|
|
return ReadResult(
|
|
title=title,
|
|
content=content,
|
|
url=url,
|
|
author=author,
|
|
platform="bilibili",
|
|
extra={"view": data.get("stat", {}).get("view", 0),
|
|
"like": data.get("stat", {}).get("like", 0)},
|
|
)
|