哔哩哔哩
第三方 via ClawHub全面的Bilibili工具包，集成了热门趋势监测、视频下载、视频观看/播放、字幕下载和视频发布等功能……
wscats v1.0.12
<p align="center">
<h1 align="center">🎬 Bilibili All-in-One</h1>
<p align="center">一站式 B站工具箱 — 热门监控 · 视频下载 · 数据追踪 · 字幕提取 · 视频播放 · 投稿发布</p>
</p>
<p align="center">
<img src="https://img.shields.io/badge/python-%3E%3D3.8-blue?logo=python&logoColor=white" />
<img src="https://img.shields.io/badge/license-MIT-green" />
<img src="https://img.shields.io/badge/version-1.0.9-orange" />
<img src="https://img.shields.io/badge/platform-Bilibili%20%7C%20YouTube-pink" />
</p>
---
## 📖 简介
**Bilibili All-in-One** 是一个综合性的 B站工具包，将 6 个独立的 B站技能整合为一个统一的 Skill，提供从热门监控到视频投稿的全链路能力。
支持作为 **AI Agent 技能**、**命令行工具** 或 **Python 库** 使用。
## ✨ 功能总览
| 模块 | 功能 | 是否需要登录 |
|:---:|---|:---:|
| 🔥 **热门监控** | 热门视频、热搜话题、每周必看、分区排行榜 | ❌ |
| ⬇️ **视频下载** | 多清晰度下载、批量下载、格式转换、音频提取 | ⚠️ 高清需要 |
| 👀 **数据追踪** | 播放/点赞/收藏统计、数据追踪、多视频对比（支持 YouTube） | ❌ |
| 📝 **字幕提取** | 字幕下载、格式转换（SRT/ASS/VTT/TXT）、多语言、字幕合并 | ❌ |
| ▶️ **视频播放** | 播放地址获取、弹幕抓取、分P/播放列表信息 | ⚠️ 高清需要 |
| 📤 **视频发布** | 上传投稿、定时发布、草稿管理、编辑/删除视频 | ✅ 必须 |
## 🚀 快速开始
### 环境要求
- **Python** >= 3.8
- **ffmpeg**（可选，用于合并视频/音频流）
### 安装依赖
```bash
git clone https://github.com/wscats/bilibili-all-in-one.git
cd bilibili-all-in-one
pip install -r requirements.txt
```
### 30 秒上手
```python
import asyncio
from main import BilibiliAllInOne
app = BilibiliAllInOne()
async def main():
# 获取 B站热门视频
hot = await app.execute("hot_monitor", "get_hot", page_size=5)
print(hot)
asyncio.run(main())
```
## ⚙️ 配置认证
部分功能（高清下载、视频发布等）需要 B站登录凭据。支持三种配置方式：
### 方式一：环境变量（推荐）
```bash
export BILIBILI_SESSDATA="你的_sessdata"
export BILIBILI_BILI_JCT="你的_bili_jct"
export BILIBILI_BUVID3="你的_buvid3"
```
### 方式二：凭据文件
创建 `credentials.json`：
```json
{
"sessdata": "你的_sessdata",
"bili_jct": "你的_bili_jct",
"buvid3": "你的_buvid3"
}
```
### 方式三：代码直接传入
```python
app = BilibiliAllInOne(
sessdata="你的_sessdata",
bili_jct="你的_bili_jct",
buvid3="你的_buvid3",
)
```
> 💡 **如何获取 Cookie？** 登录 [bilibili.com](https://www.bilibili.com) → 按 F12 打开开发者工具 → Application → Cookies → 复制 `SESSDATA`、`bili_jct`、`buvid3` 的值。
---
## 📚 使用方式
### 命令行（CLI）
```bash
python main.py <模块名> <操作> [参数JSON]
```
### Python API
```python
import asyncio
from main import BilibiliAllInOne
app = BilibiliAllInOne()
result = asyncio.run(app.execute("模块名", "操作", 参数=值))
```
---
## 🔥 模块详解
### 1. 热门监控 (`hot_monitor`)
实时监控 B站热门视频与话题趋势。
| 操作 | 说明 | 参数 |
|---|---|---|
| `get_hot` | 获取热门视频列表 | `page`, `page_size` |
| `get_trending` | 获取热搜话题 | `limit` |
| `get_weekly` | 获取每周必看榜 | `number`（期数，可选） |
| `get_rank` | 获取分区排行榜 | `category`, `limit` |
**支持的分区：** `all`、`anime`、`music`、`dance`、`game`、`tech`、`life`、`food`、`car`、`fashion`、`entertainment`、`movie`、`tv`
```bash
# 获取前10个热门视频
python main.py hot_monitor get_hot '{"page_size": 10}'
# 获取游戏区排行榜
python main.py hot_monitor get_rank '{"category": "game", "limit": 10}'
# 获取本周必看
python main.py hot_monitor get_weekly
# 获取热搜话题
python main.py hot_monitor get_trending '{"limit": 5}'
```
```python
# Python API
result = await app.execute("hot_monitor", "get_hot", page_size=10)
result = await app.execute("hot_monitor", "get_rank", category="game", limit=10)
result = await app.execute("hot_monitor", "get_weekly")
result = await app.execute("hot_monitor", "get_trending", limit=5)
```
---
### 2. 视频下载 (`downloader`)
支持多清晰度、多格式下载，可批量操作。
| 操作 | 说明 | 参数 |
|---|---|---|
| `get_info` | 获取视频信息 | `url` |
| `get_formats` | 列出可用画质/格式 | `url` |
| `download` | 下载单个视频 | `url`, `quality`, `output_dir`, `format`, `page` |
| `batch_download` | 批量下载多个视频 | `urls`, `quality`, `output_dir`, `format` |
**清晰度选项：** `360p` · `480p` · `720p` · `1080p`（默认）· `1080p+` · `4k`
**格式选项：** `mp4`（默认）· `flv` · `mp3`（仅音频）
```bash
# 获取视频信息
python main.py downloader get_info '{"url": "BV1xx411c7mD"}'
# 下载 1080p MP4
python main.py downloader download '{"url": "BV1xx411c7mD", "quality": "1080p"}'
# 提取音频
python main.py downloader download '{"url": "BV1xx411c7mD", "format": "mp3"}'
# 批量下载
python main.py downloader batch_download '{"urls": ["BV1xx411c7mD", "BV1yy411c8nE"], "quality": "720p"}'
```
```python
# Python API
info = await app.execute("downloader", "get_info", url="BV1xx411c7mD")
result = await app.execute("downloader", "download", url="BV1xx411c7mD", quality="1080p")
result = await app.execute("downloader", "batch_download", urls=["BV1xx411c7mD", "BV1yy411c8nE"])
```
---
### 3. 数据追踪 (`watcher`)
追踪 B站和 YouTube 视频的互动数据，支持多视频对比。
| 操作 | 说明 | 参数 |
|---|---|---|
| `watch` | 获取视频详细信息 | `url` |
| `get_stats` | 获取当前互动数据 | `url` |
| `track` | 持续追踪数据变化 | `url`, `interval`（分钟）, `duration`（小时） |
| `compare` | 对比多个视频数据 | `urls` |
**支持平台：**
- **B站**：`https://www.bilibili.com/video/BVxxxxxx` 或 `BVxxxxxx`
- **YouTube**：`https://www.youtube.com/watch?v=xxxxx` 或 `https://youtu.be/xxxxx`
```bash
# 查看视频详情
python main.py watcher watch '{"url": "BV1xx411c7mD"}'
# 获取互动数据
python main.py watcher get_stats '{"url": "BV1xx411c7mD"}'
# 每30分钟追踪一次，持续12小时
python main.py watcher track '{"url": "BV1xx411c7mD", "interval": 30, "duration": 12}'
# 对比多个视频
python main.py watcher compare '{"urls": ["BV1xx411c7mD", "BV1yy411c8nE"]}'
# YouTube 视频也支持
python main.py watcher watch '{"url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ"}'
```
```python
# Python API
stats = await app.execute("watcher", "get_stats", url="BV1xx411c7mD")
comparison = await app.execute("watcher", "compare", urls=["BV1xx411c7mD", "BV1yy411c8nE"])
youtube_info = await app.execute("watcher", "watch", url="https://youtu.be/dQw4w9WgXcQ")
```
---
### 4. 字幕提取 (`subtitle`)
下载和处理 B站视频字幕，支持多语言和多格式。
| 操作 | 说明 | 参数 |
|---|---|---|
| `list` | 列出可用字幕 | `url` |
| `download` | 下载字幕 | `url`, `language`, `format`, `output_dir` |
| `convert` | 转换字幕格式 | `input_path`, `output_format`, `output_dir` |
| `merge` | 合并多个字幕文件 | `input_paths`, `output_path`, `output_format` |
**支持格式：** `srt`（默认）· `ass` · `vtt` · `txt` · `json`
**支持语言：** `zh-CN`（默认）· `en` · `ja` 以及视频提供的其他语言
```bash
# 列出可用字幕
python main.py subtitle list '{"url": "BV1xx411c7mD"}'
# 下载中文字幕（SRT格式）
python main.py subtitle download '{"url": "BV1xx411c7mD", "language": "zh-CN", "format": "srt"}'
# 下载英文字幕（ASS格式）
python main.py subtitle download '{"url": "BV1xx411c7mD", "language": "en", "format": "ass"}'
# 格式转换：SRT → VTT
python main.py subtitle convert '{"input_path": "./subtitles/video.srt", "output_format": "vtt"}'
# 合并多个字幕
python main.py subtitle merge '{"input_paths": ["part1.srt", "part2.srt"], "output_path": "merged.srt"}'
```
```python
# Python API
subs = await app.execute("subtitle", "list", url="BV1xx411c7mD")
result = await app.execute("subtitle", "download", url="BV1xx411c7mD", language="zh-CN", format="srt")
result = await app.execute("subtitle", "convert", input_path="video.srt", output_format="vtt")
```
---
### 5. 视频播放 (`player`)
获取播放地址、弹幕数据和播放列表信息。
| 操作 | 说明 | 参数 |
|---|---|---|
| `play` | 获取完整播放信息 | `url`, `quality`, `page` |
| `get_playurl` | 获取直接播放地址 | `url`, `quality`, `page` |
| `get_danmaku` | 获取弹幕数据 | `url`, `page`, `segment` |
| `get_playlist` | 获取分P/播放列表信息 | `url` |
**弹幕类型：**
| 模式 | 说明 |
|:---:|---|
| 1 | 滚动弹幕（从右到左） |
| 4 | 底部固定弹幕 |
| 5 | 顶部固定弹幕 |
```bash
# 获取播放信息
python main.py player play '{"url": "BV1xx411c7mD", "quality": "1080p"}'
# 获取播放地址
python main.py player get_playurl '{"url": "BV1xx411c7mD", "quality": "720p"}'
# 获取弹幕
python main.py player get_danmaku '{"url": "BV1xx411c7mD"}'
# 获取分P列表
python main.py player get_playlist '{"url": "BV1xx411c7mD"}'
# 播放多P视频的第3P
python main.py player play '{"url": "BV1xx411c7mD", "quality": "1080p", "page": 3}'
```
```python
# Python API
play_info = await app.execute("player", "play", url="BV1xx411c7mD", quality="1080p")
danmaku = await app.execute("player", "get_danmaku", url="BV1xx411c7mD")
playlist = await app.execute("player", "get_playlist", url="BV1xx411c7mD")
```
---
### 6. 视频发布 (`publisher`)
上传视频到 B站，支持定时发布和草稿管理。
> ⚠️ **此模块所有操作均需要登录认证**
| 操作 | 说明 | 参数 |
|---|---|---|
| `upload` | 上传并发布视频 | `file_path`, `title`, `description`, `tags`, `category`, `cover_path` |
| `draft` | 保存为草稿 | `file_path`, `title`, `description`, `tags`, `category` |
| `schedule` | 定时发布 | `file_path`, `title`, `schedule_time`, `description`, `tags` |
| `edit` | 编辑已发布视频 | `bvid`, `title`, `description`, `tags`, `cover_path` |
| `delete` | 删除视频 | `bvid` |
**上传参数说明：**
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
| `file_path` | string | *必填* | 视频文件路径 |
| `title` | string | *必填* | 视频标题（最长 80 字） |
| `description` | string | `""` | 视频简介（最长 2000 字） |
| `tags` | string[] | `["bilibili"]` | 标签（最多 12 个，每个最长 20 字） |
| `category` | string | `"171"` | 分区 TID |
| `cover_path` | string | `null` | 封面图片路径（JPG/PNG） |
| `no_reprint` | int | `1` | 1=自制，0=转载 |
| `open_elec` | int | `0` | 1=开启充电，0=关闭 |
```bash
# 上传并发布
python main.py publisher upload '{"file_path": "./video.mp4", "title": "我的视频", "description": "Hello World", "tags": ["测试", "演示"]}'
# 保存为草稿
python main.py publisher draft '{"file_path": "./video.mp4", "title": "草稿视频"}'
# 定时发布
python main.py publisher schedule '{"file_path": "./video.mp4", "title": "定时视频", "schedule_time": "2025-12-31T20:00:00+08:00"}'
# 编辑视频信息
python main.py publisher edit '{"bvid": "BV1xx411c7mD", "title": "新标题", "tags": ["更新"]}'
# 删除视频
python main.py publisher delete '{"bvid": "BV1xx411c7mD"}'
```
```python
# Python API（需要认证）
app = BilibiliAllInOne(sessdata="xxx", bili_jct="xxx", buvid3="xxx")
result = await app.execute("publisher", "upload",
file_path="./video.mp4",
title="我的视频",
description="通过 bilibili-all-in-one 发布",
tags=["python", "bilibili"],
)
```
---
## 🔒 安全说明
### 凭据处理
| 关注点 | 说明 |
|---|---|
| **需要哪些凭据？** | `SESSDATA`、`bili_jct`、`buvid3` — B站浏览器 Cookie |
| **哪些功能需要认证？** | 视频发布（上传/编辑/删除/定时/草稿）、1080p+/4K 下载 |
| **哪些功能无需认证？** | 热门监控、标准画质下载、字幕获取、弹幕抓取、数据查看 |
| **凭据发送到哪里？** | **仅限** B站官方 API（`api.bilibili.com`、`member.bilibili.com`），全部 HTTPS |
| **是否持久化到磁盘？** | **否** — 除非你主动调用 `auth.save_to_file()`，凭据默认仅存在于内存 |
| **保存文件权限** | `0600`（仅所有者可读写） |
### 网络端点
| 域名 | 用途 |
|---|---|
| `api.bilibili.com` | 视频信息、统计、热门、字幕、弹幕、播放地址 |
| `member.bilibili.com` | 视频发布（上传、编辑、删除） |
| `upos-sz-upcdnbda2.bilivideo.com` | 视频文件上传 CDN |
| `www.bilibili.com` | 网页数据抓取备用 |
| `www.youtube.com` | YouTube 视频元数据获取（oEmbed API，无需认证） |
### 安全建议
1. 🧪 **使用测试账号** — 请勿使用主账号 Cookie 进行测试
2. 🔒 **优先使用内存凭据** — 通过环境变量或代码参数传入，避免保存到文件
3. 📁 **如需保存凭据** — 使用 `auth.save_to_file()`（自动设置 0600 权限），用完后及时删除
4. 🐳 **隔离运行** — 建议在容器/虚拟环境中运行，并监控网络流量
5. 🌐 **所有请求仅发往 B站官方域名**，无第三方遥测或数据收集
6. ❌ **本工具不会将你的凭据发送至任何非 B站的第三方服务**
---
## 📁 项目结构
```
bilibili-all-in-one/
├── skill.json # Skill 配置与参数 Schema
├── skill.md # Skill 英文文档
├── README.md # 中文说明文档（本文件）
├── requirements.txt # Python 依赖
├── main.py # 入口文件，统一的 BilibiliAllInOne 类
├── src/
│ ├── __init__.py # 包导出
│ ├── auth.py # 认证与凭据管理
│ ├── utils.py # 共享工具函数、API 常量
│ ├── hot_monitor.py # 🔥 热门监控模块
│ ├── downloader.py # ⬇️ 视频下载模块
│ ├── watcher.py # 👀 数据追踪模块
│ ├── subtitle.py # 📝 字幕提取模块
│ ├── player.py # ▶️ 视频播放模块
│ └── publisher.py # 📤 视频发布模块
└── tests/
├── __init__.py
└── test_all_skill_examples.py # 27 个完整测试用例
```
## 🧬 技能来源
本项目整合了以下 6 个独立 Skill 的功能：
| 原始 Skill | 来源 | 整合为 |
|---|---|---|
| bilibili-hot-monitor | [Jacobzwj/bilibili-hot-monitor](https://clawhub.ai/Jacobzwj/bilibili-hot-monitor) | `hot_monitor` |
| bililidownloader | [caiyundc880518/bililidownloader](https://clawhub.ai/caiyundc880518/bililidownloader) | `downloader` |
| bilibili-youtube-watcher | [donnycui/bilibili-youtube-watcher](https://clawhub.ai/donnycui/bilibili-youtube-watcher) | `watcher` |
| bilibili-subtitle-download-skill | [DavinciEvans/bilibili-subtitle-download-skill](https://clawhub.ai/DavinciEvans/bilibili-subtitle-download-skill) | `subtitle` |
| bilibili-player | [e421083458/bilibili-player](https://clawhub.ai/e421083458/bilibili-player) | `player` |
| bilibili-video-publish | [Johnnyxu820/bilibili-video-publish](https://clawhub.ai/Johnnyxu820/bilibili-video-publish) | `publisher` |
## 📦 统一返回格式
所有操作返回统一的 JSON 结构：
**成功：**
```json
{
"success": true,
"...": "操作相关的数据字段"
}
```
**失败：**
```json
{
"success": false,
"message": "错误描述信息"
}
```
## 🧪 运行测试
```bash
# 运行全部 27 个测试
python -m unittest tests.test_all_skill_examples -v
```
## 📄 许可证
[MIT](LICENSE)
bilibili-all-in-one
├── skill.json # Skill configuration
├── requirements.txt # Python dependencies
├── main.py # Entry point
├── src/
│ ├── __init__.py
│ ├── auth.py # Authentication & credential management
│ ├── hot_monitor.py # Hot/trending video monitoring
│ ├── downloader.py # Video downloading
│ ├── watcher.py # Video watching & stats tracking
│ ├── subtitle.py # Subtitle downloading & processing
│ ├── player.py # Video playback & danmaku
│ ├── publisher.py # Video uploading & publishing
│ └── utils.py # Shared utilities
└── tests/
└── __init__.py
"""Bilibili All-in-One Skill - Main Entry Point.
A comprehensive Bilibili toolkit that integrates:
- Hot/trending video monitoring
- Video downloading
- Video watching & stats tracking
- Subtitle downloading & processing
- Video playback & danmaku
- Video uploading & publishing
"""
import asyncio
import json
import sys
from typing import Dict, Any, Optional
from src.auth import BilibiliAuth
from src.hot_monitor import HotMonitor
from src.downloader import BilibiliDownloader
from src.watcher import BilibiliWatcher
from src.subtitle import SubtitleDownloader
from src.player import BilibiliPlayer
from src.publisher import BilibiliPublisher
class BilibiliAllInOne:
"""Unified interface for all Bilibili skill capabilities."""
def __init__(
self,
sessdata: Optional[str] = None,
bili_jct: Optional[str] = None,
buvid3: Optional[str] = None,
credential_file: Optional[str] = None,
):
"""Initialize BilibiliAllInOne.
Args:
sessdata: Bilibili SESSDATA cookie.
bili_jct: Bilibili bili_jct (CSRF) cookie.
buvid3: Bilibili buvid3 cookie.
credential_file: Path to JSON credential file.
"""
self.auth = BilibiliAuth(
sessdata=sessdata,
bili_jct=bili_jct,
buvid3=buvid3,
credential_file=credential_file,
)
# Initialize all modules
self.hot_monitor = HotMonitor(auth=self.auth)
self.downloader = BilibiliDownloader(auth=self.auth)
self.watcher = BilibiliWatcher(auth=self.auth)
self.subtitle = SubtitleDownloader(auth=self.auth)
self.player = BilibiliPlayer(auth=self.auth)
self._publisher = None # Lazy init (requires auth)
@property
def publisher(self) -> BilibiliPublisher:
"""Get the publisher module (requires authentication).
Returns:
BilibiliPublisher instance.
Raises:
ValueError: If not authenticated.
"""
if self._publisher is None:
self._publisher = BilibiliPublisher(auth=self.auth)
return self._publisher
async def execute(self, skill_name: str, action: str, **kwargs) -> Dict[str, Any]:
"""Execute any skill action through a unified interface.
Args:
skill_name: Name of the skill module.
action: Action to perform.
**kwargs: Additional parameters.
Returns:
Action result dict.
"""
skill_map = {
"bilibili_hot_monitor": lambda: self.hot_monitor,
"hot_monitor": lambda: self.hot_monitor,
"hot": lambda: self.hot_monitor,
"bilibili_downloader": lambda: self.downloader,
"downloader": lambda: self.downloader,
"download": lambda: self.downloader,
"bilibili_watcher": lambda: self.watcher,
"watcher": lambda: self.watcher,
"watch": lambda: self.watcher,
"bilibili_subtitle": lambda: self.subtitle,
"subtitle": lambda: self.subtitle,
"bilibili_player": lambda: self.player,
"player": lambda: self.player,
"play": lambda: self.player,
"bilibili_publisher": lambda: self.publisher,
"publisher": lambda: self.publisher,
"publish": lambda: self.publisher,
}
skill_factory = skill_map.get(skill_name)
if not skill_factory:
return {
"success": False,
"message": f"Unknown skill: {skill_name}. Available: {list(skill_map.keys())}",
}
skill = skill_factory()
return await skill.execute(action=action, **kwargs)
async def main():
"""CLI entry point for testing."""
if len(sys.argv) < 3:
print("Usage: python main.py <skill_name> <action> [params_json]")
print()
print("Skills:")
print(" hot_monitor - Monitor hot/trending videos")
print(" downloader - Download videos")
print(" watcher - Watch and track video stats")
print(" subtitle - Download subtitles")
print(" player - Play videos and get danmaku")
print(" publisher - Upload and publish videos")
print()
print("Examples:")
print(' python main.py hot_monitor get_hot \'{"limit": 5}\'')
print(' python main.py downloader get_info \'{"url": "BV1xx411c7mD"}\'')
print(' python main.py subtitle list \'{"url": "BV1xx411c7mD"}\'')
print(' python main.py player get_danmaku \'{"url": "BV1xx411c7mD"}\'')
sys.exit(1)
skill_name = sys.argv[1]
action = sys.argv[2]
params = json.loads(sys.argv[3]) if len(sys.argv) > 3 else {}
app = BilibiliAllInOne()
result = await app.execute(skill_name=skill_name, action=action, **params)
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())
httpx>=0.24.0
bilibili-api-python>=16.0.0
aiohttp>=3.8.0
beautifulsoup4>=4.12.0
lxml>=4.9.0
requests>=2.31.0
{
"name": "bilibili-all-in-one",
"description": "A comprehensive Bilibili toolkit that integrates hot trending monitoring, video downloading, video watching/playback, subtitle downloading, and video publishing capabilities into a single skill.",
"version": "1.0.12",
"author": "wscats",
"license": "MIT",
"homepage": "https://github.com/wscats/bilibili-all-in-one",
"repository": "https://github.com/wscats/bilibili-all-in-one",
"source": "https://github.com/wscats/bilibili-all-in-one",
"type": "code",
"runtime": "python>=3.8",
"entry_point": "main.py",
"required_env_vars": ["BILIBILI_SESSDATA", "BILIBILI_BILI_JCT"],
"optional_env_vars": ["BILIBILI_BUVID3"],
"install": {
"type": "pip",
"command": "pip install -r requirements.txt",
"packages": [
"httpx>=0.24.0",
"bilibili-api-python>=16.0.0",
"aiohttp>=3.8.0",
"beautifulsoup4>=4.12.0",
"lxml>=4.9.0",
"requests>=2.31.0"
],
"source": "pypi"
},
"env_vars": [
{
"name": "BILIBILI_SESSDATA",
"type": "string",
"description": "Bilibili SESSDATA cookie for authenticated API access (publishing, high-quality downloads).",
"required": true,
"sensitive": true
},
{
"name": "BILIBILI_BILI_JCT",
"type": "string",
"description": "Bilibili bili_jct cookie (CSRF token) for all write operations (upload, edit, delete).",
"required": true,
"sensitive": true
},
{
"name": "BILIBILI_BUVID3",
"type": "string",
"description": "Bilibili buvid3 cookie for device identification in authenticated sessions.",
"required": false,
"sensitive": true
}
],
"credentials": {
"primary_credential": "BILIBILI_SESSDATA",
"storage": "in-memory by default; disk persistence only via explicit auth.save_to_file() call with 0600 permissions",
"security_notes": [
"Credentials are NEVER persisted to disk unless the user explicitly calls save_to_file().",
"Prefer passing credentials via environment variables or in-memory parameters.",
"All API requests go to official Bilibili endpoints and YouTube oEmbed API over HTTPS.",
"Use a test/throwaway Bilibili account for evaluation purposes."
]
},
"network": {
"protocol": "HTTPS only",
"allowed_domains": [
"api.bilibili.com",
"member.bilibili.com",
"upos-sz-upcdnbda2.bilivideo.com",
"www.bilibili.com",
"www.youtube.com"
],
"third_party_telemetry": false,
"analytics": false
},
"files": [
"main.py",
"requirements.txt",
"skill.json",
"skill.md",
"LICENSE",
"README.md",
"src/__init__.py",
"src/auth.py",
"src/utils.py",
"src/hot_monitor.py",
"src/downloader.py",
"src/player.py",
"src/subtitle.py",
"src/watcher.py",
"src/publisher.py",
"tests/__init__.py",
"tests/test_all_skill_examples.py"
],
"skills": [
{
"name": "bilibili_hot_monitor",
"description": "Monitor Bilibili hot/trending videos and topics in real-time. Supports filtering by category, tracking rank changes, and sending notifications.",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["get_hot", "get_trending", "get_weekly", "get_rank"],
"description": "The action to perform: get_hot (hot videos), get_trending (trending topics), get_weekly (weekly must-watch), get_rank (category rankings)"
},
"category": {
"type": "string",
"description": "Category filter (e.g., 'all', 'anime', 'music', 'dance', 'game', 'tech', 'life', 'food', 'car', 'fashion', 'entertainment', 'movie', 'tv')",
"default": "all"
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 20
}
},
"required": ["action"]
}
},
{
"name": "bilibili_downloader",
"description": "Download Bilibili videos with support for multiple quality options, batch downloading, and format selection.",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["download", "get_info", "get_formats", "batch_download"],
"description": "The action to perform: download (download a video), get_info (get video info), get_formats (list available formats), batch_download (download multiple videos)"
},
"url": {
"type": "string",
"description": "Bilibili video URL or BV number (e.g., 'https://www.bilibili.com/video/BV1xx411c7mD' or 'BV1xx411c7mD')"
},
"urls": {
"type": "array",
"items": { "type": "string" },
"description": "List of Bilibili video URLs for batch download"
},
"quality": {
"type": "string",
"enum": ["360p", "480p", "720p", "1080p", "1080p+", "4k"],
"description": "Video quality to download",
"default": "1080p"
},
"output_dir": {
"type": "string",
"description": "Output directory for downloaded files",
"default": "./downloads"
},
"format": {
"type": "string",
"enum": ["mp4", "flv", "mp3"],
"description": "Output format",
"default": "mp4"
}
},
"required": ["action"]
}
},
{
"name": "bilibili_watcher",
"description": "Watch and monitor Bilibili (and YouTube) videos. Track view counts, comments, likes, and other engagement metrics over time.",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["watch", "get_stats", "track", "compare"],
"description": "The action to perform: watch (get video details), get_stats (get engagement stats), track (track video metrics over time), compare (compare multiple videos)"
},
"url": {
"type": "string",
"description": "Video URL (supports Bilibili and YouTube)"
},
"urls": {
"type": "array",
"items": { "type": "string" },
"description": "List of video URLs for comparison"
},
"interval": {
"type": "integer",
"description": "Tracking interval in minutes (for track action)",
"default": 60
},
"duration": {
"type": "integer",
"description": "Tracking duration in hours (for track action)",
"default": 24
}
},
"required": ["action"]
}
},
{
"name": "bilibili_subtitle",
"description": "Download and process subtitles/CC from Bilibili videos. Supports multiple subtitle formats and languages.",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["download", "list", "convert", "merge"],
"description": "The action to perform: download (download subtitles), list (list available subtitles), convert (convert subtitle format), merge (merge multiple subtitle files)"
},
"url": {
"type": "string",
"description": "Bilibili video URL or BV number"
},
"language": {
"type": "string",
"description": "Subtitle language preference (e.g., 'zh-CN', 'en', 'ja')",
"default": "zh-CN"
},
"format": {
"type": "string",
"enum": ["srt", "ass", "vtt", "txt", "json"],
"description": "Output subtitle format",
"default": "srt"
},
"output_dir": {
"type": "string",
"description": "Output directory for subtitle files",
"default": "./subtitles"
}
},
"required": ["action"]
}
},
{
"name": "bilibili_player",
"description": "Play Bilibili videos with support for playback control, playlist management, and danmaku (bullet comments) display.",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["play", "get_playurl", "get_danmaku", "get_playlist"],
"description": "The action to perform: play (play a video), get_playurl (get direct play URL), get_danmaku (get danmaku/bullet comments), get_playlist (get playlist info)"
},
"url": {
"type": "string",
"description": "Bilibili video URL or BV number"
},
"quality": {
"type": "string",
"enum": ["360p", "480p", "720p", "1080p", "1080p+", "4k"],
"description": "Playback quality",
"default": "1080p"
},
"danmaku": {
"type": "boolean",
"description": "Whether to include danmaku (bullet comments)",
"default": true
},
"page": {
"type": "integer",
"description": "Page/episode number for multi-part videos",
"default": 1
}
},
"required": ["action"]
}
},
{
"name": "bilibili_publisher",
"description": "Publish videos to Bilibili. Supports uploading videos, setting metadata, scheduling publications, and managing drafts.",
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["upload", "draft", "schedule", "edit", "delete"],
"description": "The action to perform: upload (upload and publish), draft (save as draft), schedule (schedule publication), edit (edit existing video), delete (delete a video)"
},
"file_path": {
"type": "string",
"description": "Path to the video file to upload"
},
"title": {
"type": "string",
"description": "Video title"
},
"description": {
"type": "string",
"description": "Video description"
},
"tags": {
"type": "array",
"items": { "type": "string" },
"description": "Video tags"
},
"category": {
"type": "string",
"description": "Video category (tid)"
},
"cover_path": {
"type": "string",
"description": "Path to cover image"
},
"schedule_time": {
"type": "string",
"description": "Scheduled publish time (ISO 8601 format, for schedule action)"
},
"bvid": {
"type": "string",
"description": "BV number of existing video (for edit/delete actions)"
},
"credential": {
"type": "object",
"description": "Bilibili login credential (sessdata, bili_jct, buvid3)"
}
},
"required": ["action"]
}
}
],
"configuration": {
"bilibili_sessdata": {
"type": "string",
"description": "Bilibili SESSDATA cookie for authenticated operations",
"required": false,
"sensitive": true
},
"bilibili_bili_jct": {
"type": "string",
"description": "Bilibili bili_jct cookie for CSRF protection",
"required": false,
"sensitive": true
},
"bilibili_buvid3": {
"type": "string",
"description": "Bilibili buvid3 cookie for device identification",
"required": false,
"sensitive": true
},
"default_quality": {
"type": "string",
"description": "Default video quality",
"default": "1080p"
},
"default_output_dir": {
"type": "string",
"description": "Default output directory",
"default": "./output"
}
},
"permissions": {
"network": {
"description": "HTTP/HTTPS requests to Bilibili official APIs and YouTube oEmbed API",
"allowed_domains": [
"api.bilibili.com",
"member.bilibili.com",
"www.bilibili.com",
"upos-sz-upcdnbda2.bilivideo.com",
"www.youtube.com"
]
},
"filesystem": {
"description": "Read/write access for downloading videos, subtitles, and uploading files",
"operations": ["read_local_files", "write_downloads", "write_subtitles"],
"note": "Credential files are NEVER auto-created. save_to_file() must be explicitly called by the user."
}
},
"always": false
}
---
name: bilibili-all-in-one
description: >
A comprehensive Bilibili toolkit that integrates hot trending monitoring,
video downloading, video watching/playback, subtitle downloading,
and video publishing capabilities into a single unified skill.
Supports Bilibili session cookie authentication for publishing and
high-quality downloads. Requests go to official Bilibili API endpoints
and YouTube oEmbed API (for YouTube stats) over HTTPS.
version: 1.0.12
type: code
implementation: python
interface: cli-and-api
runtime: python>=3.8
languages:
- zh-CN
- en
tags:
- bilibili
- video-download
- hot-trending
- subtitle
- danmaku
- video-publish
- video-player
- youtube-stats
- batch-download
- multi-format
author: wscats
license: MIT
homepage: https://github.com/wscats/bilibili-all-in-one
repository: https://github.com/wscats/bilibili-all-in-one
entry_point: main.py
required_env_vars:
- BILIBILI_SESSDATA
- BILIBILI_BILI_JCT
optional_env_vars:
- BILIBILI_BUVID3
install: pip install -r requirements.txt
---
# Bilibili All-in-One Skill
A comprehensive Bilibili toolkit that integrates hot trending monitoring, video downloading, video watching/playback, subtitle downloading, and video publishing capabilities into a single unified skill.
> **⚠️ Required Environment Variables:** `BILIBILI_SESSDATA`, `BILIBILI_BILI_JCT` (required), `BILIBILI_BUVID3` (optional)
> These are sensitive Bilibili session cookies needed for authenticated operations (publishing, high-quality downloads).
> Features that do NOT require authentication: hot monitoring, standard-quality downloads, subtitle listing, danmaku, stats viewing.
>
> **📦 Install:** `pip install -r requirements.txt` (all standard PyPI packages: httpx, bilibili-api-python, aiohttp, beautifulsoup4, lxml, requests)
>
> **🔗 Source:** [github.com/wscats/bilibili-all-in-one](https://github.com/wscats/bilibili-all-in-one)
---
### 何时激活
当用户说出或暗示以下内容时，本 Skill 会被激活：
| 触发场景 | 匹配的模块 | 典型触发词 |
|---|---|---|
| 查看B站热门、热搜、排行榜、必看榜 | 🔥 Hot Monitor | "热门"、"热搜"、"排行"、"趋势"、"必看"、"流行"、"榜单" |
| 下载B站视频、提取音频、批量下载 | ⬇️ Downloader | "下载"、"保存视频"、"提取音频"、"导出MP4"、"批量下载" |
| 查看视频播放量、点赞数、数据追踪、对比 | 👀 Watcher | "播放量"、"点赞"、"数据"、"统计"、"对比"、"监控"、"追踪"、"观看量" |
| 下载字幕、转换字幕格式、合并字幕 | 📝 Subtitle | "字幕"、"CC"、"SRT"、"ASS"、"字幕下载"、"字幕转换"、"翻译" |
| 播放视频、获取弹幕、播放列表 | ▶️ Player | "播放"、"弹幕"、"播放地址"、"分P"、"播放列表"、"danmaku" |
| 上传视频、发布、定时发布、草稿、编辑、删除 | 📤 Publisher | "上传"、"发布"、"投稿"、"定时发布"、"草稿"、"编辑视频"、"删除视频" |
| 涉及YouTube视频数据查询 | 👀 Watcher | "YouTube"、"油管"、"YTB"、"YouTube观看量" |
| 提及B站链接或BV号 | 自动识别 | `BV*`、`bilibili.com/video/*`、`b23.tv/*` |
> 💡 **提示**：只要用户消息中包含 B站/Bilibili 相关操作意图，或包含 BV 号、bilibili 链接，本 Skill 即会被自动激活。无需显式声明调用。
---
## Features
| Module | Description |
|---|---|
| 🔥 **Hot Monitor** | Monitor Bilibili hot/trending videos and topics in real-time |
| ⬇️ **Downloader** | Download Bilibili videos with multiple quality and format options |
| 👀 **Watcher** | Watch and track video engagement metrics (supports Bilibili & YouTube) |
| 📝 **Subtitle** | Download and process subtitles in multiple formats and languages |
| ▶️ **Player** | Get playback URLs, danmaku (bullet comments), and playlist info |
| 📤 **Publisher** | Upload, schedule, edit, and manage videos on Bilibili |
## Installation
### Requirements
- Python >= 3.8
- ffmpeg (optional, for merging video/audio streams)
### Install Dependencies
```bash
pip install -r requirements.txt
```
### Dependencies
- `httpx >= 0.24.0`
- `bilibili-api-python >= 16.0.0`
- `aiohttp >= 3.8.0`
- `beautifulsoup4 >= 4.12.0`
- `lxml >= 4.9.0`
- `requests >= 2.31.0`
## Configuration
Some features (downloading high-quality videos, publishing, etc.) require Bilibili authentication. You can provide credentials in three ways:
### 1. Environment Variables
```bash
export BILIBILI_SESSDATA="your_sessdata"
export BILIBILI_BILI_JCT="your_bili_jct"
export BILIBILI_BUVID3="your_buvid3"
```
### 2. Credential File
Create a JSON file (e.g., `credentials.json`):
```json
{
"sessdata": "your_sessdata",
"bili_jct": "your_bili_jct",
"buvid3": "your_buvid3"
}
```
### 3. Direct Parameters
Pass credentials directly when initializing:
```python
from main import BilibiliAllInOne
app = BilibiliAllInOne(
sessdata="your_sessdata",
bili_jct="your_bili_jct",
buvid3="your_buvid3",
)
```
> **How to get cookies:** Log in to [bilibili.com](https://www.bilibili.com), open browser DevTools (F12) → Application → Cookies, and copy the values of `SESSDATA`, `bili_jct`, and `buvid3`.
## ⚠️ Security & Privacy
### Credential Handling
This skill handles **sensitive Bilibili session cookies**. Please read the following carefully:
| Concern | Detail |
|---|---|
| **What credentials are needed?** | `SESSDATA`, `bili_jct`, `buvid3` — Bilibili browser cookies |
| **Which features require authentication?** | Publishing (upload/edit/delete/schedule/draft), downloading 1080p+/4K quality videos |
| **Which features work WITHOUT credentials?** | Hot monitoring, standard-quality downloads, subtitle listing, danmaku fetching, stats viewing |
| **Where are credentials sent?** | To official Bilibili API endpoints (`api.bilibili.com`, `member.bilibili.com`) over HTTPS. YouTube metadata uses `www.youtube.com/oembed` (no credentials sent) |
| **Are credentials persisted to disk?** | **NO** — unless you explicitly call `auth.save_to_file()`. Credentials stay in memory by default |
| **File permissions for saved credentials** | `0600` (owner read/write only) — restrictive by default |
### Best Practices
1. 🧪 **Use a test account** — Do NOT provide your primary Bilibili account cookies for evaluation/testing purposes.
2. 🔒 **Prefer in-memory credentials** — Pass credentials via environment variables or direct parameters rather than saving to a file.
3. 📁 **If you must save credentials** — Use `auth.save_to_file()` which creates files with `0600` permissions. Delete the file when no longer needed.
4. 🐳 **Run in isolation** — When possible, run this skill in an isolated container/environment and inspect network traffic.
5. 🌐 **Verify network traffic** — All HTTP requests go to Bilibili's official domains and YouTube oEmbed API only. You can verify by monitoring outbound connections.
6. ❌ **No exfiltration** — This skill does NOT send credentials to any third-party service, analytics endpoint, or telemetry server.
### Network Endpoints Used
| Domain | Purpose |
|---|---|
| `api.bilibili.com` | Video info, stats, hot lists, subtitles, danmaku, playback URLs |
| `member.bilibili.com` | Video publishing (upload, edit, delete) |
| `upos-sz-upcdnbda2.bilivideo.com` | Video file upload CDN |
| `www.bilibili.com` | Web page scraping fallback |
| `www.youtube.com` | YouTube video metadata via oEmbed API (no auth required) |
### Credential Requirement by Module
| Module | Auth Required? | Notes |
|---|---|---|
| 🔥 Hot Monitor | ❌ No | All public APIs |
| ⬇️ Downloader | ⚠️ Optional | Required only for 1080p+ / 4K quality |
| 👀 Watcher | ❌ No | Public stats APIs |
| 📝 Subtitle | ❌ No | Public subtitle APIs |
| ▶️ Player | ⚠️ Optional | Required for high-quality playback URLs |
| 📤 Publisher | ✅ **Required** | All operations need `SESSDATA` + `bili_jct` |
## Usage
### CLI
```bash
python main.py <skill_name> <action> [params_json]
```
### Python API
```python
import asyncio
from main import BilibiliAllInOne
app = BilibiliAllInOne()
async def demo():
result = await app.execute("hot_monitor", "get_hot", limit=5)
print(result)
asyncio.run(demo())
```
---
## Skills Reference
### 1. 🔥 Hot Monitor (`bilibili_hot_monitor`)
Monitor Bilibili hot/trending videos and topics in real-time. Supports filtering by category, tracking rank changes.
#### Actions
| Action | Description | Parameters |
|---|---|---|
| `get_hot` | Get popular/hot videos | `page`, `page_size` |
| `get_trending` | Get trending series/topics | `limit` |
| `get_weekly` | Get weekly must-watch list | `number` (week number, optional) |
| `get_rank` | Get category ranking videos | `category`, `limit` |
#### Supported Categories
`all`, `anime`, `music`, `dance`, `game`, `tech`, `life`, `food`, `car`, `fashion`, `entertainment`, `movie`, `tv`
#### Examples
```bash
# Get top 10 hot videos
python main.py hot_monitor get_hot '{"page_size": 10}'
# Get trending topics
python main.py hot_monitor get_trending '{"limit": 5}'
# Get this week's must-watch
python main.py hot_monitor get_weekly
# Get game category rankings
python main.py hot_monitor get_rank '{"category": "game", "limit": 10}'
```
```python
# Python API
result = await app.execute("hot_monitor", "get_hot", page_size=10)
result = await app.execute("hot_monitor", "get_rank", category="game", limit=10)
```
---
### 2. ⬇️ Downloader (`bilibili_downloader`)
Download Bilibili videos with support for multiple quality options, batch downloading, and format selection.
#### Actions
| Action | Description | Parameters |
|---|---|---|
| `get_info` | Get video information | `url` |
| `get_formats` | List available qualities/formats | `url` |
| `download` | Download a single video | `url`, `quality`, `output_dir`, `format`, `page` |
| `batch_download` | Download multiple videos | `urls`, `quality`, `output_dir`, `format` |
#### Quality Options
`360p`, `480p`, `720p`, `1080p` (default), `1080p+`, `4k`
#### Format Options
`mp4` (default), `flv`, `mp3` (audio only)
#### Examples
```bash
# Get video info
python main.py downloader get_info '{"url": "BV1xx411c7mD"}'
# List available formats
python main.py downloader get_formats '{"url": "BV1xx411c7mD"}'
# Download in 1080p MP4
python main.py downloader download '{"url": "BV1xx411c7mD", "quality": "1080p", "format": "mp4"}'
# Extract audio only
python main.py downloader download '{"url": "BV1xx411c7mD", "format": "mp3"}'
# Batch download
python main.py downloader batch_download '{"urls": ["BV1xx411c7mD", "BV1yy411c8nE"], "quality": "720p"}'
```
```python
# Python API
info = await app.execute("downloader", "get_info", url="BV1xx411c7mD")
result = await app.execute("downloader", "download", url="BV1xx411c7mD", quality="1080p")
```
---
### 3. 👀 Watcher (`bilibili_watcher`)
Watch and monitor Bilibili (and YouTube) videos. Track view counts, comments, likes, and other engagement metrics over time.
#### Actions
| Action | Description | Parameters |
|---|---|---|
| `watch` | Get detailed video information | `url` |
| `get_stats` | Get current engagement statistics | `url` |
| `track` | Track metrics over time | `url`, `interval` (minutes), `duration` (hours) |
| `compare` | Compare multiple videos | `urls` |
#### Supported Platforms
- **Bilibili**: `https://www.bilibili.com/video/BVxxxxxx` or `BVxxxxxx`
- **YouTube**: `https://www.youtube.com/watch?v=xxxxx` or `https://youtu.be/xxxxx`
#### Examples
```bash
# Get video details
python main.py watcher watch '{"url": "BV1xx411c7mD"}'
# Get current stats
python main.py watcher get_stats '{"url": "BV1xx411c7mD"}'
# Track views every 30 minutes for 12 hours
python main.py watcher track '{"url": "BV1xx411c7mD", "interval": 30, "duration": 12}'
# Compare multiple videos
python main.py watcher compare '{"urls": ["BV1xx411c7mD", "BV1yy411c8nE"]}'
```
```python
# Python API
details = await app.execute("watcher", "watch", url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
comparison = await app.execute("watcher", "compare", urls=["BV1xx411c7mD", "BV1yy411c8nE"])
```
---
### 4. 📝 Subtitle (`bilibili_subtitle`)
Download and process subtitles/CC from Bilibili videos. Supports multiple subtitle formats and languages.
#### Actions
| Action | Description | Parameters |
|---|---|---|
| `list` | List available subtitles | `url` |
| `download` | Download subtitles | `url`, `language`, `format`, `output_dir` |
| `convert` | Convert subtitle format | `input_path`, `output_format`, `output_dir` |
| `merge` | Merge multiple subtitle files | `input_paths`, `output_path`, `output_format` |
#### Supported Formats
`srt` (default), `ass`, `vtt`, `txt`, `json`
#### Supported Languages
`zh-CN` (default), `en`, `ja`, and other language codes available on the video.
#### Examples
```bash
# List available subtitles
python main.py subtitle list '{"url": "BV1xx411c7mD"}'
# Download Chinese subtitles in SRT format
python main.py subtitle download '{"url": "BV1xx411c7mD", "language": "zh-CN", "format": "srt"}'
# Download English subtitles in ASS format
python main.py subtitle download '{"url": "BV1xx411c7mD", "language": "en", "format": "ass"}'
# Convert SRT to VTT
python main.py subtitle convert '{"input_path": "./subtitles/video.srt", "output_format": "vtt"}'
# Merge subtitle files
python main.py subtitle merge '{"input_paths": ["part1.srt", "part2.srt"], "output_path": "merged.srt"}'
```
```python
# Python API
subs = await app.execute("subtitle", "list", url="BV1xx411c7mD")
result = await app.execute("subtitle", "download", url="BV1xx411c7mD", language="zh-CN", format="srt")
```
---
### 5. ▶️ Player (`bilibili_player`)
Play Bilibili videos with support for playback control, playlist management, and danmaku (bullet comments) display.
#### Actions
| Action | Description | Parameters |
|---|---|---|
| `play` | Get complete playback info | `url`, `quality`, `page` |
| `get_playurl` | Get direct play URLs | `url`, `quality`, `page` |
| `get_danmaku` | Get danmaku/bullet comments | `url`, `page`, `segment` |
| `get_playlist` | Get playlist/multi-part info | `url` |
#### Danmaku Modes
| Mode | Description |
|---|---|
| 1 | Scroll (right to left) |
| 4 | Bottom fixed |
| 5 | Top fixed |
#### Examples
```bash
# Get playback info
python main.py player play '{"url": "BV1xx411c7mD", "quality": "1080p"}'
# Get direct play URLs
python main.py player get_playurl '{"url": "BV1xx411c7mD", "quality": "720p"}'
# Get danmaku
python main.py player get_danmaku '{"url": "BV1xx411c7mD"}'
# Get playlist for multi-part video
python main.py player get_playlist '{"url": "BV1xx411c7mD"}'
# Get page 3 of a multi-part video
python main.py player play '{"url": "BV1xx411c7mD", "quality": "1080p", "page": 3}'
```
```python
# Python API
play_info = await app.execute("player", "play", url="BV1xx411c7mD", quality="1080p")
danmaku = await app.execute("player", "get_danmaku", url="BV1xx411c7mD")
playlist = await app.execute("player", "get_playlist", url="BV1xx411c7mD")
```
---
### 6. 📤 Publisher (`bilibili_publisher`)
Publish videos to Bilibili. Supports uploading videos, setting metadata, scheduling publications, and managing drafts.
> ⚠️ **Authentication Required**: All publisher actions require valid Bilibili credentials.
#### Actions
| Action | Description | Parameters |
|---|---|---|
| `upload` | Upload and publish a video | `file_path`, `title`, `description`, `tags`, `category`, `cover_path`, `dynamic`, `no_reprint`, `open_elec` |
| `draft` | Save as draft | `file_path`, `title`, `description`, `tags`, `category`, `cover_path` |
| `schedule` | Schedule future publication | `file_path`, `title`, `schedule_time`, `description`, `tags`, `category`, `cover_path` |
| `edit` | Edit existing video metadata | `bvid`, `title`, `description`, `tags`, `cover_path` |
| `delete` | Delete a video | `bvid` |
#### Upload Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
| `file_path` | string | *required* | Path to the video file |
| `title` | string | *required* | Video title (max 80 chars) |
| `description` | string | `""` | Video description (max 2000 chars) |
| `tags` | string[] | `["bilibili"]` | Tags (max 12, each max 20 chars) |
| `category` | string | `"171"` | Category TID |
| `cover_path` | string | `null` | Path to cover image (JPG/PNG) |
| `no_reprint` | int | `1` | 1 = original content, 0 = repost |
| `open_elec` | int | `0` | 1 = enable charging, 0 = disable |
#### Examples
```bash
# Upload and publish
python main.py publisher upload '{"file_path": "./video.mp4", "title": "My Video", "description": "Hello World", "tags": ["test", "demo"], "category": "171"}'
# Save as draft
python main.py publisher draft '{"file_path": "./video.mp4", "title": "Draft Video"}'
# Schedule publication
python main.py publisher schedule '{"file_path": "./video.mp4", "title": "Scheduled Video", "schedule_time": "2025-12-31T20:00:00+08:00"}'
# Edit video metadata
python main.py publisher edit '{"bvid": "BV1xx411c7mD", "title": "New Title", "tags": ["updated"]}'
# Delete a video
python main.py publisher delete '{"bvid": "BV1xx411c7mD"}'
```
```python
# Python API (authentication required)
app = BilibiliAllInOne(sessdata="xxx", bili_jct="xxx", buvid3="xxx")
result = await app.execute("publisher", "upload",
file_path="./video.mp4",
title="My Video",
description="Published via bilibili-all-in-one",
tags=["python", "bilibili"],
)
```
---
## Project Structure
```
bilibili-all-in-one/
├── skill.json # Skill configuration & parameter schema
├── skill.md # This documentation file
├── README.md # Project README (Chinese)
├── LICENSE # MIT License
├── requirements.txt # Python dependencies
├── .gitignore # Git ignore rules
├── main.py # Entry point & unified BilibiliAllInOne class
├── src/
│ ├── __init__.py # Package exports
│ ├── auth.py # Authentication & credential management
│ ├── utils.py # Shared utilities, API constants, helpers
│ ├── hot_monitor.py # Hot/trending video monitoring
│ ├── downloader.py # Video downloading
│ ├── watcher.py # Video watching & stats tracking
│ ├── subtitle.py # Subtitle downloading & processing
│ ├── player.py # Video playback & danmaku
│ └── publisher.py # Video uploading & publishing
└── tests/
├── __init__.py
└── test_all_skill_examples.py # Comprehensive unit tests
```
## Skill Origin
This skill integrates the functionality of the following individual skills into one unified toolkit:
| Original Skill | Source | Integrated Module |
|---|---|---|
| bilibili-hot-monitor | [Jacobzwj/bilibili-hot-monitor](https://clawhub.ai/Jacobzwj/bilibili-hot-monitor) | `hot_monitor` |
| bililidownloader | [caiyundc880518/bililidownloader](https://clawhub.ai/caiyundc880518/bililidownloader) | `downloader` |
| bilibili-youtube-watcher | [donnycui/bilibili-youtube-watcher](https://clawhub.ai/donnycui/bilibili-youtube-watcher) | `watcher` |
| bilibili-subtitle-download-skill | [DavinciEvans/bilibili-subtitle-download-skill](https://clawhub.ai/DavinciEvans/bilibili-subtitle-download-skill) | `subtitle` |
| bilibili-player | [e421083458/bilibili-player](https://clawhub.ai/e421083458/bilibili-player) | `player` |
| bilibili-video-publish | [Johnnyxu820/bilibili-video-publish](https://clawhub.ai/Johnnyxu820/bilibili-video-publish) | `publisher` |
## Response Format
All skill actions return a JSON object with a unified structure:
```json
{
"success": true,
"...": "action-specific fields"
}
```
On error:
```json
{
"success": false,
"message": "Error description"
}
```
## License
MIT
from src.auth import BilibiliAuth
from src.hot_monitor import HotMonitor
from src.downloader import BilibiliDownloader
from src.watcher import BilibiliWatcher
from src.subtitle import SubtitleDownloader
from src.player import BilibiliPlayer
from src.publisher import BilibiliPublisher
from src.utils import extract_bvid
__all__ = [
"BilibiliAuth",
"HotMonitor",
"BilibiliDownloader",
"BilibiliWatcher",
"SubtitleDownloader",
"BilibiliPlayer",
"BilibiliPublisher",
"extract_bvid",
]
"""Authentication and credential management for Bilibili API."""
import json
import os
from typing import Optional, Dict, Any
import httpx
from .utils import DEFAULT_HEADERS, API_BASE
class BilibiliAuth:
"""Manage Bilibili authentication credentials and cookies.
Supports login via SESSDATA cookie, QR code, and credential file.
"""
def __init__(
self,
sessdata: Optional[str] = None,
bili_jct: Optional[str] = None,
buvid3: Optional[str] = None,
credential_file: Optional[str] = None,
):
"""Initialize BilibiliAuth.
Args:
sessdata: SESSDATA cookie value.
bili_jct: bili_jct cookie value (CSRF token).
buvid3: buvid3 cookie value.
credential_file: Path to a JSON file containing credentials.
"""
self.sessdata = sessdata
self.bili_jct = bili_jct
self.buvid3 = buvid3
if credential_file and os.path.exists(credential_file):
self._load_from_file(credential_file)
# Try environment variables as fallback
if not self.sessdata:
self.sessdata = os.environ.get("BILIBILI_SESSDATA", "")
if not self.bili_jct:
self.bili_jct = os.environ.get("BILIBILI_BILI_JCT", "")
if not self.buvid3:
self.buvid3 = os.environ.get("BILIBILI_BUVID3", "")
def _load_from_file(self, filepath: str) -> None:
"""Load credentials from a JSON file.
Args:
filepath: Path to the credential JSON file.
"""
with open(filepath, "r", encoding="utf-8") as f:
cred = json.load(f)
self.sessdata = cred.get("sessdata", self.sessdata)
self.bili_jct = cred.get("bili_jct", self.bili_jct)
self.buvid3 = cred.get("buvid3", self.buvid3)
@property
def is_authenticated(self) -> bool:
"""Check if valid credentials are available."""
return bool(self.sessdata and self.bili_jct)
@property
def cookies(self) -> Dict[str, str]:
"""Get cookies dict for HTTP requests."""
cookies = {}
if self.sessdata:
cookies["SESSDATA"] = self.sessdata
if self.bili_jct:
cookies["bili_jct"] = self.bili_jct
if self.buvid3:
cookies["buvid3"] = self.buvid3
return cookies
@property
def csrf(self) -> str:
"""Get CSRF token (bili_jct)."""
return self.bili_jct or ""
def get_headers(self, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
"""Get HTTP headers with authentication.
Args:
extra: Additional headers to include.
Returns:
Headers dictionary.
"""
headers = DEFAULT_HEADERS.copy()
if extra:
headers.update(extra)
return headers
def get_client(self) -> httpx.AsyncClient:
"""Create an authenticated async HTTP client.
Returns:
httpx.AsyncClient configured with credentials.
"""
return httpx.AsyncClient(
headers=self.get_headers(),
cookies=self.cookies,
timeout=30.0,
follow_redirects=True,
)
async def verify(self) -> Dict[str, Any]:
"""Verify the current credentials by calling the user info API.
Returns:
User info dict if credentials are valid, error dict otherwise.
"""
if not self.is_authenticated:
return {"success": False, "message": "No credentials provided"}
async with self.get_client() as client:
resp = await client.get(f"{API_BASE}/x/web-interface/nav")
data = resp.json()
if data.get("code") == 0:
info = data["data"]
return {
"success": True,
"uid": info.get("mid"),
"username": info.get("uname"),
"vip_type": info.get("vipType"),
"level": info.get("level_info", {}).get("current_level"),
}
return {"success": False, "message": data.get("message", "Unknown error")}
def save_to_file(self, filepath: str) -> None:
"""Save current credentials to a JSON file.
WARNING: This persists sensitive session cookies to disk. Only call this
method when explicitly requested by the user. The file is created with
restrictive permissions (owner read/write only, 0600) to minimize
exposure risk.
Args:
filepath: Path to save the credential file.
"""
cred = {
"sessdata": self.sessdata,
"bili_jct": self.bili_jct,
"buvid3": self.buvid3,
}
os.makedirs(os.path.dirname(filepath) or ".", exist_ok=True)
# Open with restrictive permissions (0600 = owner read/write only)
fd = os.open(filepath, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(cred, f, indent=2)
except Exception:
os.close(fd)
raise
"""Bilibili video downloading module."""
import os
import asyncio
from typing import Optional, Dict, Any, List
import httpx
from .auth import BilibiliAuth
from .utils import (
API_VIDEO_INFO,
API_PLAY_URL,
QUALITY_MAP,
DEFAULT_HEADERS,
extract_bvid,
format_duration,
format_number,
ensure_dir,
sanitize_filename,
)
class BilibiliDownloader:
"""Download videos from Bilibili.
Supports multiple quality options, format selection, and batch downloading.
"""
def __init__(self, auth: Optional[BilibiliAuth] = None, output_dir: str = "./downloads"):
"""Initialize BilibiliDownloader.
Args:
auth: Optional BilibiliAuth instance for authenticated requests.
output_dir: Default output directory for downloaded files.
"""
self.auth = auth
self.output_dir = output_dir
def _get_client(self) -> httpx.AsyncClient:
"""Get an HTTP client, using auth if available."""
if self.auth:
return self.auth.get_client()
return httpx.AsyncClient(
headers=DEFAULT_HEADERS,
timeout=60.0,
follow_redirects=True,
)
async def get_info(self, url: str) -> Dict[str, Any]:
"""Get video information.
Args:
url: Bilibili video URL or BV number.
Returns:
Video information dict.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
video = data["data"]
stat = video.get("stat", {})
owner = video.get("owner", {})
pages = []
for p in video.get("pages", []):
pages.append({
"page": p.get("page"),
"cid": p.get("cid"),
"title": p.get("part"),
"duration": format_duration(p.get("duration", 0)),
})
return {
"success": True,
"bvid": video.get("bvid"),
"aid": video.get("aid"),
"title": video.get("title"),
"description": video.get("desc"),
"cover": video.get("pic"),
"duration": format_duration(video.get("duration", 0)),
"author": {
"mid": owner.get("mid"),
"name": owner.get("name"),
},
"stats": {
"views": format_number(stat.get("view", 0)),
"likes": format_number(stat.get("like", 0)),
"coins": format_number(stat.get("coin", 0)),
"favorites": format_number(stat.get("favorite", 0)),
"danmaku": stat.get("danmaku", 0),
},
"pages": pages,
"url": f"https://www.bilibili.com/video/{video.get('bvid')}",
}
async def get_formats(self, url: str) -> Dict[str, Any]:
"""Get available download formats and qualities for a video.
Args:
url: Bilibili video URL or BV number.
Returns:
Available formats and qualities.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
# First get video info to get cid
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
cid = data["data"]["pages"][0]["cid"]
# Get play URL to see available qualities
async with self._get_client() as client:
resp = await client.get(
API_PLAY_URL,
params={
"bvid": bvid,
"cid": cid,
"fnval": 4048,
"fourk": 1,
},
)
play_data = resp.json()
if play_data.get("code") != 0:
return {"success": False, "message": play_data.get("message", "API error")}
dash = play_data.get("data", {})
quality_names = {v: k for k, v in QUALITY_MAP.items()}
available_qualities = []
for qn in dash.get("accept_quality", []):
name = quality_names.get(qn, f"qn_{qn}")
available_qualities.append({
"quality": name,
"qn": qn,
})
return {
"success": True,
"bvid": bvid,
"available_qualities": available_qualities,
"formats": ["mp4", "flv", "mp3"],
}
async def download(
self,
url: str,
quality: str = "1080p",
output_dir: Optional[str] = None,
format: str = "mp4",
page: int = 1,
) -> Dict[str, Any]:
"""Download a single video.
Args:
url: Bilibili video URL or BV number.
quality: Desired quality ('360p', '480p', '720p', '1080p', '1080p+', '4k').
output_dir: Output directory (uses default if not specified).
format: Output format ('mp4', 'flv', 'mp3').
page: Page/episode number for multi-part videos.
Returns:
Download result dict with file path.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
out_dir = ensure_dir(output_dir or self.output_dir)
qn = QUALITY_MAP.get(quality, 80)
# Get video info
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
video = data["data"]
title = sanitize_filename(video.get("title", bvid))
pages = video.get("pages", [])
if page > len(pages):
return {"success": False, "message": f"Page {page} not found, video has {len(pages)} pages"}
cid = pages[page - 1]["cid"]
page_title = pages[page - 1].get("part", "")
if len(pages) > 1 and page_title:
filename = f"{title}_P{page}_{sanitize_filename(page_title)}.{format}"
else:
filename = f"{title}.{format}"
filepath = os.path.join(out_dir, filename)
# Get download URL
async with self._get_client() as client:
resp = await client.get(
API_PLAY_URL,
params={
"bvid": bvid,
"cid": cid,
"qn": qn,
"fnval": 4048 if format != "flv" else 0,
"fourk": 1,
},
)
play_data = resp.json()
if play_data.get("code") != 0:
return {"success": False, "message": play_data.get("message", "API error")}
# Extract download URLs from DASH or legacy format
dash_data = play_data.get("data", {}).get("dash")
if dash_data and format != "flv":
video_url = self._select_dash_stream(dash_data.get("video", []), qn)
audio_url = self._select_dash_audio(dash_data.get("audio", []))
if not video_url:
return {"success": False, "message": "No suitable video stream found"}
if format == "mp3":
# Audio only
if not audio_url:
return {"success": False, "message": "No audio stream found"}
filepath = filepath.replace(f".{format}", ".mp3")
await self._download_stream(audio_url, filepath)
else:
# Download video and audio separately, then combine
video_tmp = filepath + ".video.tmp"
audio_tmp = filepath + ".audio.tmp"
await asyncio.gather(
self._download_stream(video_url, video_tmp),
self._download_stream(audio_url, audio_tmp) if audio_url else asyncio.sleep(0),
)
if audio_url and os.path.exists(audio_tmp):
# Merge video and audio (requires ffmpeg)
merge_result = await self._merge_streams(video_tmp, audio_tmp, filepath)
# Clean up temp files
for tmp in [video_tmp, audio_tmp]:
if os.path.exists(tmp):
os.remove(tmp)
if not merge_result:
# Fallback: rename video file
os.rename(video_tmp, filepath)
else:
os.rename(video_tmp, filepath)
else:
# Legacy FLV format
durl = play_data.get("data", {}).get("durl", [])
if not durl:
return {"success": False, "message": "No download URL found"}
await self._download_stream(durl[0]["url"], filepath)
file_size = os.path.getsize(filepath) if os.path.exists(filepath) else 0
return {
"success": True,
"bvid": bvid,
"title": video.get("title"),
"quality": quality,
"format": format,
"filepath": filepath,
"file_size": file_size,
"file_size_mb": round(file_size / (1024 * 1024), 2),
}
async def batch_download(
self,
urls: List[str],
quality: str = "1080p",
output_dir: Optional[str] = None,
format: str = "mp4",
) -> Dict[str, Any]:
"""Download multiple videos.
Args:
urls: List of Bilibili video URLs or BV numbers.
quality: Desired quality.
output_dir: Output directory.
format: Output format.
Returns:
Batch download results.
"""
results = []
for url in urls:
result = await self.download(
url=url,
quality=quality,
output_dir=output_dir,
format=format,
)
results.append(result)
succeeded = sum(1 for r in results if r.get("success"))
return {
"success": True,
"total": len(urls),
"succeeded": succeeded,
"failed": len(urls) - succeeded,
"results": results,
}
async def execute(self, action: str, **kwargs) -> Dict[str, Any]:
"""Execute a downloader action.
Args:
action: Action name ('download', 'get_info', 'get_formats', 'batch_download').
**kwargs: Additional parameters for the action.
Returns:
Action result dict.
"""
actions = {
"download": self.download,
"get_info": self.get_info,
"get_formats": self.get_formats,
"batch_download": self.batch_download,
}
handler = actions.get(action)
if not handler:
return {"success": False, "message": f"Unknown action: {action}"}
import inspect
sig = inspect.signature(handler)
valid_params = {k: v for k, v in kwargs.items() if k in sig.parameters}
return await handler(**valid_params)
async def _download_stream(self, url: str, filepath: str) -> None:
"""Download a stream URL to a file.
Args:
url: Stream URL to download.
filepath: Destination file path.
"""
headers = DEFAULT_HEADERS.copy()
headers["Referer"] = "https://www.bilibili.com"
async with httpx.AsyncClient(
headers=headers,
timeout=300.0,
follow_redirects=True,
) as client:
async with client.stream("GET", url) as resp:
with open(filepath, "wb") as f:
async for chunk in resp.aiter_bytes(chunk_size=1024 * 1024):
f.write(chunk)
@staticmethod
async def _merge_streams(video_path: str, audio_path: str, output_path: str) -> bool:
"""Merge video and audio streams using ffmpeg.
Args:
video_path: Path to the video file.
audio_path: Path to the audio file.
output_path: Path for the merged output file.
Returns:
True if merge succeeded, False otherwise.
"""
try:
proc = await asyncio.create_subprocess_exec(
"ffmpeg", "-y",
"-i", video_path,
"-i", audio_path,
"-c", "copy",
output_path,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await proc.wait()
return proc.returncode == 0
except FileNotFoundError:
return False
@staticmethod
def _select_dash_stream(streams: List[Dict], target_qn: int) -> Optional[str]:
"""Select the best matching DASH video stream.
Args:
streams: List of available video streams.
target_qn: Target quality number.
Returns:
Stream URL or None.
"""
if not streams:
return None
# Sort by quality (descending) and find best match
sorted_streams = sorted(streams, key=lambda s: s.get("id", 0), reverse=True)
# Try exact match first
for s in sorted_streams:
if s.get("id") == target_qn:
return s.get("baseUrl") or s.get("base_url")
# Fall back to the best available that doesn't exceed target
for s in sorted_streams:
if s.get("id", 0) <= target_qn:
return s.get("baseUrl") or s.get("base_url")
# If nothing below target, return lowest available
return sorted_streams[-1].get("baseUrl") or sorted_streams[-1].get("base_url")
@staticmethod
def _select_dash_audio(streams: List[Dict]) -> Optional[str]:
"""Select the best DASH audio stream.
Args:
streams: List of available audio streams.
Returns:
Stream URL or None.
"""
if not streams:
return None
# Sort by bandwidth (descending) and pick the best
sorted_streams = sorted(streams, key=lambda s: s.get("bandwidth", 0), reverse=True)
return sorted_streams[0].get("baseUrl") or sorted_streams[0].get("base_url")
"""Bilibili hot/trending video monitoring module."""
import asyncio
from typing import Optional, Dict, Any, List
import httpx
from .auth import BilibiliAuth
from .utils import (
API_HOT,
API_TRENDING,
API_WEEKLY,
API_RANK,
CATEGORY_TID,
DEFAULT_HEADERS,
format_number,
format_duration,
)
class HotMonitor:
"""Monitor Bilibili hot and trending videos.
Provides access to popular videos, trending topics, weekly must-watch lists,
and category-specific rankings.
"""
def __init__(self, auth: Optional[BilibiliAuth] = None):
"""Initialize HotMonitor.
Args:
auth: Optional BilibiliAuth instance for authenticated requests.
"""
self.auth = auth
def _get_client(self) -> httpx.AsyncClient:
"""Get an HTTP client, using auth if available."""
if self.auth:
return self.auth.get_client()
return httpx.AsyncClient(
headers=DEFAULT_HEADERS,
timeout=30.0,
follow_redirects=True,
)
async def get_hot(self, page: int = 1, page_size: int = 20) -> Dict[str, Any]:
"""Get popular/hot videos from Bilibili.
Args:
page: Page number (1-indexed).
page_size: Number of results per page.
Returns:
Dict containing list of hot videos and pagination info.
"""
async with self._get_client() as client:
resp = await client.get(
API_HOT,
params={"pn": page, "ps": page_size},
)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
videos = []
for item in data.get("data", {}).get("list", []):
videos.append(self._parse_video(item))
return {
"success": True,
"videos": videos,
"page": page,
"has_more": bool(data.get("data", {}).get("no_more") is False),
}
async def get_trending(self, limit: int = 20) -> Dict[str, Any]:
"""Get trending series/topics list.
Args:
limit: Maximum number of results.
Returns:
Dict containing list of trending series.
"""
async with self._get_client() as client:
resp = await client.get(API_TRENDING)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
series_list = []
for item in data.get("data", {}).get("list", [])[:limit]:
series_list.append({
"number": item.get("number"),
"subject": item.get("subject"),
"status": item.get("status"),
"name": item.get("name"),
})
return {"success": True, "series": series_list}
async def get_weekly(self, number: Optional[int] = None) -> Dict[str, Any]:
"""Get weekly must-watch video list.
Args:
number: Specific week number. If None, gets the latest week.
Returns:
Dict containing the weekly video list.
"""
params = {}
if number is not None:
params["number"] = number
async with self._get_client() as client:
resp = await client.get(API_WEEKLY, params=params)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
config = data.get("data", {}).get("config", {})
videos = []
for item in data.get("data", {}).get("list", []):
videos.append(self._parse_video(item))
return {
"success": True,
"week_number": config.get("number"),
"subject": config.get("subject"),
"label": config.get("label"),
"videos": videos,
}
async def get_rank(
self,
category: str = "all",
limit: int = 20,
) -> Dict[str, Any]:
"""Get category ranking videos.
Args:
category: Category name (see CATEGORY_TID keys).
limit: Maximum number of results.
Returns:
Dict containing ranked video list.
"""
tid = CATEGORY_TID.get(category, 0)
async with self._get_client() as client:
resp = await client.get(
API_RANK,
params={"rid": tid, "type": "all"},
)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
videos = []
for item in data.get("data", {}).get("list", [])[:limit]:
video = self._parse_video(item)
video["score"] = item.get("score")
videos.append(video)
return {
"success": True,
"category": category,
"videos": videos,
}
async def execute(self, action: str, **kwargs) -> Dict[str, Any]:
"""Execute a hot monitor action.
Args:
action: Action name ('get_hot', 'get_trending', 'get_weekly', 'get_rank').
**kwargs: Additional parameters for the action.
Returns:
Action result dict.
"""
actions = {
"get_hot": self.get_hot,
"get_trending": self.get_trending,
"get_weekly": self.get_weekly,
"get_rank": self.get_rank,
}
handler = actions.get(action)
if not handler:
return {"success": False, "message": f"Unknown action: {action}"}
# Filter kwargs to only pass valid parameters
import inspect
sig = inspect.signature(handler)
valid_params = {k: v for k, v in kwargs.items() if k in sig.parameters}
return await handler(**valid_params)
@staticmethod
def _parse_video(item: Dict[str, Any]) -> Dict[str, Any]:
"""Parse a video item from API response.
Args:
item: Raw video data from API.
Returns:
Parsed video information dict.
"""
stat = item.get("stat", {})
owner = item.get("owner", {})
return {
"bvid": item.get("bvid"),
"aid": item.get("aid"),
"title": item.get("title"),
"description": item.get("desc", ""),
"cover": item.get("pic"),
"duration": format_duration(item.get("duration", 0)),
"duration_seconds": item.get("duration", 0),
"author": {
"mid": owner.get("mid"),
"name": owner.get("name"),
"face": owner.get("face"),
},
"stats": {
"views": stat.get("view", 0),
"views_formatted": format_number(stat.get("view", 0)),
"danmaku": stat.get("danmaku", 0),
"likes": stat.get("like", 0),
"coins": stat.get("coin", 0),
"favorites": stat.get("favorite", 0),
"shares": stat.get("share", 0),
"comments": stat.get("reply", 0),
},
"url": f"https://www.bilibili.com/video/{item.get('bvid')}",
"publish_time": item.get("pubdate"),
}
"""Bilibili video playback and danmaku module."""
import asyncio
import struct
import zlib
from typing import Optional, Dict, Any, List
import httpx
from .auth import BilibiliAuth
from .utils import (
API_VIDEO_INFO,
API_PLAY_URL,
API_DANMAKU,
QUALITY_MAP,
DEFAULT_HEADERS,
extract_bvid,
format_duration,
format_number,
)
class BilibiliPlayer:
"""Play Bilibili videos with danmaku (bullet comments) support.
Provides playback URLs, danmaku retrieval, and playlist management.
"""
def __init__(self, auth: Optional[BilibiliAuth] = None):
"""Initialize BilibiliPlayer.
Args:
auth: Optional BilibiliAuth instance for authenticated requests.
"""
self.auth = auth
def _get_client(self) -> httpx.AsyncClient:
"""Get an HTTP client, using auth if available."""
if self.auth:
return self.auth.get_client()
return httpx.AsyncClient(
headers=DEFAULT_HEADERS,
timeout=30.0,
follow_redirects=True,
)
async def play(self, url: str, quality: str = "1080p", page: int = 1) -> Dict[str, Any]:
"""Get complete playback information for a video.
Args:
url: Bilibili video URL or BV number.
quality: Desired playback quality.
page: Page/episode number for multi-part videos.
Returns:
Complete playback info including video details and play URLs.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
# Get video info and play URL in parallel
info_task = self._get_video_info(bvid)
play_url_task = self._get_play_url(bvid, quality, page)
info, play_url_data = await asyncio.gather(info_task, play_url_task)
if not info.get("success"):
return info
result = info.copy()
result.update(play_url_data)
return result
async def get_playurl(
self,
url: str,
quality: str = "1080p",
page: int = 1,
) -> Dict[str, Any]:
"""Get direct play URLs for a video.
Args:
url: Bilibili video URL or BV number.
quality: Desired quality.
page: Page/episode number.
Returns:
Play URL information.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
return await self._get_play_url(bvid, quality, page)
async def get_danmaku(
self,
url: str,
page: int = 1,
segment: int = 1,
) -> Dict[str, Any]:
"""Get danmaku (bullet comments) for a video.
Args:
url: Bilibili video URL or BV number.
page: Page/episode number.
segment: Danmaku segment index (each ~6 minutes).
Returns:
List of danmaku entries.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
# Get CID
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
pages = data["data"].get("pages", [])
if page > len(pages):
return {"success": False, "message": f"Page {page} not found"}
cid = pages[page - 1]["cid"]
# Get danmaku XML
async with self._get_client() as client:
resp = await client.get(
API_DANMAKU,
params={"oid": cid},
)
danmaku_list = self._parse_danmaku_xml(resp.text)
return {
"success": True,
"bvid": bvid,
"cid": cid,
"page": page,
"danmaku_count": len(danmaku_list),
"danmaku": danmaku_list,
}
async def get_playlist(self, url: str) -> Dict[str, Any]:
"""Get playlist/multi-part video information.
Args:
url: Bilibili video URL or BV number.
Returns:
Playlist information with all pages/episodes.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
video = data["data"]
pages = []
for p in video.get("pages", []):
pages.append({
"page": p.get("page"),
"cid": p.get("cid"),
"title": p.get("part"),
"duration": format_duration(p.get("duration", 0)),
"duration_seconds": p.get("duration", 0),
})
# Also check for season/collection info
ugc_season = video.get("ugc_season")
season_info = None
if ugc_season:
episodes = []
for section in ugc_season.get("sections", []):
for ep in section.get("episodes", []):
episodes.append({
"bvid": ep.get("bvid"),
"aid": ep.get("aid"),
"title": ep.get("title"),
"arc": {
"duration": format_duration(ep.get("arc", {}).get("duration", 0)),
},
})
season_info = {
"title": ugc_season.get("title"),
"episode_count": len(episodes),
"episodes": episodes,
}
return {
"success": True,
"bvid": video.get("bvid"),
"title": video.get("title"),
"page_count": len(pages),
"pages": pages,
"season": season_info,
}
async def execute(self, action: str, **kwargs) -> Dict[str, Any]:
"""Execute a player action.
Args:
action: Action name ('play', 'get_playurl', 'get_danmaku', 'get_playlist').
**kwargs: Additional parameters for the action.
Returns:
Action result dict.
"""
actions = {
"play": self.play,
"get_playurl": self.get_playurl,
"get_danmaku": self.get_danmaku,
"get_playlist": self.get_playlist,
}
handler = actions.get(action)
if not handler:
return {"success": False, "message": f"Unknown action: {action}"}
import inspect
sig = inspect.signature(handler)
valid_params = {k: v for k, v in kwargs.items() if k in sig.parameters}
return await handler(**valid_params)
async def _get_video_info(self, bvid: str) -> Dict[str, Any]:
"""Get video information.
Args:
bvid: BV ID.
Returns:
Video info dict.
"""
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
video = data["data"]
stat = video.get("stat", {})
owner = video.get("owner", {})
return {
"success": True,
"bvid": video.get("bvid"),
"aid": video.get("aid"),
"title": video.get("title"),
"description": video.get("desc"),
"cover": video.get("pic"),
"duration": format_duration(video.get("duration", 0)),
"author": {
"mid": owner.get("mid"),
"name": owner.get("name"),
},
"stats": {
"views": format_number(stat.get("view", 0)),
"danmaku": stat.get("danmaku", 0),
"likes": format_number(stat.get("like", 0)),
},
}
async def _get_play_url(self, bvid: str, quality: str, page: int) -> Dict[str, Any]:
"""Get play URL for a video.
Args:
bvid: BV ID.
quality: Desired quality.
page: Page number.
Returns:
Play URL info dict.
"""
# Get CID
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"play_url": None, "message": data.get("message", "API error")}
pages = data["data"].get("pages", [])
if page > len(pages):
return {"play_url": None, "message": f"Page {page} not found"}
cid = pages[page - 1]["cid"]
qn = QUALITY_MAP.get(quality, 80)
async with self._get_client() as client:
resp = await client.get(
API_PLAY_URL,
params={
"bvid": bvid,
"cid": cid,
"qn": qn,
"fnval": 4048,
"fourk": 1,
},
)
play_data = resp.json()
if play_data.get("code") != 0:
return {"play_url": None, "message": play_data.get("message", "API error")}
dash = play_data.get("data", {}).get("dash")
if dash:
video_streams = []
for v in dash.get("video", []):
quality_names = {val: key for key, val in QUALITY_MAP.items()}
video_streams.append({
"quality": quality_names.get(v.get("id"), f"qn_{v.get('id')}"),
"qn": v.get("id"),
"codecs": v.get("codecs"),
"bandwidth": v.get("bandwidth"),
"url": v.get("baseUrl") or v.get("base_url"),
})
audio_streams = []
for a in dash.get("audio", []):
audio_streams.append({
"bandwidth": a.get("bandwidth"),
"codecs": a.get("codecs"),
"url": a.get("baseUrl") or a.get("base_url"),
})
return {
"play_type": "dash",
"video_streams": video_streams,
"audio_streams": audio_streams,
"current_quality": quality,
}
# Fallback to durl
durl = play_data.get("data", {}).get("durl", [])
urls = [{"url": d.get("url"), "size": d.get("size")} for d in durl]
return {
"play_type": "durl",
"urls": urls,
"current_quality": quality,
}
@staticmethod
def _parse_danmaku_xml(xml_text: str) -> List[Dict[str, Any]]:
"""Parse danmaku XML response.
Args:
xml_text: XML response text.
Returns:
List of danmaku entries.
"""
import re
danmaku_list = []
pattern = re.compile(r'<d p="([^"]+)">(.*?)</d>')
for match in pattern.finditer(xml_text):
params = match.group(1).split(",")
content = match.group(2)
if len(params) >= 8:
danmaku_list.append({
"time": float(params[0]),
"mode": int(params[1]), # 1=scroll, 4=bottom, 5=top
"font_size": int(params[2]),
"color": int(params[3]),
"timestamp": int(params[4]),
"pool": int(params[5]), # 0=normal, 1=subtitle, 2=special
"user_hash": params[6],
"dmid": params[7],
"content": content,
})
# Sort by time
danmaku_list.sort(key=lambda d: d["time"])
return danmaku_list
"""Bilibili video uploading and publishing module."""
import os
import json
import hashlib
import asyncio
from typing import Optional, Dict, Any, List
import httpx
from .auth import BilibiliAuth
from .utils import DEFAULT_HEADERS, API_BASE, ensure_dir
# Publishing API endpoints
PREUPLOAD_URL = "https://member.bilibili.com/preupload"
UPLOAD_URL = "https://upos-sz-upcdnbda2.bilivideo.com"
ADD_VIDEO_URL = f"{API_BASE}/x/vu/web/add"
EDIT_VIDEO_URL = f"{API_BASE}/x/vu/web/edit"
DELETE_VIDEO_URL = f"{API_BASE}/x/vu/web/archive/del"
DRAFT_ADD_URL = f"{API_BASE}/x/vu/web/draft/add"
COVER_UPLOAD_URL = f"{API_BASE}/x/vu/web/cover/up"
class BilibiliPublisher:
"""Publish videos to Bilibili.
Supports uploading videos, setting metadata, scheduling publications,
and managing drafts.
"""
def __init__(self, auth: BilibiliAuth):
"""Initialize BilibiliPublisher.
Args:
auth: BilibiliAuth instance (required for publishing).
Raises:
ValueError: If auth is not provided or not authenticated.
"""
if not auth or not auth.is_authenticated:
raise ValueError("Valid authentication is required for publishing")
self.auth = auth
def _get_client(self) -> httpx.AsyncClient:
"""Get an authenticated HTTP client."""
return self.auth.get_client()
async def upload(
self,
file_path: str,
title: str,
description: str = "",
tags: Optional[List[str]] = None,
category: str = "171",
cover_path: Optional[str] = None,
dynamic: str = "",
no_reprint: int = 1,
open_elec: int = 0,
) -> Dict[str, Any]:
"""Upload and publish a video to Bilibili.
Args:
file_path: Path to the video file.
title: Video title (max 80 chars).
description: Video description (max 2000 chars).
tags: List of tags (max 12 tags, each max 20 chars).
category: Category TID (default '171' for electronic gaming).
cover_path: Path to cover image (optional).
dynamic: Dynamic/feed text.
no_reprint: 1 = original, 0 = repost.
open_elec: 1 = enable charging, 0 = disable.
Returns:
Upload result with video info.
"""
if not os.path.exists(file_path):
return {"success": False, "message": f"File not found: {file_path}"}
# Validate inputs
if len(title) > 80:
return {"success": False, "message": "Title must be 80 characters or less"}
tags = tags or ["bilibili"]
if len(tags) > 12:
tags = tags[:12]
# Step 1: Pre-upload to get upload params
preupload_result = await self._preupload(file_path)
if not preupload_result.get("success"):
return preupload_result
# Step 2: Upload video file
upload_result = await self._upload_file(
file_path,
preupload_result,
)
if not upload_result.get("success"):
return upload_result
# Step 3: Upload cover if provided
cover_url = ""
if cover_path and os.path.exists(cover_path):
cover_result = await self._upload_cover(cover_path)
if cover_result.get("success"):
cover_url = cover_result.get("url", "")
# Step 4: Submit video
submit_result = await self._submit_video(
filename=upload_result["filename"],
title=title,
desc=description,
tags=tags,
tid=int(category),
cover=cover_url,
dynamic=dynamic,
no_reprint=no_reprint,
open_elec=open_elec,
)
return submit_result
async def draft(
self,
file_path: str,
title: str,
description: str = "",
tags: Optional[List[str]] = None,
category: str = "171",
cover_path: Optional[str] = None,
) -> Dict[str, Any]:
"""Save a video as draft.
Args:
file_path: Path to the video file.
title: Video title.
description: Video description.
tags: List of tags.
category: Category TID.
cover_path: Path to cover image.
Returns:
Draft save result.
"""
if not os.path.exists(file_path):
return {"success": False, "message": f"File not found: {file_path}"}
tags = tags or ["bilibili"]
# Upload video file
preupload_result = await self._preupload(file_path)
if not preupload_result.get("success"):
return preupload_result
upload_result = await self._upload_file(file_path, preupload_result)
if not upload_result.get("success"):
return upload_result
# Upload cover if provided
cover_url = ""
if cover_path and os.path.exists(cover_path):
cover_result = await self._upload_cover(cover_path)
if cover_result.get("success"):
cover_url = cover_result.get("url", "")
# Save as draft
async with self._get_client() as client:
resp = await client.post(
DRAFT_ADD_URL,
json={
"videos": [{
"filename": upload_result["filename"],
"title": title,
"desc": "",
}],
"title": title,
"desc": description,
"tag": ",".join(tags),
"tid": int(category),
"cover": cover_url,
"csrf": self.auth.csrf,
},
)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "Draft save failed")}
return {
"success": True,
"draft_id": data.get("data", {}).get("aid"),
"message": "Draft saved successfully",
}
async def schedule(
self,
file_path: str,
title: str,
schedule_time: str,
description: str = "",
tags: Optional[List[str]] = None,
category: str = "171",
cover_path: Optional[str] = None,
) -> Dict[str, Any]:
"""Schedule a video for future publication.
Args:
file_path: Path to the video file.
title: Video title.
schedule_time: Scheduled publish time (ISO 8601 format).
description: Video description.
tags: List of tags.
category: Category TID.
cover_path: Path to cover image.
Returns:
Schedule result.
"""
import datetime
if not os.path.exists(file_path):
return {"success": False, "message": f"File not found: {file_path}"}
# Parse schedule time
try:
dt = datetime.datetime.fromisoformat(schedule_time.replace("Z", "+00:00"))
timestamp = int(dt.timestamp())
except ValueError:
return {"success": False, "message": f"Invalid schedule time format: {schedule_time}"}
tags = tags or ["bilibili"]
# Upload video file
preupload_result = await self._preupload(file_path)
if not preupload_result.get("success"):
return preupload_result
upload_result = await self._upload_file(file_path, preupload_result)
if not upload_result.get("success"):
return upload_result
# Upload cover if provided
cover_url = ""
if cover_path and os.path.exists(cover_path):
cover_result = await self._upload_cover(cover_path)
if cover_result.get("success"):
cover_url = cover_result.get("url", "")
# Submit with schedule
submit_result = await self._submit_video(
filename=upload_result["filename"],
title=title,
desc=description,
tags=tags,
tid=int(category),
cover=cover_url,
dtime=timestamp,
)
if submit_result.get("success"):
submit_result["scheduled_time"] = schedule_time
return submit_result
async def edit(
self,
bvid: str,
title: Optional[str] = None,
description: Optional[str] = None,
tags: Optional[List[str]] = None,
cover_path: Optional[str] = None,
) -> Dict[str, Any]:
"""Edit an existing video's metadata.
Args:
bvid: BV number of the video to edit.
title: New title (if changing).
description: New description (if changing).
tags: New tags (if changing).
cover_path: New cover image path (if changing).
Returns:
Edit result.
"""
# First get current video info
async with self._get_client() as client:
resp = await client.get(
f"{API_BASE}/x/web-interface/view",
params={"bvid": bvid},
)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "Video not found")}
video = data["data"]
# Build edit payload
edit_data = {
"aid": video["aid"],
"title": title or video.get("title"),
"desc": description if description is not None else video.get("desc", ""),
"tag": ",".join(tags) if tags else ",".join(
t.get("tag_name", "") for t in video.get("tags", []) if t.get("tag_name")
),
"tid": video.get("tid"),
"csrf": self.auth.csrf,
}
# Upload new cover if provided
if cover_path and os.path.exists(cover_path):
cover_result = await self._upload_cover(cover_path)
if cover_result.get("success"):
edit_data["cover"] = cover_result.get("url", "")
async with self._get_client() as client:
resp = await client.post(EDIT_VIDEO_URL, json=edit_data)
result = resp.json()
if result.get("code") != 0:
return {"success": False, "message": result.get("message", "Edit failed")}
return {
"success": True,
"bvid": bvid,
"message": "Video edited successfully",
}
async def delete(self, bvid: str) -> Dict[str, Any]:
"""Delete a video.
Args:
bvid: BV number of the video to delete.
Returns:
Deletion result.
"""
# Get AID from BVID
async with self._get_client() as client:
resp = await client.get(
f"{API_BASE}/x/web-interface/view",
params={"bvid": bvid},
)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "Video not found")}
aid = data["data"]["aid"]
async with self._get_client() as client:
resp = await client.post(
DELETE_VIDEO_URL,
data={
"aid": aid,
"csrf": self.auth.csrf,
},
)
result = resp.json()
if result.get("code") != 0:
return {"success": False, "message": result.get("message", "Delete failed")}
return {
"success": True,
"bvid": bvid,
"aid": aid,
"message": "Video deleted successfully",
}
async def execute(self, action: str, **kwargs) -> Dict[str, Any]:
"""Execute a publisher action.
Args:
action: Action name ('upload', 'draft', 'schedule', 'edit', 'delete').
**kwargs: Additional parameters for the action.
Returns:
Action result dict.
"""
actions = {
"upload": self.upload,
"draft": self.draft,
"schedule": self.schedule,
"edit": self.edit,
"delete": self.delete,
}
handler = actions.get(action)
if not handler:
return {"success": False, "message": f"Unknown action: {action}"}
import inspect
sig = inspect.signature(handler)
valid_params = {k: v for k, v in kwargs.items() if k in sig.parameters}
return await handler(**valid_params)
async def _preupload(self, file_path: str) -> Dict[str, Any]:
"""Request pre-upload parameters from Bilibili.
Args:
file_path: Path to the video file.
Returns:
Pre-upload parameters dict.
"""
file_size = os.path.getsize(file_path)
file_name = os.path.basename(file_path)
async with self._get_client() as client:
resp = await client.get(
PREUPLOAD_URL,
params={
"name": file_name,
"size": file_size,
"r": "upos",
"profile": "ugcupos/bup",
"ssl": 0,
"version": "2.14.0",
"build": 2140000,
"upcdn": "bda2",
"probe_version": 20221109,
},
)
data = resp.json()
if "upos_uri" not in data:
return {"success": False, "message": "Pre-upload failed"}
return {
"success": True,
"upos_uri": data["upos_uri"],
"auth": data.get("auth"),
"biz_id": data.get("biz_id"),
"chunk_size": data.get("chunk_size", 4 * 1024 * 1024),
"endpoints": data.get("endpoints", []),
"file_size": file_size,
"file_name": file_name,
}
async def _upload_file(
self,
file_path: str,
preupload: Dict[str, Any],
) -> Dict[str, Any]:
"""Upload a video file in chunks.
Args:
file_path: Path to the video file.
preupload: Pre-upload parameters from _preupload().
Returns:
Upload result with filename.
"""
upos_uri = preupload["upos_uri"]
auth = preupload.get("auth")
biz_id = preupload.get("biz_id")
chunk_size = preupload.get("chunk_size", 4 * 1024 * 1024)
file_size = preupload["file_size"]
# Extract upos key from URI
upos_key = upos_uri.replace("upos://", "")
filename = upos_key.split("/")[-1].split(".")[0]
# Calculate chunk count
chunk_count = (file_size + chunk_size - 1) // chunk_size
# Init upload
upload_base = f"https://upos-sz-upcdnbda2.bilivideo.com/{upos_key}"
async with self._get_client() as client:
# Fetch upload ID
resp = await client.post(
upload_base,
params={
"uploads": "",
"output": "json",
},
headers={"X-Upos-Auth": auth} if auth else {},
)
try:
init_data = resp.json()
upload_id = init_data.get("upload_id", "")
except Exception:
return {"success": False, "message": "Failed to initialize upload"}
# Upload chunks
with open(file_path, "rb") as f:
for chunk_idx in range(chunk_count):
chunk_data = f.read(chunk_size)
start = chunk_idx * chunk_size
end = min(start + len(chunk_data), file_size)
async with self._get_client() as client:
resp = await client.put(
upload_base,
params={
"partNumber": chunk_idx + 1,
"uploadId": upload_id,
"chunk": chunk_idx,
"chunks": chunk_count,
"size": len(chunk_data),
"start": start,
"end": end,
"total": file_size,
},
headers={
"X-Upos-Auth": auth or "",
"Content-Type": "application/octet-stream",
},
content=chunk_data,
)
if resp.status_code not in (200, 202):
return {
"success": False,
"message": f"Upload chunk {chunk_idx + 1}/{chunk_count} failed",
}
# Complete upload
parts = [{"partNumber": i + 1, "eTag": "etag"} for i in range(chunk_count)]
async with self._get_client() as client:
resp = await client.post(
upload_base,
params={
"output": "json",
"name": preupload["file_name"],
"profile": "ugcupos/bup",
"uploadId": upload_id,
"biz_id": biz_id or 0,
},
json={"parts": parts},
headers={"X-Upos-Auth": auth or ""},
)
return {
"success": True,
"filename": filename,
"upos_uri": upos_uri,
}
async def _upload_cover(self, cover_path: str) -> Dict[str, Any]:
"""Upload a cover image.
Args:
cover_path: Path to the cover image.
Returns:
Upload result with cover URL.
"""
if not os.path.exists(cover_path):
return {"success": False, "message": f"Cover file not found: {cover_path}"}
with open(cover_path, "rb") as f:
cover_data = f.read()
# Detect MIME type
if cover_path.lower().endswith(".png"):
mime_type = "image/png"
elif cover_path.lower().endswith((".jpg", ".jpeg")):
mime_type = "image/jpeg"
else:
mime_type = "image/jpeg"
import base64
cover_b64 = f"data:{mime_type};base64,{base64.b64encode(cover_data).decode()}"
async with self._get_client() as client:
resp = await client.post(
COVER_UPLOAD_URL,
data={
"cover": cover_b64,
"csrf": self.auth.csrf,
},
)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "Cover upload failed")}
return {
"success": True,
"url": data.get("data", {}).get("url", ""),
}
async def _submit_video(
self,
filename: str,
title: str,
desc: str = "",
tags: Optional[List[str]] = None,
tid: int = 171,
cover: str = "",
dynamic: str = "",
no_reprint: int = 1,
open_elec: int = 0,
dtime: int = 0,
) -> Dict[str, Any]:
"""Submit a video for publishing.
Args:
filename: Uploaded video filename.
title: Video title.
desc: Description.
tags: Tags list.
tid: Category TID.
cover: Cover image URL.
dynamic: Dynamic text.
no_reprint: Original flag.
open_elec: Charging flag.
dtime: Scheduled publish timestamp (0 = immediate).
Returns:
Submit result.
"""
tags = tags or ["bilibili"]
payload = {
"videos": [{
"filename": filename,
"title": title,
"desc": "",
}],
"title": title,
"desc": desc,
"tag": ",".join(tags),
"tid": tid,
"cover": cover,
"dynamic": dynamic,
"copyright": 1 if no_reprint else 2,
"no_reprint": no_reprint,
"open_elec": open_elec,
"csrf": self.auth.csrf,
}
if dtime > 0:
payload["dtime"] = dtime
async with self._get_client() as client:
resp = await client.post(ADD_VIDEO_URL, json=payload)
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "Submit failed")}
result_data = data.get("data", {})
return {
"success": True,
"aid": result_data.get("aid"),
"bvid": result_data.get("bvid"),
"message": "Video published successfully",
}
"""Bilibili subtitle downloading and processing module."""
import json
import os
from typing import Optional, Dict, Any, List
import httpx
from .auth import BilibiliAuth
from .utils import (
API_VIDEO_INFO,
API_SUBTITLE,
DEFAULT_HEADERS,
extract_bvid,
ensure_dir,
sanitize_filename,
)
class SubtitleDownloader:
"""Download and process subtitles from Bilibili videos.
Supports multiple subtitle formats (SRT, ASS, VTT, TXT, JSON) and languages.
"""
def __init__(self, auth: Optional[BilibiliAuth] = None, output_dir: str = "./subtitles"):
"""Initialize SubtitleDownloader.
Args:
auth: Optional BilibiliAuth instance for authenticated requests.
output_dir: Default output directory for subtitle files.
"""
self.auth = auth
self.output_dir = output_dir
def _get_client(self) -> httpx.AsyncClient:
"""Get an HTTP client, using auth if available."""
if self.auth:
return self.auth.get_client()
return httpx.AsyncClient(
headers=DEFAULT_HEADERS,
timeout=30.0,
follow_redirects=True,
)
async def list_subtitles(self, url: str) -> Dict[str, Any]:
"""List available subtitles for a video.
Args:
url: Bilibili video URL or BV number.
Returns:
List of available subtitles with language info.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
# Get video info to get cid
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
video = data["data"]
cid = video["pages"][0]["cid"]
title = video.get("title", bvid)
# Get subtitle info
async with self._get_client() as client:
resp = await client.get(
API_SUBTITLE,
params={"bvid": bvid, "cid": cid},
)
sub_data = resp.json()
if sub_data.get("code") != 0:
return {"success": False, "message": sub_data.get("message", "API error")}
subtitles_info = sub_data.get("data", {}).get("subtitle", {})
subtitles = []
for sub in subtitles_info.get("subtitles", []):
subtitles.append({
"id": sub.get("id"),
"language": sub.get("lan"),
"language_name": sub.get("lan_doc"),
"url": sub.get("subtitle_url"),
"ai_type": sub.get("ai_type", 0),
"ai_status": sub.get("ai_status", 0),
})
return {
"success": True,
"bvid": bvid,
"title": title,
"cid": cid,
"subtitles": subtitles,
"count": len(subtitles),
}
async def download(
self,
url: str,
language: str = "zh-CN",
format: str = "srt",
output_dir: Optional[str] = None,
) -> Dict[str, Any]:
"""Download subtitles for a video.
Args:
url: Bilibili video URL or BV number.
language: Subtitle language code (e.g., 'zh-CN', 'en', 'ja').
format: Output format ('srt', 'ass', 'vtt', 'txt', 'json').
output_dir: Output directory.
Returns:
Download result with file path.
"""
out_dir = ensure_dir(output_dir or self.output_dir)
# List available subtitles
sub_list = await self.list_subtitles(url)
if not sub_list.get("success"):
return sub_list
# Find matching subtitle
target_sub = None
for sub in sub_list.get("subtitles", []):
if sub["language"] == language or sub["language"].startswith(language.split("-")[0]):
target_sub = sub
break
if not target_sub:
available = [s["language"] for s in sub_list.get("subtitles", [])]
return {
"success": False,
"message": f"Subtitle for language '{language}' not found. Available: {available}",
}
# Download subtitle JSON
sub_url = target_sub["url"]
if sub_url.startswith("//"):
sub_url = "https:" + sub_url
async with self._get_client() as client:
resp = await client.get(sub_url)
sub_data = resp.json()
# Convert and save
title = sanitize_filename(sub_list.get("title", "subtitle"))
filename = f"{title}_{language}.{format}"
filepath = os.path.join(out_dir, filename)
body = sub_data.get("body", [])
converters = {
"srt": self._to_srt,
"ass": self._to_ass,
"vtt": self._to_vtt,
"txt": self._to_txt,
"json": self._to_json,
}
converter = converters.get(format, self._to_srt)
content = converter(body, title)
with open(filepath, "w", encoding="utf-8") as f:
f.write(content)
return {
"success": True,
"bvid": sub_list.get("bvid"),
"title": sub_list.get("title"),
"language": language,
"format": format,
"filepath": filepath,
"entries": len(body),
}
async def convert(
self,
input_path: str,
output_format: str,
output_dir: Optional[str] = None,
) -> Dict[str, Any]:
"""Convert a subtitle file to a different format.
Args:
input_path: Path to the input subtitle file.
output_format: Target format ('srt', 'ass', 'vtt', 'txt').
output_dir: Output directory (defaults to same as input).
Returns:
Conversion result.
"""
if not os.path.exists(input_path):
return {"success": False, "message": f"File not found: {input_path}"}
with open(input_path, "r", encoding="utf-8") as f:
content = f.read()
# Try to detect input format and parse
body = self._parse_subtitle(content, input_path)
if body is None:
return {"success": False, "message": "Cannot parse input subtitle file"}
out_dir = output_dir or os.path.dirname(input_path)
base_name = os.path.splitext(os.path.basename(input_path))[0]
output_path = os.path.join(out_dir, f"{base_name}.{output_format}")
converters = {
"srt": self._to_srt,
"ass": self._to_ass,
"vtt": self._to_vtt,
"txt": self._to_txt,
"json": self._to_json,
}
converter = converters.get(output_format, self._to_srt)
output_content = converter(body, base_name)
with open(output_path, "w", encoding="utf-8") as f:
f.write(output_content)
return {
"success": True,
"input": input_path,
"output": output_path,
"format": output_format,
"entries": len(body),
}
async def merge(
self,
input_paths: List[str],
output_path: str,
output_format: str = "srt",
) -> Dict[str, Any]:
"""Merge multiple subtitle files into one.
Args:
input_paths: List of input subtitle file paths.
output_path: Output file path.
output_format: Output format.
Returns:
Merge result.
"""
all_body = []
time_offset = 0.0
for path in input_paths:
if not os.path.exists(path):
return {"success": False, "message": f"File not found: {path}"}
with open(path, "r", encoding="utf-8") as f:
content = f.read()
body = self._parse_subtitle(content, path)
if body is None:
return {"success": False, "message": f"Cannot parse: {path}"}
# Offset timestamps
for entry in body:
entry["from"] = entry.get("from", 0) + time_offset
entry["to"] = entry.get("to", 0) + time_offset
if body:
time_offset = body[-1].get("to", 0) + 0.5
all_body.extend(body)
converters = {
"srt": self._to_srt,
"ass": self._to_ass,
"vtt": self._to_vtt,
"txt": self._to_txt,
"json": self._to_json,
}
converter = converters.get(output_format, self._to_srt)
output_content = converter(all_body, "merged")
ensure_dir(os.path.dirname(output_path) or ".")
with open(output_path, "w", encoding="utf-8") as f:
f.write(output_content)
return {
"success": True,
"output": output_path,
"format": output_format,
"total_entries": len(all_body),
"merged_files": len(input_paths),
}
async def execute(self, action: str, **kwargs) -> Dict[str, Any]:
"""Execute a subtitle action.
Args:
action: Action name ('download', 'list', 'convert', 'merge').
**kwargs: Additional parameters for the action.
Returns:
Action result dict.
"""
actions = {
"download": self.download,
"list": self.list_subtitles,
"convert": self.convert,
"merge": self.merge,
}
handler = actions.get(action)
if not handler:
return {"success": False, "message": f"Unknown action: {action}"}
import inspect
sig = inspect.signature(handler)
valid_params = {k: v for k, v in kwargs.items() if k in sig.parameters}
return await handler(**valid_params)
# --- Format converters ---
@staticmethod
def _format_time_srt(seconds: float) -> str:
"""Format seconds to SRT timestamp (HH:MM:SS,mmm)."""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
millis = int((seconds % 1) * 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"
@staticmethod
def _format_time_vtt(seconds: float) -> str:
"""Format seconds to VTT timestamp (HH:MM:SS.mmm)."""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
millis = int((seconds % 1) * 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}"
@staticmethod
def _format_time_ass(seconds: float) -> str:
"""Format seconds to ASS timestamp (H:MM:SS.cc)."""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
centis = int((seconds % 1) * 100)
return f"{hours}:{minutes:02d}:{secs:02d}.{centis:02d}"
@classmethod
def _to_srt(cls, body: List[Dict], title: str = "") -> str:
"""Convert subtitle body to SRT format."""
lines = []
for i, entry in enumerate(body, 1):
start = cls._format_time_srt(entry.get("from", 0))
end = cls._format_time_srt(entry.get("to", 0))
content = entry.get("content", "")
lines.append(f"{i}\n{start} --> {end}\n{content}\n")
return "\n".join(lines)
@classmethod
def _to_vtt(cls, body: List[Dict], title: str = "") -> str:
"""Convert subtitle body to WebVTT format."""
lines = ["WEBVTT", ""]
for i, entry in enumerate(body, 1):
start = cls._format_time_vtt(entry.get("from", 0))
end = cls._format_time_vtt(entry.get("to", 0))
content = entry.get("content", "")
lines.append(f"{i}\n{start} --> {end}\n{content}\n")
return "\n".join(lines)
@classmethod
def _to_ass(cls, body: List[Dict], title: str = "") -> str:
"""Convert subtitle body to ASS format."""
header = f"""[Script Info]
Title: {title}
ScriptType: v4.00+
Collisions: Normal
PlayDepth: 0
[V4+ Styles]
Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding
Style: Default,Arial,20,&H00FFFFFF,&H0000FFFF,&H00000000,&H00000000,0,0,0,0,100,100,0,0,1,2,2,2,10,10,10,1
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
lines = [header]
for entry in body:
start = cls._format_time_ass(entry.get("from", 0))
end = cls._format_time_ass(entry.get("to", 0))
content = entry.get("content", "").replace("\n", "\\N")
lines.append(f"Dialogue: 0,{start},{end},Default,,0,0,0,,{content}")
return "\n".join(lines)
@staticmethod
def _to_txt(body: List[Dict], title: str = "") -> str:
"""Convert subtitle body to plain text."""
lines = []
for entry in body:
content = entry.get("content", "")
if content.strip():
lines.append(content)
return "\n".join(lines)
@staticmethod
def _to_json(body: List[Dict], title: str = "") -> str:
"""Convert subtitle body to JSON format."""
return json.dumps(
{"title": title, "body": body},
ensure_ascii=False,
indent=2,
)
@staticmethod
def _parse_subtitle(content: str, filepath: str) -> Optional[List[Dict]]:
"""Parse a subtitle file into internal body format.
Args:
content: File content.
filepath: File path (used to detect format).
Returns:
List of subtitle entries or None.
"""
ext = os.path.splitext(filepath)[1].lower()
if ext == ".json":
try:
data = json.loads(content)
if isinstance(data, dict) and "body" in data:
return data["body"]
if isinstance(data, list):
return data
except json.JSONDecodeError:
return None
if ext == ".srt":
return SubtitleDownloader._parse_srt(content)
if ext == ".vtt":
# Remove WEBVTT header
content = content.replace("WEBVTT", "").strip()
return SubtitleDownloader._parse_srt(content)
if ext == ".txt":
lines = content.strip().split("\n")
body = []
for i, line in enumerate(lines):
if line.strip():
body.append({
"from": i * 3.0,
"to": (i + 1) * 3.0,
"content": line.strip(),
})
return body
return None
@staticmethod
def _parse_srt(content: str) -> List[Dict]:
"""Parse SRT format content."""
import re
body = []
blocks = re.split(r"\n\s*\n", content.strip())
for block in blocks:
lines = block.strip().split("\n")
if len(lines) < 3:
continue
time_match = re.match(
r"(\d{2}:\d{2}:\d{2}[,\.]\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2}[,\.]\d{3})",
lines[1],
)
if not time_match:
continue
start_str = time_match.group(1).replace(",", ".")
end_str = time_match.group(2).replace(",", ".")
def parse_ts(ts: str) -> float:
parts = ts.split(":")
return float(parts[0]) * 3600 + float(parts[1]) * 60 + float(parts[2])
body.append({
"from": parse_ts(start_str),
"to": parse_ts(end_str),
"content": "\n".join(lines[2:]),
})
return body
"""Shared utility functions for the Bilibili All-in-One skill."""
import re
import os
import json
import time
import hashlib
from typing import Optional, Dict, Any
from urllib.parse import urlparse, parse_qs
# Bilibili API endpoints
API_BASE = "https://api.bilibili.com"
API_HOT = f"{API_BASE}/x/web-interface/popular"
API_TRENDING = f"{API_BASE}/x/web-interface/popular/series/list"
API_WEEKLY = f"{API_BASE}/x/web-interface/popular/series/one"
API_RANK = f"{API_BASE}/x/web-interface/ranking/v2"
API_VIDEO_INFO = f"{API_BASE}/x/web-interface/view"
API_VIDEO_DETAIL = f"{API_BASE}/x/web-interface/view/detail"
API_PLAY_URL = f"{API_BASE}/x/player/playurl"
API_DANMAKU = f"{API_BASE}/x/v1/dm/list.so"
API_SUBTITLE = f"{API_BASE}/x/player/v2"
API_STAT = f"{API_BASE}/x/relation/stat"
API_SEARCH = f"{API_BASE}/x/web-interface/search/type"
# Video quality mapping
QUALITY_MAP = {
"360p": 16,
"480p": 32,
"720p": 64,
"1080p": 80,
"1080p+": 112,
"4k": 120,
}
# Category TID mapping
CATEGORY_TID = {
"all": 0,
"anime": 1,
"music": 3,
"dance": 129,
"game": 4,
"tech": 188,
"life": 160,
"food": 211,
"car": 223,
"fashion": 155,
"entertainment": 5,
"movie": 23,
"tv": 11,
}
# Default headers
DEFAULT_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Referer": "https://www.bilibili.com",
"Origin": "https://www.bilibili.com",
}
def extract_bvid(url_or_bvid: str) -> Optional[str]:
"""Extract BV ID from a Bilibili URL or return it directly if already a BV ID.
Args:
url_or_bvid: Bilibili video URL or BV number.
Returns:
The extracted BV ID, or None if extraction fails.
"""
if not url_or_bvid:
return None
# Already a BV ID
bv_match = re.match(r"^(BV[a-zA-Z0-9]+)$", url_or_bvid.strip())
if bv_match:
return bv_match.group(1)
# Extract from URL
patterns = [
r"bilibili\.com/video/(BV[a-zA-Z0-9]+)",
r"b23\.tv/(BV[a-zA-Z0-9]+)",
r"bilibili\.com/bangumi/play/(BV[a-zA-Z0-9]+)",
]
for pattern in patterns:
match = re.search(pattern, url_or_bvid)
if match:
return match.group(1)
return None
def extract_aid(url_or_aid: str) -> Optional[int]:
"""Extract AV ID from a Bilibili URL or return it directly.
Args:
url_or_aid: Bilibili video URL or AV number.
Returns:
The extracted AV ID as integer, or None if extraction fails.
"""
if not url_or_aid:
return None
# Already an AV ID
av_match = re.match(r"^av(\d+)$", url_or_aid.strip(), re.IGNORECASE)
if av_match:
return int(av_match.group(1))
# Extract from URL
match = re.search(r"bilibili\.com/video/av(\d+)", url_or_aid)
if match:
return int(match.group(1))
return None
def format_duration(seconds: int) -> str:
"""Format duration in seconds to a human-readable string.
Args:
seconds: Duration in seconds.
Returns:
Formatted duration string (e.g., '1:23:45' or '12:34').
"""
hours = seconds // 3600
minutes = (seconds % 3600) // 60
secs = seconds % 60
if hours > 0:
return f"{hours}:{minutes:02d}:{secs:02d}"
return f"{minutes}:{secs:02d}"
def format_number(num: int) -> str:
"""Format a large number to a human-readable string.
Args:
num: The number to format.
Returns:
Formatted number string (e.g., '1.2万', '3.4亿').
"""
if num >= 100_000_000:
return f"{num / 100_000_000:.1f}亿"
if num >= 10_000:
return f"{num / 10_000:.1f}万"
return str(num)
def ensure_dir(directory: str) -> str:
"""Ensure a directory exists, creating it if necessary.
Args:
directory: Path to the directory.
Returns:
The absolute path to the directory.
"""
abs_path = os.path.abspath(directory)
os.makedirs(abs_path, exist_ok=True)
return abs_path
def sanitize_filename(filename: str) -> str:
"""Sanitize a string for use as a filename.
Args:
filename: The original filename.
Returns:
Sanitized filename safe for all operating systems.
"""
# Remove or replace invalid characters
invalid_chars = r'[<>:"/\\|?*\x00-\x1f]'
sanitized = re.sub(invalid_chars, "_", filename)
# Remove trailing dots and spaces
sanitized = sanitized.strip(". ")
# Limit length
if len(sanitized) > 200:
sanitized = sanitized[:200]
return sanitized or "untitled"
def generate_wbi_sign(params: Dict[str, Any], img_key: str, sub_key: str) -> Dict[str, Any]:
"""Generate WBI signature for Bilibili API requests.
Args:
params: Request parameters.
img_key: WBI img key.
sub_key: WBI sub key.
Returns:
Parameters dict with wts and w_rid added.
"""
mixin_key_enc_tab = [
46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35,
27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13,
37, 48, 7, 16, 24, 55, 40, 61, 26, 17, 0, 1, 60, 51, 30, 4,
22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, 36, 20, 34, 44, 52,
]
raw_key = img_key + sub_key
mixin_key = "".join(raw_key[i] for i in mixin_key_enc_tab)[:32]
params["wts"] = int(time.time())
# Sort parameters
sorted_params = dict(sorted(params.items()))
query = "&".join(f"{k}={v}" for k, v in sorted_params.items())
wbi_sign = hashlib.md5((query + mixin_key).encode()).hexdigest()
params["w_rid"] = wbi_sign
return params
def parse_video_url(url: str) -> Dict[str, Any]:
"""Parse a video URL and extract platform and identifier.
Args:
url: Video URL (supports Bilibili and YouTube).
Returns:
Dict with 'platform' and 'id' keys.
"""
parsed = urlparse(url)
# Bilibili
if "bilibili.com" in parsed.hostname or "b23.tv" in parsed.hostname:
bvid = extract_bvid(url)
aid = extract_aid(url)
return {
"platform": "bilibili",
"bvid": bvid,
"aid": aid,
"url": url,
}
# YouTube
if "youtube.com" in parsed.hostname or "youtu.be" in parsed.hostname:
if "youtu.be" in parsed.hostname:
video_id = parsed.path.strip("/")
else:
qs = parse_qs(parsed.query)
video_id = qs.get("v", [None])[0]
return {
"platform": "youtube",
"video_id": video_id,
"url": url,
}
return {"platform": "unknown", "url": url}
"""Bilibili/YouTube video watching and stats tracking module."""
import asyncio
import time
from typing import Optional, Dict, Any, List
import httpx
from .auth import BilibiliAuth
from .utils import (
API_VIDEO_INFO,
API_VIDEO_DETAIL,
DEFAULT_HEADERS,
extract_bvid,
format_number,
format_duration,
parse_video_url,
)
class BilibiliWatcher:
"""Watch and monitor Bilibili (and YouTube) videos.
Track view counts, comments, likes, and other engagement metrics over time.
"""
def __init__(self, auth: Optional[BilibiliAuth] = None):
"""Initialize BilibiliWatcher.
Args:
auth: Optional BilibiliAuth instance for authenticated requests.
"""
self.auth = auth
self._tracking_data: Dict[str, List[Dict]] = {}
def _get_client(self) -> httpx.AsyncClient:
"""Get an HTTP client, using auth if available."""
if self.auth:
return self.auth.get_client()
return httpx.AsyncClient(
headers=DEFAULT_HEADERS,
timeout=30.0,
follow_redirects=True,
)
async def watch(self, url: str) -> Dict[str, Any]:
"""Get detailed video information for watching.
Args:
url: Video URL (supports Bilibili and YouTube).
Returns:
Detailed video information.
"""
video_info = parse_video_url(url)
if video_info["platform"] == "bilibili":
return await self._watch_bilibili(video_info.get("bvid"), url)
elif video_info["platform"] == "youtube":
return await self._watch_youtube(video_info.get("video_id"), url)
else:
return {"success": False, "message": f"Unsupported platform for URL: {url}"}
async def _watch_bilibili(self, bvid: Optional[str], url: str) -> Dict[str, Any]:
"""Get detailed Bilibili video information.
Args:
bvid: BV ID of the video.
url: Original URL.
Returns:
Video details dict.
"""
if not bvid:
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Cannot extract BV ID from: {url}"}
async with self._get_client() as client:
resp = await client.get(API_VIDEO_DETAIL, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
video = data["data"].get("View", {})
stat = video.get("stat", {})
owner = video.get("owner", {})
tags = [t.get("tag_name") for t in data["data"].get("Tags", []) if t.get("tag_name")]
related = []
for r in data["data"].get("Related", [])[:5]:
related.append({
"bvid": r.get("bvid"),
"title": r.get("title"),
"author": r.get("owner", {}).get("name"),
})
return {
"success": True,
"platform": "bilibili",
"bvid": video.get("bvid"),
"aid": video.get("aid"),
"title": video.get("title"),
"description": video.get("desc"),
"cover": video.get("pic"),
"duration": format_duration(video.get("duration", 0)),
"author": {
"mid": owner.get("mid"),
"name": owner.get("name"),
"face": owner.get("face"),
},
"stats": {
"views": stat.get("view", 0),
"views_formatted": format_number(stat.get("view", 0)),
"danmaku": stat.get("danmaku", 0),
"likes": stat.get("like", 0),
"coins": stat.get("coin", 0),
"favorites": stat.get("favorite", 0),
"shares": stat.get("share", 0),
"comments": stat.get("reply", 0),
},
"tags": tags,
"related_videos": related,
"url": f"https://www.bilibili.com/video/{video.get('bvid')}",
}
async def _watch_youtube(self, video_id: Optional[str], url: str) -> Dict[str, Any]:
"""Get YouTube video information via oEmbed API.
Args:
video_id: YouTube video ID.
url: Original URL.
Returns:
Video details dict.
"""
if not video_id:
return {"success": False, "message": f"Cannot extract video ID from: {url}"}
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.get(
"https://www.youtube.com/oembed",
params={"url": f"https://www.youtube.com/watch?v={video_id}", "format": "json"},
)
if resp.status_code != 200:
return {"success": False, "message": "Failed to fetch YouTube video info"}
data = resp.json()
return {
"success": True,
"platform": "youtube",
"video_id": video_id,
"title": data.get("title"),
"author": data.get("author_name"),
"author_url": data.get("author_url"),
"thumbnail": data.get("thumbnail_url"),
"url": f"https://www.youtube.com/watch?v={video_id}",
}
async def get_stats(self, url: str) -> Dict[str, Any]:
"""Get current engagement statistics for a video.
Args:
url: Video URL.
Returns:
Current engagement statistics.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
async with self._get_client() as client:
resp = await client.get(API_VIDEO_INFO, params={"bvid": bvid})
data = resp.json()
if data.get("code") != 0:
return {"success": False, "message": data.get("message", "API error")}
stat = data["data"].get("stat", {})
return {
"success": True,
"bvid": bvid,
"title": data["data"].get("title"),
"timestamp": int(time.time()),
"stats": {
"views": stat.get("view", 0),
"danmaku": stat.get("danmaku", 0),
"likes": stat.get("like", 0),
"coins": stat.get("coin", 0),
"favorites": stat.get("favorite", 0),
"shares": stat.get("share", 0),
"comments": stat.get("reply", 0),
},
}
async def track(
self,
url: str,
interval: int = 60,
duration: int = 24,
callback=None,
) -> Dict[str, Any]:
"""Track video metrics over time.
Args:
url: Video URL.
interval: Tracking interval in minutes.
duration: Total tracking duration in hours.
callback: Optional async callback function called with each data point.
Returns:
Tracking summary with all collected data points.
"""
bvid = extract_bvid(url)
if not bvid:
return {"success": False, "message": f"Invalid URL or BV number: {url}"}
data_points = []
end_time = time.time() + duration * 3600
interval_seconds = interval * 60
while time.time() < end_time:
stats = await self.get_stats(url)
if stats.get("success"):
data_points.append(stats)
if callback:
await callback(stats)
self._tracking_data[bvid] = data_points
remaining = end_time - time.time()
if remaining <= 0:
break
await asyncio.sleep(min(interval_seconds, remaining))
# Calculate changes
summary = self._calculate_changes(data_points)
return {
"success": True,
"bvid": bvid,
"data_points": len(data_points),
"duration_hours": duration,
"interval_minutes": interval,
"summary": summary,
"data": data_points,
}
async def compare(self, urls: List[str]) -> Dict[str, Any]:
"""Compare engagement metrics of multiple videos.
Args:
urls: List of video URLs to compare.
Returns:
Comparison results.
"""
results = []
tasks = [self.get_stats(url) for url in urls]
stats_list = await asyncio.gather(*tasks, return_exceptions=True)
for url, stats in zip(urls, stats_list):
if isinstance(stats, Exception):
results.append({"url": url, "success": False, "message": str(stats)})
else:
results.append(stats)
# Rank by views
successful = [r for r in results if r.get("success")]
successful.sort(key=lambda r: r.get("stats", {}).get("views", 0), reverse=True)
return {
"success": True,
"total": len(urls),
"compared": len(successful),
"ranking": [
{
"rank": i + 1,
"bvid": r.get("bvid"),
"title": r.get("title"),
"views": r.get("stats", {}).get("views", 0),
"likes": r.get("stats", {}).get("likes", 0),
}
for i, r in enumerate(successful)
],
"results": results,
}
async def execute(self, action: str, **kwargs) -> Dict[str, Any]:
"""Execute a watcher action.
Args:
action: Action name ('watch', 'get_stats', 'track', 'compare').
**kwargs: Additional parameters for the action.
Returns:
Action result dict.
"""
actions = {
"watch": self.watch,
"get_stats": self.get_stats,
"track": self.track,
"compare": self.compare,
}
handler = actions.get(action)
if not handler:
return {"success": False, "message": f"Unknown action: {action}"}
import inspect
sig = inspect.signature(handler)
valid_params = {k: v for k, v in kwargs.items() if k in sig.parameters}
return await handler(**valid_params)
@staticmethod
def _calculate_changes(data_points: List[Dict]) -> Dict[str, Any]:
"""Calculate metric changes from data points.
Args:
data_points: List of stat snapshots.
Returns:
Summary of changes.
"""
if len(data_points) < 2:
return {"message": "Not enough data points for comparison"}
first = data_points[0].get("stats", {})
last = data_points[-1].get("stats", {})
changes = {}
for key in first:
if isinstance(first.get(key), (int, float)):
changes[key] = {
"start": first[key],
"end": last[key],
"change": last[key] - first[key],
"change_percent": round(
((last[key] - first[key]) / first[key] * 100) if first[key] > 0 else 0, 2
),
}
return changes
# Tests package
"""Test all skill examples from skill.md.
This test module covers every example listed in the skill.md OpenClaw section.
All HTTP requests are mocked so no real API calls are made.
"""
import asyncio
import json
import os
import tempfile
import unittest
from unittest.mock import AsyncMock, MagicMock, patch, mock_open
import httpx
# ── Helpers ──────────────────────────────────────────────────────────
def _run(coro):
"""Run an async coroutine synchronously."""
return asyncio.run(coro)
def _mock_response(data: dict, status_code: int = 200) -> httpx.Response:
"""Create a mock httpx.Response."""
resp = MagicMock(spec=httpx.Response)
resp.status_code = status_code
resp.json.return_value = data
resp.text = json.dumps(data)
resp.content = json.dumps(data).encode()
return resp
def _bilibili_ok(data: dict) -> dict:
"""Wrap data in standard Bilibili API success envelope."""
return {"code": 0, "message": "0", "data": data}
def _video_info_payload(bvid="BV1xx411c7mD", title="Test Video", pages=1):
"""Generate a standard video info API response payload."""
page_list = []
for i in range(1, pages + 1):
page_list.append({
"cid": 10000 + i,
"page": i,
"part": f"Part {i}",
"duration": 300,
})
return {
"bvid": bvid,
"aid": 12345,
"title": title,
"desc": "Test description",
"pic": "https://example.com/cover.jpg",
"duration": 300,
"owner": {"mid": 100, "name": "TestUser", "face": "https://example.com/face.jpg"},
"stat": {
"view": 100000, "danmaku": 500, "like": 5000,
"coin": 1000, "favorite": 2000, "share": 300, "reply": 800,
},
"pages": page_list,
"pubdate": 1700000000,
"tid": 171,
"tags": [{"tag_name": "test"}, {"tag_name": "bilibili"}],
"subtitle": {
"list": [
{
"lan": "zh-CN",
"lan_doc": "中文（中国）",
"subtitle_url": "//example.com/subtitle_zh.json",
},
{
"lan": "en",
"lan_doc": "English",
"subtitle_url": "//example.com/subtitle_en.json",
},
]
},
}
# ── Mock Client Context Manager ─────────────────────────────────────
class MockAsyncClient:
"""A mock async HTTP client that returns pre-configured responses."""
def __init__(self, responses=None):
"""
Args:
responses: list of httpx.Response mocks. Each call to get/post
pops the next response from the front.
"""
self._responses = list(responses or [])
self._call_index = 0
def _next_response(self):
if self._call_index < len(self._responses):
resp = self._responses[self._call_index]
self._call_index += 1
return resp
# Default: return a generic success
return _mock_response(_bilibili_ok({}))
async def get(self, *args, **kwargs):
return self._next_response()
async def post(self, *args, **kwargs):
return self._next_response()
async def __aenter__(self):
return self
async def __aexit__(self, *args):
pass
def _patch_client(module_obj, responses):
"""Patch _get_client on a module instance to return a MockAsyncClient."""
mock_client = MockAsyncClient(responses)
module_obj._get_client = lambda: mock_client
return mock_client
# ══════════════════════════════════════════════════════════════════════
# Test cases — one per skill.md example
# ══════════════════════════════════════════════════════════════════════
class TestHotMonitorExamples(unittest.TestCase):
"""Examples 1, 7, 15 from skill.md — Hot Monitor module."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne()
# ── Example 1: get_hot ───────────────────────────────────────────
def test_example_01_get_hot(self):
"""用户: 帮我看看B站现在有什么热门视频
Agent → bilibili_hot_monitor(action="get_hot", limit=10)
"""
hot_data = {
"list": [
{
"bvid": f"BV{i}xxx",
"aid": 1000 + i,
"title": f"Hot Video {i}",
"desc": "",
"pic": "",
"duration": 200,
"owner": {"mid": i, "name": f"User{i}", "face": ""},
"stat": {"view": 10000 * i, "danmaku": 100, "like": 500,
"coin": 100, "favorite": 200, "share": 50, "reply": 80},
"pubdate": 1700000000,
}
for i in range(1, 11)
],
"no_more": False,
}
resp = _mock_response(_bilibili_ok(hot_data))
_patch_client(self.app.hot_monitor, [resp])
result = _run(self.app.execute("hot_monitor", "get_hot", page_size=10))
self.assertTrue(result["success"])
self.assertEqual(len(result["videos"]), 10)
# ── Example 7: get_rank (game) ───────────────────────────────────
def test_example_07_get_rank_game(self):
"""用户: B站游戏区排行榜前10名是什么？
Agent → bilibili_hot_monitor(action="get_rank", category="game", limit=10)
"""
rank_data = {
"list": [
{
"bvid": f"BV_rank_{i}",
"aid": 2000 + i,
"title": f"Game Rank {i}",
"desc": "",
"pic": "",
"duration": 600,
"score": 10000 - i * 100,
"owner": {"mid": i, "name": f"Gamer{i}", "face": ""},
"stat": {"view": 50000, "danmaku": 200, "like": 3000,
"coin": 500, "favorite": 1000, "share": 100, "reply": 400},
"pubdate": 1700000000,
}
for i in range(1, 15)
],
}
resp = _mock_response(_bilibili_ok(rank_data))
_patch_client(self.app.hot_monitor, [resp])
result = _run(self.app.execute("hot_monitor", "get_rank", category="game", limit=10))
self.assertTrue(result["success"])
self.assertEqual(result["category"], "game")
self.assertLessEqual(len(result["videos"]), 10)
# ── Example 15: get_weekly ───────────────────────────────────────
def test_example_15_get_weekly(self):
"""用户: 本周B站必看榜单有什么？
Agent → bilibili_hot_monitor(action="get_weekly")
"""
weekly_data = {
"config": {"number": 200, "subject": "本周必看", "label": "第200期"},
"list": [
{
"bvid": f"BV_weekly_{i}",
"aid": 3000 + i,
"title": f"Weekly Must Watch {i}",
"desc": "",
"pic": "",
"duration": 400,
"owner": {"mid": i, "name": f"Creator{i}", "face": ""},
"stat": {"view": 200000, "danmaku": 800, "like": 10000,
"coin": 3000, "favorite": 5000, "share": 500, "reply": 2000},
"pubdate": 1700000000,
}
for i in range(1, 6)
],
}
resp = _mock_response(_bilibili_ok(weekly_data))
_patch_client(self.app.hot_monitor, [resp])
result = _run(self.app.execute("hot_monitor", "get_weekly"))
self.assertTrue(result["success"])
self.assertEqual(result["week_number"], 200)
self.assertGreater(len(result["videos"]), 0)
class TestDownloaderExamples(unittest.TestCase):
"""Examples 2, 12, 16 from skill.md — Downloader module."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne()
# ── Example 2: download 1080p mp4 ────────────────────────────────
def test_example_02_download_video(self):
"""用户: 下载这个B站视频 BV1xx411c7mD，要1080p的MP4格式
Agent → bilibili_downloader(action="download", url="BV1xx411c7mD", quality="1080p", format="mp4")
"""
info_resp = _mock_response(_bilibili_ok(_video_info_payload()))
play_resp = _mock_response(_bilibili_ok({
"dash": {
"video": [{"id": 80, "baseUrl": "https://example.com/video.m4s",
"bandwidth": 2000000, "codecs": "avc1"}],
"audio": [{"id": 30280, "baseUrl": "https://example.com/audio.m4s",
"bandwidth": 128000, "codecs": "mp4a"}],
}
}))
_patch_client(self.app.downloader, [info_resp, play_resp])
# Mock the download stream to create dummy files
async def fake_download(url, filepath):
with open(filepath, "wb") as f:
f.write(b"\x00" * 100)
async def fake_merge(video_path, audio_path, output_path):
with open(output_path, "wb") as f:
f.write(b"\x00" * 200)
return True
self.app.downloader._download_stream = fake_download
self.app.downloader._merge_streams = fake_merge
with tempfile.TemporaryDirectory() as tmpdir:
result = _run(self.app.execute(
"downloader", "download",
url="BV1xx411c7mD", quality="1080p", format="mp4",
output_dir=tmpdir,
))
# The download itself won't create real files, so it may not report success
# but the action dispatch and parameter handling should work
self.assertIn("success", result)
# ── Example 12: download audio (mp3) ─────────────────────────────
def test_example_12_download_audio(self):
"""用户: 提取这个B站视频的音频
Agent → bilibili_downloader(action="download", url="BV1xx411c7mD", format="mp3")
"""
info_resp = _mock_response(_bilibili_ok(_video_info_payload()))
play_resp = _mock_response(_bilibili_ok({
"dash": {
"video": [{"id": 80, "baseUrl": "https://example.com/video.m4s",
"bandwidth": 2000000, "codecs": "avc1"}],
"audio": [{"id": 30280, "baseUrl": "https://example.com/audio.m4s",
"bandwidth": 128000, "codecs": "mp4a"}],
}
}))
_patch_client(self.app.downloader, [info_resp, play_resp])
async def fake_download(url, filepath):
with open(filepath, "wb") as f:
f.write(b"\x00" * 100)
self.app.downloader._download_stream = fake_download
with tempfile.TemporaryDirectory() as tmpdir:
result = _run(self.app.execute(
"downloader", "download",
url="BV1xx411c7mD", format="mp3",
output_dir=tmpdir,
))
self.assertIn("success", result)
# ── Example 16: batch_download ───────────────────────────────────
def test_example_16_batch_download(self):
"""用户: 批量下载这些视频 BV1xx411c7mD BV1yy411c8nE
Agent → bilibili_downloader(action="batch_download",
urls=["BV1xx411c7mD", "BV1yy411c8nE"], quality="1080p")
"""
# Each video needs info + play_url responses
responses = []
for bvid in ["BV1xx411c7mD", "BV1yy411c8nE"]:
responses.append(_mock_response(_bilibili_ok(
_video_info_payload(bvid=bvid, title=f"Video {bvid}")
)))
responses.append(_mock_response(_bilibili_ok({
"dash": {
"video": [{"id": 80, "baseUrl": f"https://example.com/{bvid}_video.m4s",
"bandwidth": 2000000, "codecs": "avc1"}],
"audio": [{"id": 30280, "baseUrl": f"https://example.com/{bvid}_audio.m4s",
"bandwidth": 128000, "codecs": "mp4a"}],
}
})))
_patch_client(self.app.downloader, responses)
async def fake_download(url, filepath):
with open(filepath, "wb") as f:
f.write(b"\x00" * 100)
async def fake_merge(video_path, audio_path, output_path):
with open(output_path, "wb") as f:
f.write(b"\x00" * 200)
return True
self.app.downloader._download_stream = fake_download
self.app.downloader._merge_streams = fake_merge
with tempfile.TemporaryDirectory() as tmpdir:
result = _run(self.app.execute(
"downloader", "batch_download",
urls=["BV1xx411c7mD", "BV1yy411c8nE"],
quality="1080p",
output_dir=tmpdir,
))
self.assertIn("success", result)
class TestWatcherExamples(unittest.TestCase):
"""Examples 3, 8, 9, 19 from skill.md — Watcher module."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne()
# ── Example 3: get_stats ─────────────────────────────────────────
def test_example_03_get_stats(self):
"""用户: 这个视频有多少播放量和点赞？BV1xx411c7mD
Agent → bilibili_watcher(action="get_stats", url="BV1xx411c7mD")
"""
resp = _mock_response(_bilibili_ok(_video_info_payload()))
_patch_client(self.app.watcher, [resp])
result = _run(self.app.execute("watcher", "get_stats", url="BV1xx411c7mD"))
self.assertTrue(result["success"])
self.assertIn("stats", result)
self.assertEqual(result["stats"]["views"], 100000)
self.assertEqual(result["stats"]["likes"], 5000)
# ── Example 8: compare ───────────────────────────────────────────
def test_example_08_compare(self):
"""用户: 对比一下这两个视频的数据 BV1xx411c7mD 和 BV1yy411c8nE
Agent → bilibili_watcher(action="compare",
urls=["BV1xx411c7mD", "BV1yy411c8nE"])
"""
resp1 = _mock_response(_bilibili_ok(
_video_info_payload(bvid="BV1xx411c7mD", title="Video A")
))
resp2 = _mock_response(_bilibili_ok(
_video_info_payload(bvid="BV1yy411c8nE", title="Video B")
))
_patch_client(self.app.watcher, [resp1, resp2])
result = _run(self.app.execute(
"watcher", "compare",
urls=["BV1xx411c7mD", "BV1yy411c8nE"],
))
self.assertTrue(result["success"])
self.assertIn("ranking", result)
self.assertEqual(len(result["ranking"]), 2)
# ── Example 9: watch YouTube ─────────────────────────────────────
def test_example_09_watch_youtube(self):
"""用户: 这个YouTube视频有多少观看量？
Agent → bilibili_watcher(action="watch",
url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
"""
# YouTube watch falls back to oembed or returns platform info
oembed_resp = _mock_response({
"title": "Never Gonna Give You Up",
"author_name": "Rick Astley",
"author_url": "https://www.youtube.com/@RickAstley",
"thumbnail_url": "https://i.ytimg.com/vi/dQw4w9WgXcQ/hqdefault.jpg",
})
_patch_client(self.app.watcher, [oembed_resp])
result = _run(self.app.execute(
"watcher", "watch",
url="https://www.youtube.com/watch?v=dQw4w9WgXcQ",
))
# YouTube handling may vary — just verify the action dispatched correctly
self.assertIn("success", result)
# ── Example 19: track ────────────────────────────────────────────
def test_example_19_track(self):
"""用户: 监控这个视频的数据变化，每半小时看一次，跟踪6小时
Agent → bilibili_watcher(action="track",
url="BV1xx411c7mD", interval=30, duration=6)
Note: We override duration to near-zero to avoid long waits.
"""
resp = _mock_response(_bilibili_ok(_video_info_payload()))
_patch_client(self.app.watcher, [resp])
# Override to very short duration so the test finishes instantly
import time as time_module
original_track = self.app.watcher.track
async def fast_track(url, interval=60, duration=24, callback=None):
"""Shortened track that collects one data point."""
stats = await self.app.watcher.get_stats(url)
return {
"success": True,
"bvid": "BV1xx411c7mD",
"data_points": 1,
"duration_hours": duration,
"interval_minutes": interval,
"summary": {},
"data": [stats] if stats.get("success") else [],
}
self.app.watcher.track = fast_track
result = _run(self.app.execute(
"watcher", "track",
url="BV1xx411c7mD", interval=30, duration=6,
))
self.assertTrue(result["success"])
self.assertEqual(result["interval_minutes"], 30)
self.assertEqual(result["duration_hours"], 6)
# Restore
self.app.watcher.track = original_track
class TestSubtitleExamples(unittest.TestCase):
"""Examples 4, 11, 17 from skill.md — Subtitle module."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne()
# ── Example 4: download subtitle ─────────────────────────────────
def test_example_04_download_subtitle(self):
"""用户: 帮我下载这个B站视频的中文字幕
Agent → bilibili_subtitle(action="download",
url="BV1xx411c7mD", language="zh-CN", format="srt")
"""
# First call: get video info (for list_subtitles)
info_resp = _mock_response(_bilibili_ok({
"subtitle": {
"subtitles": [
{"lan": "zh-CN", "lan_doc": "中文（中国）",
"subtitle_url": "//example.com/sub_zh.json"},
]
},
"bvid": "BV1xx411c7mD",
"aid": 12345,
"title": "Test Video",
"pages": [{"cid": 10001, "page": 1}],
}))
# Second call: player API for subtitle list
player_resp = _mock_response(_bilibili_ok({
"subtitle": {
"subtitles": [
{"lan": "zh-CN", "lan_doc": "中文（中国）",
"subtitle_url": "//example.com/sub_zh.json"},
]
},
}))
# Third call: download subtitle JSON
sub_content_resp = _mock_response({
"body": [
{"from": 0.0, "to": 2.0, "content": "你好世界"},
{"from": 2.5, "to": 5.0, "content": "测试字幕"},
]
})
_patch_client(self.app.subtitle, [info_resp, player_resp, sub_content_resp])
with tempfile.TemporaryDirectory() as tmpdir:
result = _run(self.app.execute(
"subtitle", "download",
url="BV1xx411c7mD", language="zh-CN", format="srt",
output_dir=tmpdir,
))
self.assertIn("success", result)
# ── Example 11: list subtitles ───────────────────────────────────
def test_example_11_list_subtitles(self):
"""用户: 列出这个视频有哪些字幕可以下载
Agent → bilibili_subtitle(action="list", url="BV1xx411c7mD")
"""
resp = _mock_response(_bilibili_ok({
"subtitle": {
"subtitles": [
{"lan": "zh-CN", "lan_doc": "中文（中国）",
"subtitle_url": "//example.com/sub_zh.json"},
{"lan": "en", "lan_doc": "English",
"subtitle_url": "//example.com/sub_en.json"},
]
},
"bvid": "BV1xx411c7mD",
"aid": 12345,
"title": "Test Video",
"pages": [{"cid": 10001, "page": 1}],
}))
_patch_client(self.app.subtitle, [resp])
result = _run(self.app.execute("subtitle", "list", url="BV1xx411c7mD"))
self.assertIn("success", result)
# ── Example 17: convert subtitle ─────────────────────────────────
def test_example_17_convert_subtitle(self):
"""用户: 把SRT字幕转换成VTT格式
Agent → bilibili_subtitle(action="convert",
input_path="./video.srt", output_format="vtt")
"""
srt_content = (
"1\n"
"00:00:00,000 --> 00:00:02,000\n"
"Hello World\n\n"
"2\n"
"00:00:02,500 --> 00:00:05,000\n"
"Test subtitle\n\n"
)
with tempfile.TemporaryDirectory() as tmpdir:
srt_path = os.path.join(tmpdir, "video.srt")
with open(srt_path, "w", encoding="utf-8") as f:
f.write(srt_content)
result = _run(self.app.execute(
"subtitle", "convert",
input_path=srt_path, output_format="vtt",
))
self.assertIn("success", result)
if result.get("success"):
self.assertEqual(result["format"], "vtt")
class TestPlayerExamples(unittest.TestCase):
"""Examples 5, 13, 20 from skill.md — Player module."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne()
# ── Example 5: get_danmaku ───────────────────────────────────────
def test_example_05_get_danmaku(self):
"""用户: 获取这个视频的弹幕 BV1xx411c7mD
Agent → bilibili_player(action="get_danmaku", url="BV1xx411c7mD")
"""
info_resp = _mock_response(_bilibili_ok(_video_info_payload()))
# Danmaku response is XML
danmaku_xml_resp = MagicMock(spec=httpx.Response)
danmaku_xml_resp.status_code = 200
danmaku_xml_resp.text = (
'<?xml version="1.0" encoding="UTF-8"?>'
'<i>'
'<d p="1.0,1,25,16777215,1700000000,0,abc123,99999">Hello弹幕</d>'
'<d p="5.0,1,25,16777215,1700000001,0,def456,99998">Test弹幕</d>'
'</i>'
)
_patch_client(self.app.player, [info_resp, danmaku_xml_resp])
result = _run(self.app.execute("player", "get_danmaku", url="BV1xx411c7mD"))
self.assertTrue(result["success"])
self.assertIn("danmaku", result)
self.assertEqual(result["bvid"], "BV1xx411c7mD")
# ── Example 13: get_playurl ──────────────────────────────────────
def test_example_13_get_playurl(self):
"""用户: 获取这个视频的播放地址，720p的
Agent → bilibili_player(action="get_playurl",
url="BV1xx411c7mD", quality="720p")
"""
info_resp = _mock_response(_bilibili_ok(_video_info_payload()))
play_resp = _mock_response(_bilibili_ok({
"dash": {
"video": [
{"id": 64, "baseUrl": "https://example.com/720p.m4s",
"bandwidth": 1500000, "codecs": "avc1"},
],
"audio": [
{"id": 30280, "baseUrl": "https://example.com/audio.m4s",
"bandwidth": 128000, "codecs": "mp4a"},
],
},
"quality": 64,
"accept_quality": [80, 64, 32, 16],
}))
_patch_client(self.app.player, [info_resp, play_resp])
result = _run(self.app.execute(
"player", "get_playurl",
url="BV1xx411c7mD", quality="720p",
))
# get_playurl returns play stream info directly (no "success" key)
self.assertIn("play_type", result)
self.assertIn("video_streams", result)
self.assertEqual(result["current_quality"], "720p")
# ── Example 20: get_playlist ─────────────────────────────────────
def test_example_20_get_playlist(self):
"""用户: 这个视频有几个分P？列出播放列表
Agent → bilibili_player(action="get_playlist", url="BV1xx411c7mD")
"""
resp = _mock_response(_bilibili_ok(
_video_info_payload(pages=5)
))
_patch_client(self.app.player, [resp])
result = _run(self.app.execute("player", "get_playlist", url="BV1xx411c7mD"))
self.assertTrue(result["success"])
self.assertEqual(len(result["pages"]), 5)
self.assertEqual(result["page_count"], 5)
class TestPublisherExamples(unittest.TestCase):
"""Examples 6, 10, 14, 18 from skill.md — Publisher module."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne(
sessdata="test_sessdata",
bili_jct="test_csrf",
buvid3="test_buvid3",
)
# ── Example 6: upload ────────────────────────────────────────────
def test_example_06_upload(self):
"""用户: 帮我把这个视频上传到B站，标题叫"我的视频"
Agent → bilibili_publisher(action="upload",
file_path="./video.mp4", title="我的视频")
"""
with tempfile.TemporaryDirectory() as tmpdir:
video_path = os.path.join(tmpdir, "video.mp4")
with open(video_path, "wb") as f:
f.write(b"\x00" * 1024) # dummy video file
# Mock the internal methods
async def mock_preupload(file_path):
return {"success": True, "upload_url": "https://example.com/upload",
"auth": "test_auth", "biz_id": 12345, "upos_uri": "test_uri"}
async def mock_upload_file(file_path, preupload_result):
return {"success": True, "filename": "test_uploaded.mp4"}
async def mock_upload_cover(cover_path):
return {"success": True, "url": "https://example.com/cover.jpg"}
async def mock_submit_video(**kwargs):
return {
"success": True,
"bvid": "BV_new_video",
"aid": 99999,
"message": "Video published successfully",
}
self.app.publisher._preupload = mock_preupload
self.app.publisher._upload_file = mock_upload_file
self.app.publisher._upload_cover = mock_upload_cover
self.app.publisher._submit_video = mock_submit_video
result = _run(self.app.execute(
"publisher", "upload",
file_path=video_path, title="我的视频",
))
self.assertTrue(result["success"])
self.assertEqual(result["bvid"], "BV_new_video")
# ── Example 10: schedule ─────────────────────────────────────────
def test_example_10_schedule(self):
"""用户: 把这个视频定时明天晚上8点发布
Agent → bilibili_publisher(action="schedule",
file_path="./video.mp4", title="定时发布",
schedule_time="2025-12-31T20:00:00+08:00")
"""
with tempfile.TemporaryDirectory() as tmpdir:
video_path = os.path.join(tmpdir, "video.mp4")
with open(video_path, "wb") as f:
f.write(b"\x00" * 1024)
async def mock_preupload(file_path):
return {"success": True, "upload_url": "https://example.com/upload",
"auth": "test_auth", "biz_id": 12345, "upos_uri": "test_uri"}
async def mock_upload_file(file_path, preupload_result):
return {"success": True, "filename": "test_scheduled.mp4"}
async def mock_submit_video(**kwargs):
return {
"success": True,
"bvid": "BV_scheduled",
"aid": 88888,
"message": "Video scheduled successfully",
}
self.app.publisher._preupload = mock_preupload
self.app.publisher._upload_file = mock_upload_file
self.app.publisher._submit_video = mock_submit_video
result = _run(self.app.execute(
"publisher", "schedule",
file_path=video_path,
title="定时发布",
schedule_time="2025-12-31T20:00:00+08:00",
))
self.assertTrue(result["success"])
# ── Example 14: draft ────────────────────────────────────────────
def test_example_14_draft(self):
"""用户: 把这个视频存为草稿先不发布
Agent → bilibili_publisher(action="draft",
file_path="./video.mp4", title="草稿视频")
"""
with tempfile.TemporaryDirectory() as tmpdir:
video_path = os.path.join(tmpdir, "video.mp4")
with open(video_path, "wb") as f:
f.write(b"\x00" * 1024)
async def mock_preupload(file_path):
return {"success": True, "upload_url": "https://example.com/upload",
"auth": "test_auth", "biz_id": 12345, "upos_uri": "test_uri"}
async def mock_upload_file(file_path, preupload_result):
return {"success": True, "filename": "test_draft.mp4"}
draft_resp = _mock_response(_bilibili_ok({"aid": 77777}))
_patch_client(self.app.publisher, [draft_resp])
self.app.publisher._preupload = mock_preupload
self.app.publisher._upload_file = mock_upload_file
result = _run(self.app.execute(
"publisher", "draft",
file_path=video_path, title="草稿视频",
))
self.assertIn("success", result)
# ── Example 18: edit ─────────────────────────────────────────────
def test_example_18_edit(self):
"""用户: 修改我那个视频的标题和标签 BV1xx411c7mD
Agent → bilibili_publisher(action="edit",
bvid="BV1xx411c7mD", title="新标题", tags=["新标签"])
"""
# First call: get current video info
info_resp = _mock_response(_bilibili_ok(_video_info_payload()))
# Second call: edit API response
edit_resp = _mock_response(_bilibili_ok({}))
_patch_client(self.app.publisher, [info_resp, edit_resp])
result = _run(self.app.execute(
"publisher", "edit",
bvid="BV1xx411c7mD",
title="新标题",
tags=["新标签"],
))
self.assertTrue(result["success"])
self.assertEqual(result["bvid"], "BV1xx411c7mD")
class TestUnknownActions(unittest.TestCase):
"""Test that unknown skill names and actions are handled gracefully."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne()
def test_unknown_skill(self):
result = _run(self.app.execute("nonexistent_skill", "some_action"))
self.assertFalse(result["success"])
self.assertIn("Unknown skill", result["message"])
def test_unknown_action(self):
result = _run(self.app.execute("hot_monitor", "nonexistent_action"))
self.assertFalse(result["success"])
self.assertIn("Unknown action", result["message"])
class TestSkillNameAliases(unittest.TestCase):
"""Verify that all skill name aliases work correctly."""
def setUp(self):
from main import BilibiliAllInOne
self.app = BilibiliAllInOne()
def test_hot_monitor_aliases(self):
"""All hot_monitor aliases should resolve to the same module."""
hot_data = {"list": [], "no_more": True}
for alias in ["bilibili_hot_monitor", "hot_monitor", "hot"]:
_patch_client(self.app.hot_monitor, [
_mock_response(_bilibili_ok(hot_data))
])
result = _run(self.app.execute(alias, "get_hot"))
self.assertTrue(result["success"], f"Alias '{alias}' failed")
def test_downloader_aliases(self):
for alias in ["bilibili_downloader", "downloader", "download"]:
_patch_client(self.app.downloader, [
_mock_response(_bilibili_ok(_video_info_payload()))
])
result = _run(self.app.execute(alias, "get_info", url="BV1xx411c7mD"))
self.assertIn("success", result, f"Alias '{alias}' failed")
def test_watcher_aliases(self):
for alias in ["bilibili_watcher", "watcher", "watch"]:
_patch_client(self.app.watcher, [
_mock_response(_bilibili_ok(_video_info_payload()))
])
result = _run(self.app.execute(alias, "get_stats", url="BV1xx411c7mD"))
self.assertIn("success", result, f"Alias '{alias}' failed")
def test_subtitle_aliases(self):
for alias in ["bilibili_subtitle", "subtitle"]:
_patch_client(self.app.subtitle, [
_mock_response(_bilibili_ok({
"subtitle": {"subtitles": []},
"bvid": "BV1xx411c7mD", "aid": 12345,
"title": "Test", "pages": [{"cid": 10001, "page": 1}],
}))
])
result = _run(self.app.execute(alias, "list", url="BV1xx411c7mD"))
self.assertIn("success", result, f"Alias '{alias}' failed")
def test_player_aliases(self):
for alias in ["bilibili_player", "player", "play"]:
_patch_client(self.app.player, [
_mock_response(_bilibili_ok(_video_info_payload()))
])
result = _run(self.app.execute(alias, "get_playlist", url="BV1xx411c7mD"))
self.assertIn("success", result, f"Alias '{alias}' failed")
if __name__ == "__main__":
unittest.main()