starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加技能模块

Browse Source
This commit is contained in:
starmen1265
2026-02-15 01:17:39 +08:00
parent c417a8f7a9
commit 86daba5771
48 changed files with 10121 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kb-bilibili-content/SKILL.md
+91
View File
@@ -0,0 +1,91 @@
# KB Bilibili Content Skill
**kb-bilibili-content** - Download Bilibili videos and extract content.
## Features
- Video download using **yt-dlp** (recommended)
- Subtitle extraction
- Article content extraction
- Multiple quality options (360P/480P/720P/1080P)
- Supports `BILIBILI_COOKIE` environment variable
## Environment Variables
| Variable | Required | Description |
|----------|----------|-------------|
| `BILIBILI_COOKIE` | No | B站登录 Cookie (for member-only videos) |
## Commands
```bash
# Download video (720P default)
python bilibili_content.py --id "BV1xx411c7mD" --type video --output "/path/to/downloads/"
# Download video (1080P)
python bilibili_content.py --id "BV1xx411c7mD" --type video --quality 80 --output "/path/to/downloads/"
# Download video (480P)
python bilibili_content.py --id "BV1xx411c7mD" --type video --quality 32 --output "/path/to/downloads/"
# Extract subtitle
python bilibili_content.py --id "BV1xx411c7mD" --type subtitle --output "/path/to/subtitles/"
# Get article content
python bilibili_content.py --id "cv43780186" --type article --output "/path/to/article.md"
```
## Output
```json
{
"success": true,
"type": "video",
"filename": "video_title.mp4",
"filepath": "/path/to/downloads/video_title.mp4",
"message": "下载完成: video_title.mp4"
}
```
## Cookie Setup
### Get Cookie from Browser
1. Login to bilibili.com
2. Press F12 → Network → Refresh page
3. Copy Cookie from request headers
### Set Environment Variable
```bash
# Add to ~/.bashrc for persistence
export BILIBILI_COOKIE="SESSDATA=xxx; bili_jct=xxx; DedeUserID=xxx; DedeUserID__ckMd5=xxx"
# Or set temporarily
export BILIBILI_COOKIE="your_cookie_here"
```
## Installation
```bash
cp -r kb-bilibili-content/ ~/skills/
pip install aiohttp playwright
playwright install chromium
```
## How It Works
Video download uses **yt-dlp** with Netscape-format cookie file:
1. Parses `BILIBILI_COOKIE` environment variable
2. Creates temporary Netscape-format cookie file
3. Runs `yt-dlp --cookies <file> -f <format> -o <output>`
4. Cleans up temporary file
## Quality Options
| Quality | ID | Resolution | Notes |
|---------|-----|------------|-------|
| 16 | 360P | 640x360 | Fastest |
| 32 | 480P | 852x480 | Default for low bandwidth |
| 64 | 720P | 1280x720 | **Recommended** |
| 80 | 1080P | 1920x1080 | Best quality |
Default quality: **64 (720P)**
kb-bilibili-content/__pycache__/bilibili_content.cpython-312.pyc
BIN
View File
Binary file not shown.
kb-bilibili-content/bilibili_content.py
+419
View File
@@ -0,0 +1,419 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Bilibili Content Downloader - B站内容获取工具
功能:
- download_video(bvid, quality=480) - 下载视频（使用 yt-dlp）
- extract_subtitle(bvid) - 提取字幕
- extract_article(cvid) - 获取专栏全文（使用Playwright）
- 命令行参数支持
- JSON 输出
"""
import argparse
import asyncio
import json
import os
import re
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List
class BilibiliContent:
"""B站内容获取器"""
def __init__(self, output_dir: str = "./output", cookie: str = None):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
self.cookie = cookie
def _create_cookie_file(self) -> str:
"""创建 Netscape 格式的 Cookie 文件"""
if not self.cookie:
return None
# 解析 Cookie 字符串
cookies = {}
for item in self.cookie.split(';'):
item = item.strip()
if '=' in item:
key, value = item.split('=', 1)
cookies[key.strip()] = value.strip()
# 创建 Netscape 格式的 Cookie 文件
cookie_content = """# Netscape HTTP Cookie File
# Generated by bilibili_content.py
"""
for name, value in cookies.items():
cookie_content += f".bilibili.com\tTRUE\t/\tFALSE\t0\t{name}\t{value}\n"
# 写入临时文件
fd, path = tempfile.mkstemp(suffix='.txt')
with os.fdopen(fd, 'w') as f:
f.write(cookie_content)
return path
async def download_video(self, bvid: str, quality: int = 64) -> Dict[str, Any]:
"""下载B站视频（使用 yt-dlp）"""
result = {"success": False, "bvid": bvid, "message": "", "filepath": None, "metadata": {}}
try:
# 先获取视频信息
print(f"获取视频信息: {bvid}")
cookie_file = self._create_cookie_file()
url = f"https://www.bilibili.com/video/{bvid}"
try:
# 获取视频标题（用于文件名）
if cookie_file:
cmd = ["yt-dlp", "--cookies", cookie_file, "--get-title", url]
else:
cmd = ["yt-dlp", "--get-title", url]
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
title = stdout.decode('utf-8').strip()
if not title or 'error' in title.lower():
# 尝试备用方法
title = f"video_{bvid}"
except Exception as e:
title = f"video_{bvid}"
# 构建 yt-dlp 命令
safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_', '丨')).strip()[:50]
# 选择视频格式
if quality == 80:
# 1080P
format_str = "30112+30280"
elif quality == 64:
# 720P
format_str = "30064+30280"
elif quality == 32:
# 480P
format_str = "30032+30280"
else:
# 360P
format_str = "30016+30280"
print(f"下载视频: {title}")
print(f"格式: {format_str}")
output_template = str(self.output_dir / f"{safe_title}.%(ext)s")
cmd = [
"yt-dlp",
"-f", format_str,
"-o", output_template,
"--no-download-archive",
"-q",
url
]
if cookie_file:
cmd.extend(["--cookies", cookie_file])
print(f"执行命令: {' '.join(cmd[:4])}...")
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
# 查找下载的文件
downloaded_files = list(self.output_dir.glob(f"{safe_title}.*"))
if downloaded_files:
video_file = downloaded_files[0]
result["success"] = True
result["filepath"] = str(video_file)
result["message"] = f"下载完成: {video_file.name}"
result["metadata"] = {"title": title, "bvid": bvid}
print(f"✅ {result['message']}")
else:
result["message"] = "未找到下载的文件"
print(f"❌ {result['message']}")
# 清理临时 Cookie 文件
if cookie_file and os.path.exists(cookie_file):
os.remove(cookie_file)
except Exception as e:
result["message"] = f"错误: {str(e)}"
print(f"❌ {result['message']}")
return result
async def extract_subtitle(self, bvid: str) -> Dict[str, Any]:
"""
提取字幕
注意：没有找到官方字幕是正常的！很多视频没有自动生成的字幕。
返回 success=True 即使没有字幕，让工作流继续执行后续步骤（如 Whisper）。
"""
result = {"success": False, "bvid": bvid, "subtitles": [], "has_subtitle": False, "message": ""}
try:
import aiohttp
api_url = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}"
async with aiohttp.ClientSession() as session:
async with session.get(api_url) as resp:
data = await resp.json()
pages = data.get('data', [])
if not pages:
result["message"] = "未找到分P（这是正常的）"
# 仍然返回 success=True，让工作流继续
result["success"] = True
return result
cid = pages[0].get('cid', '')
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}") as resp:
sub_data = await resp.json()
subtitles = sub_data.get('data', {}).get('subtitle', {}).get('list', [])
if not subtitles:
result["message"] = "没有官方字幕（这是正常的），将使用 Whisper 进行语音识别"
# 仍然返回 success=True，让工作流继续
result["success"] = True
return result
result["success"] = True
result["has_subtitle"] = True
result["subtitles"] = [{"lan": s.get('lan'), "title": s.get('lan_doc')} for s in subtitles]
result["message"] = f"✅ 找到 {len(subtitles)} 个字幕，将使用官方字幕进行深度分析"
print(result["message"])
except Exception as e:
result["message"] = f"获取字幕出错（这是正常的），将使用 Whisper: {str(e)}"
# 出错时也返回 success=True，让工作流继续
result["success"] = True
return result
async def download_subtitle(self, bvid: str) -> Dict[str, Any]:
"""
下载字幕并转换为 SRT 格式
如果没有官方字幕，返回 success=True 但 has_subtitle=False，
让工作流继续使用 Whisper 进行语音识别。
"""
result = {"success": False, "bvid": bvid, "filepath": None, "has_subtitle": False, "message": ""}
try:
import aiohttp
# 获取 CID
api_url = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}"
async with aiohttp.ClientSession() as session:
async with session.get(api_url) as resp:
data = await resp.json()
pages = data.get('data', [])
if not pages:
result["message"] = "未找到分P，将使用 Whisper 进行语音识别"
result["success"] = True
return result
cid = pages[0].get('cid', '')
# 获取字幕列表
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.bilibili.com/x/player/v2?bvid={bvid}&cid={cid}") as resp:
sub_data = await resp.json()
subtitles = sub_data.get('data', {}).get('subtitle', {}).get('list', [])
if not subtitles:
result["message"] = "没有官方字幕，将使用 Whisper 进行语音识别"
result["success"] = True
return result
# 选择中文字幕，如果没有则选第一个
zh_sub = None
for s in subtitles:
if s.get('lan') in ('ai-zh', 'zh-CN', 'zh'):
zh_sub = s
break
if not zh_sub:
zh_sub = subtitles[0]
# 下载字幕文件
sub_url = zh_sub.get('subtitle_url', '')
if not sub_url:
result["message"] = "字幕URL为空，将使用 Whisper"
result["success"] = True
return result
if sub_url.startswith('//'):
sub_url = 'https:' + sub_url
async with aiohttp.ClientSession() as session:
async with session.get(sub_url) as resp:
sub_content = await resp.json()
# 转换为 SRT 格式
body = sub_content.get('body', [])
if not body:
result["message"] = "字幕内容为空，将使用 Whisper"
result["success"] = True
return result
def format_srt_time(t: float) -> str:
"""将秒数转换为 SRT 时间格式"""
h = int(t // 3600)
m = int((t % 3600) // 60)
s = int(t % 60)
ms = int((t - int(t)) * 1000)
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
srt_lines = []
for i, line in enumerate(body, 1):
start = line.get('from', 0)
end = line.get('to', 0)
content = line.get('content', '')
srt_lines.extend([str(i), f"{format_srt_time(start)} --> {format_srt_time(end)}", content, ""])
srt_content = "\n".join(srt_lines)
# 保存 SRT 文件
output_path = self.output_dir / f"{bvid}.srt"
with open(output_path, 'w', encoding='utf-8') as f:
f.write(srt_content)
result["success"] = True
result["has_subtitle"] = True
result["filepath"] = str(output_path)
result["message"] = f"✅ 官方字幕已下载: {output_path.name} ({len(body)}条)"
print(result["message"])
except Exception as e:
result["message"] = f"下载字幕出错: {str(e)}，将使用 Whisper"
result["success"] = True # 仍然返回成功，让工作流继续
return result
async def extract_article(self, cvid: str) -> Dict[str, Any]:
"""获取专栏全文（使用Playwright异步API）"""
result = {"success": False, "cvid": cvid, "title": "", "content": "", "message": ""}
try:
from playwright.async_api import async_playwright
cvid_clean = cvid.replace('cv', '')
url = f"https://www.bilibili.com/read/cv{cvid_clean}"
print(f"获取专栏: {url}")
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
await page.goto(url, wait_until="networkidle")
await page.wait_for_timeout(3000)
# 提取标题
title_elem = await page.query_selector('h1')
if title_elem:
result["title"] = (await title_elem.inner_text()).strip()
print(f"标题: {result['title'][:50]}...")
# 提取正文内容
content_elem = await page.query_selector('.read-article-holder')
if content_elem:
html_content = await content_elem.inner_html()
text_content = await content_elem.inner_text()
result["content"] = text_content.strip()
result["html_content"] = html_content.strip()
result["success"] = bool(result["title"] and result["content"])
print(f"内容长度: {len(result['content'])} 字符")
else:
result["message"] = "未找到专栏内容"
await browser.close()
if result["success"]:
result["message"] = f"✅ 获取成功: {result['title'][:30]}..."
else:
result["message"] = f"获取失败: {result.get('message', '未知错误')}"
except Exception as e:
result["message"] = f"错误: {str(e)}"
print(f"错误: {e}")
return result
async def main():
parser = argparse.ArgumentParser(description='Bilibili Content Downloader')
parser.add_argument('--id', '-i', required=True, help='B站内容ID')
parser.add_argument('--type', '-t', required=True, choices=['video', 'subtitle', 'subtitle-download', 'article'], help='内容类型')
parser.add_argument('--output', '-o', default='./output', help='输出目录')
parser.add_argument('--quality', '-q', type=int, default=64, help='视频质量: 16=360P, 32=480P, 64=720P, 80=1080P')
args = parser.parse_args()
cookie = os.environ.get('BILIBILI_COOKIE')
downloader = BilibiliContent(output_dir=args.output, cookie=cookie)
if args.type == 'video':
result = await downloader.download_video(args.id, quality=args.quality)
elif args.type == 'subtitle':
result = await downloader.extract_subtitle(args.id)
elif args.type == 'subtitle-download':
# 下载并转换字幕为 SRT 格式
result = await downloader.download_subtitle(args.id)
else:
result = await downloader.extract_article(args.id.replace('cv', ''))
# 统一输出格式
if result.get("success"):
output = {
"success": True,
"type": args.type,
"has_subtitle": result.get("has_subtitle", False),
"filename": Path(result.get('filepath', '')).name if result.get('filepath') else None,
"filepath": str(Path(result.get('filepath', '')).resolve()) if result.get('filepath') else None,
"message": result.get('message', '处理成功')
}
else:
output = {
"success": False,
"error": result.get('message', '处理失败'),
"detail": f"ID: {args.id}, 类型: {args.type}"
}
print(json.dumps(output, ensure_ascii=False, indent=2))
return result
if __name__ == '__main__':
try:
result = asyncio.run(main())
sys.exit(0 if result.get("success") else 1)
except KeyboardInterrupt:
print(json.dumps({"success": False, "error": "用户取消"}, ensure_ascii=False, indent=2))
sys.exit(1)
except Exception as e:
print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2))
sys.exit(1)
kb-bilibili-content/bilibili_extractor.py
+155
View File
@@ -0,0 +1,155 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
B站视频提取脚本
提取视频信息、统计数据、字幕并生成Markdown笔记
"""
import sys
import os
import json
import re
from datetime import datetime
# 添加skills路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'skills', 'obsidian-kb'))
from src.extractors.bilibili_extractor import BilibiliExtractor
from src.config import get_config
def format_number(num):
"""格式化数字"""
if num >= 100000000:
return f"{num / 100000000:.1f}亿"
elif num >= 10000:
return f"{num / 10000:.1f}万"
else:
return f"{num:,}"
def generate_markdown(result, bvid):
"""生成Markdown笔记"""
metadata = result['metadata']
stats = result['stats']
description = result['description']
subtitle = result['subtitle']
# 获取当前时间
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
md = f"""# {metadata['title']}
> **来源**: [B站](https://www.bilibili.com/video/{bvid})
> **作者**: {metadata['author']}
> **发布时间**: {metadata['pubtime']}
> **提取时间**: {now}
> **时长**: {metadata['duration']}
## 📊 统计数据
| 指标 | 数量 |
|------|------|
| 👀 播放 | {format_number(stats['view'])} |
| 💬 弹幕 | {format_number(stats['danmaku'])} |
| 📝 评论 | {format_number(stats['reply'])} |
| ⭐ 收藏 | {format_number(stats['favorite'])} |
| 🪙 投币 | {format_number(stats['coin'])} |
| 🔗 分享 | {format_number(stats['share'])} |
| 👍 点赞 | {format_number(stats['like'])} |
## 📝 视频简介
{description if description else '(暂无简介)'}
## 📌 标签
{metadata['tags'] if metadata['tags'] else '(无标签)'}
## 🎬 字幕
{subtitle if subtitle else '(无字幕)'}
---
**元数据**:
- BV号: `{metadata['bvid']}`
- AV号: `{metadata['aid']}`
- UP主ID: `{metadata['mid']}`
"""
return md
def main():
"""主函数"""
# BV号
bvid = "BV1N5foBjE5K" # 从 https://b23.tv/btMGoOi 解析得到
print(f"开始提取B站视频: {bvid}")
print("=" * 60)
# 获取配置
try:
config = get_config()
except Exception as e:
print(f"⚠️ 获取配置失败，使用默认配置: {e}")
config = None
# 创建提取器
extractor = BilibiliExtractor(config=config)
# 提取视频信息
try:
result = extractor.extract(bvid)
print("\n✅ 提取成功！\n")
print("=" * 60)
print("基本信息:")
print("=" * 60)
print(f"标题: {result['metadata']['title']}")
print(f"作者: {result['metadata']['author']}")
print(f"时长: {result['metadata']['duration']}")
print(f"发布: {result['metadata']['pubtime']}")
print("\n" + "=" * 60)
print("统计数据:")
print("=" * 60)
print(f"播放: {format_number(result['stats']['view'])}")
print(f"弹幕: {format_number(result['stats']['danmaku'])}")
print(f"评论: {format_number(result['stats']['reply'])}")
print(f"收藏: {format_number(result['stats']['favorite'])}")
print(f"投币: {format_number(result['stats']['coin'])}")
print(f"点赞: {format_number(result['stats']['like'])}")
# 生成 Markdown 笔记
md_content = generate_markdown(result, bvid)
# 确保 notes 目录存在
notes_dir = "/home/admin/.openclaw/workspace/notes"
os.makedirs(notes_dir, exist_ok=True)
# 生成文件名（使用BV号作为文件名）
filename = f"{bvid}.md"
filepath = os.path.join(notes_dir, filename)
# 写入文件
with open(filepath, 'w', encoding='utf-8') as f:
f.write(md_content)
print("\n" + "=" * 60)
print(f"✅ Markdown笔记已生成: {filepath}")
print("=" * 60)
# 返回笔记内容
return md_content
except Exception as e:
print(f"\n❌ 提取失败: {e}")
import traceback
traceback.print_exc()
return None
if __name__ == '__main__':
main()
kb-bilibili-content/video_downloader.py
+175
View File
@@ -0,0 +1,175 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
使用旧版 API 获取并下载视频流
"""
import asyncio
import json
import os
from playwright.async_api import async_playwright
async def download_video_stream(bvid, output_dir="/tmp/bilibili_downloads"):
"""下载视频流"""
url = f"https://www.bilibili.com/video/{bvid}"
# 创建输出目录
os.makedirs(output_dir, exist_ok=True)
async with async_playwright() as playwright:
browser = await playwright.chromium.launch(headless=True)
context = await browser.new_context()
page = await context.new_page()
print(f"正在下载视频: {url}")
print("=" * 80)
# 首先获取 aid 和 cid
print("\n1. 获取视频信息...")
try:
api_url = "https://api.bilibili.com/x/web-interface/view"
response = await page.request.get(api_url, params={'bvid': bvid})
data = await response.json()
if data.get('code') != 0:
print(f" ❌ 失败: {data.get('message')}")
await browser.close()
return None
video_data = data.get('data', {})
title = video_data.get('title')
aid = video_data.get('aid')
cid = video_data.get('cid')
duration = video_data.get('duration')
print(f" 标题: {title}")
print(f" 时长: {duration}秒")
except Exception as e:
print(f" ❌ 异常: {e}")
await browser.close()
return None
# 获取视频流 URL
print("\n2. 获取视频流 URL...")
try:
playurl_api = "https://api.bilibili.com/x/player/playurl"
params = {
'bvid': bvid,
'cid': cid,
'qn': 80, # 1080P
'otype': 'json',
'fnver': 0,
'fnval': 0, # FLV 格式
}
playurl_response = await page.request.get(playurl_api, params=params)
if not playurl_response.ok:
print(f" ❌ HTTP 错误: {playurl_response.status}")
await browser.close()
return None
playurl_data = await playurl_response.json()
if playurl_data.get('code') != 0:
print(f" ❌ API 错误: {playurl_data.get('message')}")
await browser.close()
return None
play_data = playurl_data.get('data', {})
# 检查支持的清晰度
print(f"\n 支持的清晰度:")
for i, (q, desc) in enumerate(zip(play_data.get('accept_quality', []), play_data.get('accept_description', []))):
current = " (当前)" if q == play_data.get('quality') else ""
print(f" {desc} ({q}){current}")
# 获取视频 URL
durl = play_data.get('durl', [])
if not durl:
print(f" ❌ 没有找到视频流")
await browser.close()
return None
print(f"\n 找到 {len(durl)} 个视频分段")
video_urls = []
for i, segment in enumerate(durl):
segment_url = segment.get('url')
segment_size = segment.get('size', 0)
video_urls.append(segment_url)
print(f" 分段 {i+1}: {segment_size / 1024 / 1024:.2f}MB")
print(f" {segment_url[:100]}...")
except Exception as e:
print(f" ❌ 异常: {e}")
import traceback
traceback.print_exc()
await browser.close()
return None
# 下载视频
print(f"\n3. 开始下载视频...")
all_segments = []
total_size = 0
for i, video_url in enumerate(video_urls, 1):
print(f"\n 下载分段 {i}/{len(video_urls)}...")
try:
# 添加必要的请求头
headers = {
'Referer': 'https://www.bilibili.com/',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
}
# 下载分段
segment_response = await page.request.get(
video_url,
headers=headers,
timeout=60000
)
if not segment_response.ok:
print(f" ❌ 下载失败: {segment_response.status}")
continue
segment_data = await segment_response.body()
segment_size = len(segment_data)
total_size += segment_size
all_segments.append(segment_data)
print(f" ✅ 下载成功: {segment_size / 1024 / 1024:.2f}MB")
except Exception as e:
print(f" ❌ 下载异常: {e}")
continue
if not all_segments:
print(f"\n ❌ 没有成功下载任何分段")
await browser.close()
return None
# 合并分段
print(f"\n4. 合并视频分段...")
video_data = b''.join(all_segments)
# 保存文件
safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:50]
output_path = os.path.join(output_dir, f"{safe_title}.flv")
with open(output_path, 'wb') as f:
f.write(video_data)
print(f" ✅ 视频已保存: {output_path}")
print(f" 文件大小: {total_size / 1024 / 1024:.2f}MB")
await browser.close()
return output_path
if __name__ == '__main__':
bvid = "BV1KtFzzqEUG"
result = asyncio.run(download_video_stream(bvid))
if result:
print(f"\n✅ 下载完成！")
print(f"文件路径: {result}")
else:
print(f"\n❌ 下载失败")