refactor: remove deprecated kb-xxx sub-skills
PR Test / lint (pull_request) Successful in 2s

Remove sub-skills that have been replaced by kb_pipeline:
- kb-link-parser -> kb_pipeline/link_parser.py
- kb-whisper -> kb_pipeline/processors/asr.py
- kb-whisper-diarization -> kb_pipeline/processors/asr.py
- kb-bilibili-content -> BilibiliAdapter.download_video()
- kb-zhihu-content -> ZhihuAdapter.fetch_content()
- kb-ocr-processor -> unused
- kb-funasr -> unused (ASR uses Whisper)

Retained skills (still used by kb_pipeline):
- kb-bilibili-metadata
- kb-deep-analyzer
- kb-zhihu-metadata

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-16 12:03:57 +08:00
parent ca57a2f901
commit 30552f2e5c
29 changed files with 0 additions and 4859 deletions
-91
View File
@@ -1,91 +0,0 @@
# 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)**
-419
View File
@@ -1,419 +0,0 @@
#!/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)
-155
View File
@@ -1,155 +0,0 @@
#!/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()
-175
View File
@@ -1,175 +0,0 @@
#!/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❌ 下载失败")
-2
View File
@@ -1,2 +0,0 @@
# FunASR 模型缓存目录
FUNASR_CACHE_DIR=~/.cache/modelscope/hub
-212
View File
@@ -1,212 +0,0 @@
# KB FunASR - 本地语音识别
基于阿里 FunASR 的本地语音识别工具,支持将音视频文件转换为文本和 JSON 格式。
## 功能特性
- 🎯 **本地执行** - 完全离线运行,保护隐私
- 🚀 **多模型支持** - Paraformer、SenseVoice 等模型可选
- 📝 **多格式输出** - JSON、TXT 两种格式
- ⚡ **异步执行** - 支持后台运行,不阻塞主代理
- 🎬 **广泛兼容** - 支持常见音视频格式(MP3、MP4、WAV 等)
- 🇨🇳 **中文优化** - Paraformer 模型针对中文语音优化
## 安装前提
### 1. 安装依赖
```bash
pip install funasr modelscope
```
### 2. 配置环境变量(可选)
在 `~/.clawdbot/clawdbot.json` 中添加:
```json5
{
skills: {
entries: {
"kb-funasr": {
env: {
FUNASR_CACHE_DIR: "~/.cache/modelscope/hub"
}
}
}
}
}
```
或通过命令行配置:
```bash
clawdbot skills config kb-funasr FUNASR_CACHE_DIR "~/.cache/modelscope/hub"
```
## 使用方法
### 基本用法
```bash
# 使用默认 Paraformer 中文模型处理
python funasr.py --input video.mp4
# 后台异步执行
python funasr.py --input video.mp4 --background
```
### 指定模型
```bash
# Paraformer 中文模型(推荐)
python funasr.py --input video.mp4 --model paraformer-zh
# Paraformer 多语言模型
python funasr.py --input video.mp4 --model paraformer
# SenseVoice 多语言模型
python funasr.py --input video.mp4 --model sensevoice
```
### 指定输出目录
```bash
python funasr.py --input video.mp4 --output ./transcripts
```
## 输出文件
对于输入 `video.mp4`,将生成以下文件:
```
output/
├── video.json # 详细结果(含时间戳)
└── video.txt # 纯文本
```
### JSON 格式
```json
{
"success": true,
"input_file": "video.mp4",
"model": "paraformer-zh",
"output_files": {
"json": "video.json",
"txt": "video.txt"
},
"duration_seconds": 120.5,
"segments": [
{
"id": 0,
"start": 0.0,
"end": 3.2,
"text": "这是第一段识别结果"
},
{
"id": 1,
"start": 3.2,
"end": 6.8,
"text": "这是第二段识别结果"
}
]
}
```
## 模型对比
| 模型 | 速度 | 精度 | 适用场景 |
|------|------|------|----------|
| paraformer-zh | ⚡⚡⚡ | ⭐⭐⭐⭐ | 中文语音(推荐) |
| paraformer | ⚡⚡ | ⭐⭐⭐⭐ | 多语言识别 |
| sensevoice | ⚡ | ⭐⭐⭐⭐⭐ | 高精度多语言 |
## 完整示例
### 示例 1: 转录播客
```bash
python funasr.py --input podcast.mp3 --model paraformer-zh --background
```
### 示例 2: 生成视频识别结果
```bash
python funasr.py --input tutorial.mp4 --model paraformer
```
### 示例 3: 批量处理(配合脚本)
```bash
#!/bin/bash
for video in videos/*.mp4; do
python funasr.py --input "$video" --model paraformer-zh
done
```
## 在 Workflow 中使用
在知识库处理流程中,如果视频没有识别结果:
```bash
# 步骤4: 下载视频后
# 步骤5: 使用 FunASR 生成识别结果
python funasr.py --input ${video_path} --model paraformer-zh --background
# 等待 FunASR 完成,然后继续后续步骤
```
## 技术细节
- **引擎**: 阿里 FunASR
- **模型框架**: ModelScope
- **音频采样率**: 自动转换为 16kHz
- **语言**: 根据模型支持不同语言
## 故障排查
### 问题: 模型下载失败
```bash
# 首次运行会自动下载,确保网络稳定
python funasr.py --input test.mp4
# 手动检查缓存目录
ls ~/.cache/modelscope/hub/
```
### 问题: 音频格式不支持
```bash
# 转换音频格式
ffmpeg -i input.avi -acodec libmp3lame output.mp3
python funasr.py --input output.mp3
```
### 问题: 依赖安装失败
```bash
# 使用清华镜像源
pip install funasr modelscope -i https://pypi.tuna.tsinghua.edu.cn/simple
```
## 性能参考
在 CPU 环境下(Intel i7):
- **paraformer-zh 模型**: ~2x 实时速度(1分钟音频 ~30秒处理)
- **paraformer 模型**: ~1.5x 实时速度(1分钟音频 ~40秒处理)
- **sensevoice 模型**: ~1x 实时速度(1分钟音频 ~1分钟处理)
## 与 Whisper 对比
| 特性 | FunASR | Whisper |
|------|--------|---------|
| 中文识别 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ |
| 速度 | ⚡⚡⚡ | ⚡⚡ |
| 模型大小 | 较小 | 较大 |
| 多语言支持 | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
## 许可证
本技能使用 MIT 许可证,FunASR 使用 Apache 2.0 许可证。
-113
View File
@@ -1,113 +0,0 @@
---
name: kb-funasr
description: 本地语音识别,支持音视频转文字(基于阿里 FunASR)
---
# KB FunASR Skill
**kb-funasr** - 本地语音识别,使用阿里 FunASR 将音视频文件转换为文本和 JSON 格式输出。
## 什么时候使用
当用户需要:
- 将视频/音频文件转换为文字
- 获取带时间戳的详细识别结果
- 需要中文优化的语音识别(Paraformer 模型)
## 环境变量
在使用之前,必须检查环境变量是否已经设置。
| 变量名 | 用途 | 示例 |
|--------|------|------|
| `FUNASR_CACHE_DIR` | FunASR 模型缓存目录 | `~/.cache/modelscope/hub` |
## 使用方法
```bash
# 基本用法(使用 Paraformer 模型)
python funasr.py --input video.mp4
# 指定模型
python funasr.py --input video.mp4 --model paraformer-zh
# 指定输出目录
python funasr.py --input video.mp4 --output ./output
# 异步执行(推荐)
python funasr.py --input video.mp4 --background
```
## 命令参数
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `--input` | 输入音视频文件路径 | 必需 |
| `--model` | FunASR 模型类型 | `paraformer-zh` |
| `--output` | 输出目录 | 文件所在目录 |
| `--background` | 后台异步执行 | False |
## 模型选项
- `paraformer-zh` - Paraformer 中文模型(轻量,中文优化,推荐)
- `paraformer` - Paraformer 多语言模型
- `sensevoice` - SenseVoice 多语言模型
## 输出文件
对于输入 `video.mp4`,将生成:
- `video.json` - 详细结果(包含时间戳、分段信息)
- `video.txt` - 纯文本内容
## 输出格式示例
### JSON 格式
```json
{
"success": true,
"input_file": "video.mp4",
"model": "paraformer-zh",
"output_files": {
"json": "video.json",
"txt": "video.txt"
},
"duration_seconds": 120.5,
"segments": [
{
"id": 0,
"start": 0.0,
"end": 3.2,
"text": "这是第一段识别结果"
}
]
}
```
## Workflow 中的使用
```bash
# 步骤4之后:如果已有音频但无字幕,可以使用 FunASR
python funasr.py --input ${video_path} --model paraformer-zh --background
```
## 安装
```bash
# 1. 安装依赖
pip install funasr modelscope
# 2. 首次运行时会自动下载模型到缓存目录
python funasr.py --input test.mp4
# 3. 复制技能到 skills 目录
cp -r kb-funasr/ ~/skills/
```
## 注意事项
- **异步执行**: 使用 `--background` 参数时,命令会在后台运行,不阻塞主代理
- **模型选择**: paraformer-zh 模型适合中文语音识别,速度较快,精度较高
- **文件格式**: 支持 MP3、MP4、WAV、FLAC、OGG、M4A 等常见格式
- **首次运行**: 首次运行时会自动下载模型文件,需要稳定的网络连接
Binary file not shown.
-313
View File
@@ -1,313 +0,0 @@
#!/usr/bin/env python3
"""
KB FunASR - 本地语音识别工具
基于阿里 FunASR 实现音视频转文字
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional
class FunASRTranscriber:
"""FunASR 语音识别器"""
# 模型映射
MODEL_MAP = {
"paraformer-zh": "damo/speech_paraformer-large-vad_punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch",
"paraformer": "damo/speech_paraformer-large_asr_nat-zh-cn-16k-common-vocab8404-pytorch",
"sensevoice": "FunASR/SenseVoiceSmall"
}
def __init__(self, model: str = "paraformer-zh", cache_dir: Optional[str] = None):
"""初始化 FunASR 识别器
Args:
model: 模型名称
cache_dir: 模型缓存目录
"""
self.model_name = model
self.model_id = self.MODEL_MAP.get(model, self.MODEL_MAP["paraformer-zh"])
self.cache_dir = cache_dir or os.path.expanduser("~/.cache/modelscope/hub")
self._model = None
def _load_model(self):
"""延迟加载模型"""
if self._model is None:
try:
from funasr import AutoModel
print(f"正在加载模型: {self.model_name} ({self.model_id})")
print(f"缓存目录: {self.cache_dir}")
self._model = AutoModel(
model=self.model_id,
disable_pbar=False,
disable_log=False,
)
print("模型加载完成")
except ImportError:
print("错误: 未安装 funasr 库")
print("请运行: pip install funasr modelscope")
sys.exit(1)
def _convert_audio(self, input_path: str) -> str:
"""转换音频格式为 16kHz WAV
Args:
input_path: 输入文件路径
Returns:
转换后的文件路径(如果不需要转换则返回原路径)
"""
# 检查是否已经是 WAV 格式
if input_path.lower().endswith('.wav'):
# 检查采样率
try:
result = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries',
'stream=sample_rate', '-of', 'default=noprint_wrappers=1:nokey=1',
input_path],
capture_output=True, text=True
)
if result.returncode == 0:
sample_rate = int(result.stdout.strip())
if sample_rate == 16000:
return input_path
except Exception:
pass
# 需要转换
output_path = input_path.rsplit('.', 1)[0] + '_converted.wav'
print(f"正在转换音频格式: {input_path} -> {output_path}")
try:
subprocess.run([
'ffmpeg', '-y', '-i', input_path,
'-acodec', 'pcm_s16le',
'-ar', '16000',
'-ac', '1',
output_path
], check=True, capture_output=True)
print("音频转换完成")
return output_path
except subprocess.CalledProcessError as e:
print(f"音频转换失败: {e}")
# 如果转换失败,尝试使用原文件
print("尝试使用原文件继续处理")
return input_path
except FileNotFoundError:
print("警告: 未找到 ffmpeg,跳过音频转换")
return input_path
def transcribe(self, input_path: str) -> Dict:
"""转录音频文件
Args:
input_path: 输入文件路径
Returns:
识别结果字典
"""
input_path = os.path.abspath(input_path)
if not os.path.exists(input_path):
return {
"success": False,
"error": f"文件不存在: {input_path}"
}
print(f"\n开始处理: {input_path}")
print(f"使用模型: {self.model_name}")
# 转换音频格式
audio_path = self._convert_audio(input_path)
# 加载模型
self._load_model()
# 执行识别
print("正在进行语音识别...")
try:
result = self._model.generate(
input=audio_path,
batch_size_s=300,
)
except Exception as e:
return {
"success": False,
"error": f"识别失败: {str(e)}"
}
# 解析结果
if not result or not isinstance(result, list):
return {
"success": False,
"error": "识别结果格式错误"
}
# FunASR 返回格式处理
segments = []
full_text = ""
for seg in result:
if isinstance(seg, dict):
text = seg.get("text", "").strip()
if "timestamp" in seg and seg["timestamp"]:
# 包含时间戳
timestamp = seg["timestamp"]
if len(timestamp) >= 2:
start_time = timestamp[0][0] if timestamp[0] else 0
end_time = timestamp[-1][1] if timestamp[-1] else start_time
else:
start_time = 0
end_time = 0
else:
start_time = 0
end_time = 0
if text:
segments.append({
"id": len(segments),
"start": round(start_time, 2),
"end": round(end_time, 2),
"text": text
})
full_text += text + "\n"
# 清理临时文件
if audio_path != input_path and "_converted.wav" in audio_path:
try:
os.remove(audio_path)
except Exception:
pass
# 计算时长
duration = segments[-1]["end"] if segments else 0
return {
"success": True,
"input_file": os.path.basename(input_path),
"model": self.model_name,
"output_files": {},
"duration_seconds": round(duration, 2),
"segments": segments,
"text": full_text.strip()
}
def save_outputs(self, result: Dict, output_dir: str):
"""保存输出文件
Args:
result: 识别结果
output_dir: 输出目录
"""
if not result.get("success"):
return
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
basename = Path(result["input_file"]).stem
# 保存 JSON
json_path = output_dir / f"{basename}.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=2)
print(f"已保存: {json_path}")
result["output_files"]["json"] = str(json_path)
# 保存 TXT
txt_path = output_dir / f"{basename}.txt"
with open(txt_path, "w", encoding="utf-8") as f:
f.write(result.get("text", ""))
print(f"已保存: {txt_path}")
result["output_files"]["txt"] = str(txt_path)
def run_as_background(input_path: str, model: str, output_dir: str):
"""后台运行识别
Args:
input_path: 输入文件路径
model: 模型名称
output_dir: 输出目录
"""
# 创建日志文件
log_file = os.path.join(output_dir, f"funasr_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log")
# 重新执行命令(后台模式)
cmd = [
sys.executable, __file__,
"--input", input_path,
"--model", model,
"--output", output_dir
]
with open(log_file, "w") as f:
process = subprocess.Popen(
cmd,
stdout=f,
stderr=subprocess.STDOUT,
start_new_session=True
)
print(f"\n✓ 后台任务已启动 (PID: {process.pid})")
print(f"日志文件: {log_file}")
print(f"输出目录: {output_dir}")
print("\n使用以下命令查看进度:")
print(f" tail -f {log_file}")
def main():
parser = argparse.ArgumentParser(
description="KB FunASR - 本地语音识别工具",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
python funasr.py --input video.mp4
python funasr.py --input podcast.mp3 --model paraformer-zh
python funasr.py --input video.mp4 --output ./transcripts --background
"""
)
parser.add_argument("--input", "-i", required=True, help="输入音视频文件路径")
parser.add_argument("--model", "-m", default="paraformer-zh",
choices=["paraformer-zh", "paraformer", "sensevoice"],
help="模型名称 (默认: paraformer-zh)")
parser.add_argument("--output", "-o", default=None, help="输出目录 (默认: 文件所在目录)")
parser.add_argument("--background", "-b", action="store_true", help="后台异步执行")
args = parser.parse_args()
# 确定输出目录
if args.output:
output_dir = args.output
else:
output_dir = os.path.dirname(args.input) or "."
# 后台模式
if args.background:
run_as_background(args.input, args.model, output_dir)
return
# 前台模式
transcriber = FunASRTranscriber(model=args.model)
result = transcriber.transcribe(args.input)
if result.get("success"):
transcriber.save_outputs(result, output_dir)
print(f"\n✓ 处理完成!")
print(f" 时长: {result['duration_seconds']} 秒")
print(f" 分段数: {len(result['segments'])}")
else:
print(f"\n✗ 处理失败: {result.get('error')}")
sys.exit(1)
if __name__ == "__main__":
main()
-43
View File
@@ -1,43 +0,0 @@
# KB Link Parser Skill
**kb-link-parser** - Extract and resolve IDs from various platform URLs.
## Features
- Multi-platform support: Bilibili, YouTube, Zhihu
- Short link detection: b23.tv and other short links
- Multiple content types: video, article, up主, playlist, collection
- Batch processing from file
- JSON output format
## Commands
```bash
# Parse single URL
python link-parser.py --url "https://www.bilibili.com/video/BV1xx411c7mD"
# Batch processing
python link-parser.py --batch urls.txt --output results.json
# Output format
python link-parser.py --url "..." --format json
```
## Output
```json
{
"success": true,
"data": {
"platform": "bilibili",
"type": "video",
"id": "BV1xx411c7mD"
}
}
```
## Installation
```bash
cp -r kb-link-parser/ ~/skills/
```
-637
View File
@@ -1,637 +0,0 @@
#!/usr/bin/env python3
"""
Link Parser - Extract and resolve IDs from various platform URLs.
Supports:
- Bilibili (video, article, up主, collection, series)
- YouTube (video, channel, playlist)
- Zhihu (question, answer, user, collection)
- Xiaohongshu (note, user)
- Short links (b23.tv, xhslink.com, etc.)
Usage:
python link-parser.py --url "https://www.bilibili.com/video/BV1xx411c7mD"
python link-parser.py --batch urls.txt --output results.json
"""
import argparse
import json
import re
import sys
from typing import Optional, Dict, Any
from urllib.parse import urlparse
from link_parser_shortlink import ShortLinkResolver
class LinkParser:
"""Main link parser class"""
# Bilibili patterns
BILIBILI_VIDEO_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/video/(BV[a-zA-Z0-9]+)')
BILIBILI_VIDEO_SHORT_PATTERN = re.compile(r'https?://b23\.tv/([a-zA-Z0-9]+)')
BILIBILI_ARTICLE_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/(?:article|read)/([a-zA-Z0-9]+)')
BILIBILI_SPACE_PATTERN = re.compile(r'https?://(?:space\.)?bilibili\.com/(\d+)')
BILIBILI_MOBILE_SPACE_PATTERN = re.compile(r'https?://m\.bilibili\.com/space/(\d+)')
BILIBILI_MOBILE_OPUS_PATTERN = re.compile(r'https?://m\.bilibili\.com/opus/(\d+)')
BILIBILI_FAVLIST_PATTERN = re.compile(r'https?://(?:space\.)?bilibili\.com/\d+/favlist\?fid=(\d+)')
BILIBILI_OPUS_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/opus/(\d+)')
BILIBILI_SERIES_PATTERN = re.compile(r'https?://(?:space\.)?bilibili\.com/\d+/lists/(\d+)\?type=series')
BILIBILI_READLIST_PATTERN = re.compile(r'https?://(?:www\.)?bilibili\.com/read/readlist/(rl[a-zA-Z0-9]+)')
# YouTube patterns
YOUTUBE_VIDEO_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?(?:youtube\.com/watch\?v=|youtu\.be/)([a-zA-Z0-9_-]{11})')
YOUTUBE_CHANNEL_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?youtube\.com/channel/([a-zA-Z0-9_-]+)')
YOUTUBE_USER_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?youtube\.com/user/([a-zA-Z0-9_-]+)')
YOUTUBE_PLAYLIST_PATTERN = re.compile(r'(?:https?://)?(?:www\.)?youtube\.com/playlist\?list=([a-zA-Z0-9_-]+)')
# Xiaohongshu (小红书) patterns
XIAOHONGSHU_SHORT_PATTERN = re.compile(r'https?://xhslink\.com/o/([a-zA-Z0-9]+)')
XIAOHONGSHU_NOTE_PATTERN = re.compile(r'https?://(?:www\.)?xiaohongshu\.com/explore/([a-zA-Z0-9]+)')
XIAOHONGSHU_NOTE_PATTERN2 = re.compile(r'https?://(?:www\.)?xiaohongshu\.com/notes/([a-zA-Z0-9]+)')
XIAOHONGSHU_DISCOVERY_PATTERN = re.compile(r'https?://(?:www\.)?xiaohongshu\.com/discovery/item/([a-zA-Z0-9]+)')
XIAOHONGSHU_USER_PATTERN = re.compile(r'https?://(?:www\.)?xiaohongshu\.com/user/profile/(\d+)')
# Zhihu patterns
ZHIHAU_QUESTION_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/question/(\d+)')
ZHIHAU_ANSWER_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)')
ZHIHAU_USER_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/people/([a-zA-Z0-9_-]+)')
ZHIHAU_COLUMN_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/column/([a-zA-Z0-9_-]+)')
ZHIHAU_COLLECTION_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/collection/(\d+)')
# Additional Zhihu patterns
ZHIHAU_ZHUANLAN_PATTERN = re.compile(r'https?://zhuanlan\.zhihu\.com/p/(\d+)')
ZHIHAU_ORG_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/org/([a-zA-Z0-9_-]+)')
ZHIHAU_ORG_ANSWERS_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/org/[a-zA-Z0-9_-]+/answers')
ZHIHAU_ORG_VIDEOS_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/org/[a-zA-Z0-9_-]+/zvideos')
ZHIHAU_ORG_ASKS_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/org/[a-zA-Z0-9_-]+/asks')
ZHIHAU_ORG_POSTS_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/org/[a-zA-Z0-9_-]+/posts')
ZHIHAU_ORG_PINS_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/org/[a-zA-Z0-9_-]+/pins')
ZHIHAU_ZVIDEO_PATTERN = re.compile(r'https?://(?:www\.)?zhihu\.com/zvideo/(\d+)')
def __init__(self):
self.shortlink_resolver = ShortLinkResolver()
def resolve_shortlink(self, url: str) -> Optional[str]:
"""
Resolve short link to get final URL.
Args:
url: Short URL to resolve
Returns:
Final URL after resolution, or None if not resolvable
"""
# Check if it's a known short link pattern
if self.shortlink_resolver.is_bilibili_short_link(url):
# Try to resolve using requests (follows redirects)
try:
import requests
response = requests.head(url, allow_redirects=True, timeout=10)
return response.url
except Exception:
# Fallback: return original URL
return url
# Check if it's Xiaohongshu short link
if self.XIAOHONGSHU_SHORT_PATTERN.match(url):
try:
import requests
response = requests.get(url, allow_redirects=True, timeout=10)
return response.url
except Exception:
return url
return url
def extract_bilibili_id(self, url: str) -> Optional[Dict[str, str]]:
"""
Extract Bilibili ID from URL.
Args:
url: Bilibili URL
Returns:
Dict with platform, type, and id fields, or None if not matched
"""
# Try series/playlist (must check before space pattern)
match = self.BILIBILI_SERIES_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "playlist",
"id": match.group(1)
}
# Try readlist (专栏合集)
match = self.BILIBILI_READLIST_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "readlist",
"id": match.group(1)
}
# Try favlist (收藏夹)
match = self.BILIBILI_FAVLIST_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "favlist",
"id": match.group(1)
}
# Try opus (动态)
match = self.BILIBILI_OPUS_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "opus",
"id": match.group(1)
}
# Try mobile opus (移动端动态)
match = self.BILIBILI_MOBILE_OPUS_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "opus",
"id": match.group(1)
}
# Try video
match = self.BILIBILI_VIDEO_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "video",
"id": match.group(1)
}
# Try article/read
match = self.BILIBILI_ARTICLE_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "article",
"id": match.group(1)
}
# Try space (UP主) - must check after specific patterns
match = self.BILIBILI_SPACE_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "up主",
"id": match.group(1)
}
# Try mobile space (移动端UP主空间)
match = self.BILIBILI_MOBILE_SPACE_PATTERN.match(url)
if match:
return {
"platform": "bilibili",
"type": "up主",
"id": match.group(1)
}
return None
def extract_youtube_id(self, url: str) -> Optional[Dict[str, str]]:
"""
Extract YouTube ID from URL.
Args:
url: YouTube URL
Returns:
Dict with platform, type, and id fields, or None if not matched
"""
# Try video
match = self.YOUTUBE_VIDEO_PATTERN.match(url)
if match:
return {
"platform": "youtube",
"type": "video",
"id": match.group(1)
}
# Try channel
match = self.YOUTUBE_CHANNEL_PATTERN.match(url)
if match:
return {
"platform": "youtube",
"type": "channel",
"id": match.group(1)
}
# Try user
match = self.YOUTUBE_USER_PATTERN.match(url)
if match:
return {
"platform": "youtube",
"type": "user",
"id": match.group(1)
}
# Try playlist
match = self.YOUTUBE_PLAYLIST_PATTERN.match(url)
if match:
return {
"platform": "youtube",
"type": "playlist",
"id": match.group(1)
}
return None
def extract_zhihu_id(self, url: str) -> Optional[Dict[str, str]]:
"""
Extract Zhihu ID from URL.
Args:
url: Zhihu URL
Returns:
Dict with platform, type, and id fields, or None if not matched
"""
# Try question
match = self.ZHIHAU_QUESTION_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "question",
"id": match.group(1)
}
# Try answer
match = self.ZHIHAU_ANSWER_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "answer",
"id": match.group(1)
}
# Try user
match = self.ZHIHAU_USER_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "user",
"id": match.group(1)
}
# Try column
match = self.ZHIHAU_COLUMN_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "column",
"id": match.group(1)
}
# Try collection
match = self.ZHIHAU_COLLECTION_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "collection",
"id": match.group(1)
}
# Try zhuanlan article (专栏文章)
match = self.ZHIHAU_ZHUANLAN_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "zhuanlan",
"id": match.group(1)
}
# Try org page (机构页面)
match = self.ZHIHAU_ORG_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "org",
"id": match.group(1)
}
# Try org answers (机构回答)
if self.ZHIHAU_ORG_ANSWERS_PATTERN.match(url):
match = self.ZHIHAU_ORG_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "org_answers",
"id": match.group(1)
}
# Try org zvideos (机构视频)
if self.ZHIHAU_ORG_VIDEOS_PATTERN.match(url):
match = self.ZHIHAU_ORG_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "org_zvideos",
"id": match.group(1)
}
# Try org asks (机构提问)
if self.ZHIHAU_ORG_ASKS_PATTERN.match(url):
match = self.ZHIHAU_ORG_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "org_asks",
"id": match.group(1)
}
# Try org posts (机构文章)
if self.ZHIHAU_ORG_POSTS_PATTERN.match(url):
match = self.ZHIHAU_ORG_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "org_posts",
"id": match.group(1)
}
# Try org pins (机构想法)
if self.ZHIHAU_ORG_PINS_PATTERN.match(url):
match = self.ZHIHAU_ORG_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "org_pins",
"id": match.group(1)
}
# Try zvideo (知乎视频)
match = self.ZHIHAU_ZVIDEO_PATTERN.match(url)
if match:
return {
"platform": "zhihu",
"type": "zvideo",
"id": match.group(1)
}
return None
def extract_xiaohongshu_id(self, url: str) -> Optional[Dict[str, str]]:
"""
Extract Xiaohongshu ID from URL.
Args:
url: Xiaohongshu URL
Returns:
Dict with platform, type, and id fields, or None if not matched
"""
# Try short link
match = self.XIAOHONGSHU_SHORT_PATTERN.match(url)
if match:
return {
"platform": "xiaohongshu",
"type": "note",
"id": match.group(1)
}
# Try explore link
match = self.XIAOHONGSHU_NOTE_PATTERN.match(url)
if match:
return {
"platform": "xiaohongshu",
"type": "note",
"id": match.group(1)
}
# Try notes link
match = self.XIAOHONGSHU_NOTE_PATTERN2.match(url)
if match:
return {
"platform": "xiaohongshu",
"type": "note",
"id": match.group(1)
}
# Try discovery item link (resolved from short link)
match = self.XIAOHONGSHU_DISCOVERY_PATTERN.match(url)
if match:
return {
"platform": "xiaohongshu",
"type": "note",
"id": match.group(1)
}
# Try user profile
match = self.XIAOHONGSHU_USER_PATTERN.match(url)
if match:
return {
"platform": "xiaohongshu",
"type": "user",
"id": match.group(1)
}
return None
def parse(self, url: str) -> Dict[str, Any]:
"""
Parse URL and extract platform-specific ID.
Args:
url: URL to parse
Returns:
Dict with platform, type, id, and original_url fields
"""
result = {
"platform": None,
"type": None,
"id": None,
"original_url": url,
"resolved_url": None,
"success": False,
"error": None
}
# First, resolve short links
resolved = self.resolve_shortlink(url)
if resolved:
result["resolved_url"] = resolved
url = resolved
# Try each platform
bilibili_result = self.extract_bilibili_id(url)
if bilibili_result:
result.update(bilibili_result)
result["success"] = True
return result
youtube_result = self.extract_youtube_id(url)
if youtube_result:
result.update(youtube_result)
result["success"] = True
return result
zhihu_result = self.extract_zhihu_id(url)
if zhihu_result:
result.update(zhihu_result)
result["success"] = True
return result
# Try Xiaohongshu
xiaohongshu_result = self.extract_xiaohongshu_id(url)
if xiaohongshu_result:
result.update(xiaohongshu_result)
result["success"] = True
return result
# Check for short links that need resolution
short_links = self.shortlink_resolver.detect_short_links(url)
if short_links:
result["error"] = "Short link detected. Please provide the full URL."
return result
result["error"] = "Unsupported URL format"
return result
def format_output(data: Dict[str, Any], output_format: str = "json") -> str:
"""Format output based on specified format."""
if output_format == "json":
return json.dumps(data, ensure_ascii=False, indent=2)
elif output_format == "compact":
return json.dumps(data, ensure_ascii=False, separators=(',', ':'))
elif output_format == "text":
lines = []
for key, value in data.items():
if value is not None:
lines.append(f"{key}: {value}")
return "\n".join(lines)
else:
return json.dumps(data, ensure_ascii=False, indent=2)
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(
description="Parse and extract IDs from various platform URLs",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
%(prog)s --url "https://www.bilibili.com/video/BV1xx411c7mD"
%(prog)s --url "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
%(prog)s --url "https://www.zhihu.com/question/266309957"
%(prog)s --batch urls.txt --output results.json
"""
)
parser.add_argument(
"--url", "-u",
help="Single URL to parse"
)
parser.add_argument(
"--batch", "-b",
help="File containing URLs to parse (one per line)"
)
parser.add_argument(
"--output", "-o",
help="Output file path (for batch mode)"
)
parser.add_argument(
"--format", "-f",
choices=["json", "compact", "text"],
default="json",
help="Output format (default: json)"
)
args = parser.parse_args()
# Check if we're in shortlink help mode
if len(sys.argv) == 2 and sys.argv[1] in ['--help', '-h']:
parser.print_help()
return 0
if not args.url and not args.batch:
parser.print_help()
print("\nError: Please provide --url or --batch")
return 1
link_parser = LinkParser()
if args.url:
# Single URL mode
parsed = link_parser.parse(args.url)
# 统一输出格式
if parsed["success"]:
output = {
"success": True,
"platform": parsed.get("platform"),
"type": parsed.get("type"),
"id": parsed.get("id"),
"filename": f"{parsed.get('platform')}_{parsed.get('id')}.link.json" if parsed.get("platform") and parsed.get("id") else None,
"message": "URL解析成功"
}
else:
output = {
"success": False,
"error": parsed.get("error", "解析失败"),
"detail": f"URL: {args.url}"
}
print(json.dumps(output, ensure_ascii=False, indent=2))
return 0 if output["success"] else 1
elif args.batch:
try:
with open(args.batch, 'r', encoding='utf-8') as f:
urls = [line.strip() for line in f if line.strip()]
except Exception as e:
print(json.dumps({
"success": False,
"error": "读取文件失败",
"detail": str(e)
}, ensure_ascii=False, indent=2))
return 1
results = []
for url in urls:
parsed = link_parser.parse(url)
if parsed["success"]:
results.append({
"success": True,
"platform": parsed.get("platform"),
"type": parsed.get("type"),
"id": parsed.get("id"),
"filename": f"{parsed.get('platform')}_{parsed.get('id')}.link.json" if parsed.get("platform") and parsed.get("id") else None
})
else:
results.append({
"success": False,
"error": parsed.get("error", "解析失败"),
"detail": f"URL: {url}"
})
if args.output:
with open(args.output, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(json.dumps({
"success": True,
"message": f"结果已保存到 {args.output}",
"total": len(results)
}, ensure_ascii=False, indent=2))
else:
for result in results:
print(json.dumps(result, ensure_ascii=False, indent=2))
print("-" * 40)
return 0
if __name__ == '__main__':
sys.exit(main())
-296
View File
@@ -1,296 +0,0 @@
#!/usr/bin/env python3
"""
Short link resolver for link-parser skill.
Provides utilities for detecting and resolving short links, particularly
Bilibili short links (b23.tv) which may be blocked by anti-scraping measures.
Supported short link services:
- Bilibili: b23.tv
- Generic URL shorteners (detection only)
"""
import re
from typing import Optional, Dict, List, Tuple
from urllib.parse import urlparse
class ShortLinkResolver:
"""Short link detection and resolution handler"""
# Pattern for Bilibili short links
BILIBILI_SHORT_PATTERN = re.compile(r'https?://b23\.tv/([a-zA-Z0-9]+)')
# Pattern for other short link services
OTHER_SHORT_PATTERNS = {
'bit.ly': re.compile(r'https?://bit\.ly/\w+'),
't.cn': re.compile(r'https?://t\.cn/\w+'),
'dwz.cn': re.compile(r'https?://dwz\.cn/\w+'),
'goo.gl': re.compile(r'https?://goo\.gl/\w+'),
'tinyurl.com': re.compile(r'https?://tinyurl\.com/\w+'),
}
@staticmethod
def detect_short_links(text: str) -> List[Dict[str, str]]:
"""
Detect short links in text.
Args:
text: Input text to search
Returns:
List of detected short links with metadata:
[
{
'url': 'https://b23.tv/xxxxx',
'platform': 'bilibili',
'is_short': True
},
...
]
"""
detected = []
# Check for Bilibili short links
for match in ShortLinkResolver.BILIBILI_SHORT_PATTERN.finditer(text):
detected.append({
'url': match.group(0),
'code': match.group(1),
'platform': 'bilibili',
'is_short': True
})
# Check for other short link services
for platform, pattern in ShortLinkResolver.OTHER_SHORT_PATTERNS.items():
for match in pattern.finditer(text):
detected.append({
'url': match.group(0),
'platform': platform,
'is_short': True
})
return detected
@staticmethod
def is_bilibili_short_link(url: str) -> bool:
"""
Check if URL is a Bilibili short link.
Args:
url: URL to check
Returns:
True if URL is a b23.tv short link
"""
return bool(ShortLinkResolver.BILIBILI_SHORT_PATTERN.match(url))
@staticmethod
def format_short_link_help(short_link: Optional[str] = None) -> str:
"""
Format help message for short links.
Args:
short_link: Optional short link URL to mention
Returns:
Help message
"""
lines = [
"Detection: Short link",
""
]
if short_link:
lines.append(f"Short link: {short_link}")
lines.append("")
lines.extend([
"Due to anti-scraping protection, short links cannot be automatically resolved.",
"Please provide the full URL or the content ID (e.g., BV号 for Bilibili).",
"",
"Full URL format: https://www.bilibili.com/video/BVxxxxx"
])
return "\n".join(lines)
@staticmethod
def format_short_link_warning(short_links: List[Dict[str, str]]) -> str:
"""
Format warning message for multiple short links.
Args:
short_links: List of short link metadata
Returns:
Warning message
"""
if not short_links:
return ""
lines = [
"Short links detected:",
""
]
for i, link in enumerate(short_links, 1):
platform = link.get('platform', 'unknown')
url = link['url']
if platform == 'bilibili':
lines.append(f" [{i}] Bilibili short link: {url}")
lines.append(" Full URL or BV ID required")
else:
lines.append(f" [{i}] {platform}: {url}")
lines.append("")
return "\n".join(lines)
class PlaywrightResolver:
"""
Browser-based short link resolver using Playwright.
This is an optional implementation for environments where Playwright
is available. It can bypass anti-scraping measures by simulating
a real browser.
Requirements:
pip install playwright
playwright install chromium
"""
def __init__(self, headless: bool = True, timeout: int = 10000):
"""
Initialize Playwright resolver.
Args:
headless: Whether to run browser in headless mode
timeout: Page navigation timeout in milliseconds
"""
self.headless = headless
self.timeout = timeout
self._playwright = None
def _check_playwright_available(self) -> bool:
"""Check if Playwright is installed and available"""
try:
from playwright.sync_api import sync_playwright
return True
except ImportError:
return False
def resolve(self, short_url: str) -> Optional[str]:
"""
Resolve short link using Playwright browser automation.
Args:
short_url: Short URL to resolve
Returns:
Final URL after redirects, or None if resolution fails
"""
if not self._check_playwright_available():
raise ImportError(
"Playwright not installed. Install with: pip install playwright && playwright install chromium"
)
from playwright.sync_api import sync_playwright
try:
with sync_playwright() as p:
# Launch browser
browser = p.chromium.launch(headless=self.headless)
context = browser.new_context(
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',
viewport={'width': 1920, 'height': 1080}
)
page = context.new_page()
# Navigate to short URL
page.goto(short_url, timeout=self.timeout, wait_until='networkidle')
# Get final URL
final_url = page.url
# Close browser
browser.close()
return final_url
except Exception as e:
print(f"Playwright resolution failed: {e}")
return None
def resolve_bilibili_short(self, short_url: str) -> Optional[str]:
"""
Resolve Bilibili short link and extract BV ID.
Args:
short_url: Bilibili short URL (e.g., https://b23.tv/xxxxx)
Returns:
Full Bilibili URL with BV ID, or None if resolution fails
"""
final_url = self.resolve(short_url)
if not final_url:
return None
# Check if final URL contains BV ID
if 'bilibili.com/video/' in final_url:
return final_url
# Extract BV ID if present in URL
bv_pattern = re.compile(r'(BV[a-zA-Z0-9]{10})')
match = bv_pattern.search(final_url)
if match:
return f"https://www.bilibili.com/video/{match.group(1)}"
return None
def get_resolver() -> ShortLinkResolver:
"""
Get the default resolver instance.
Returns:
ShortLinkResolver instance
"""
return ShortLinkResolver()
def detect_short_links(text: str) -> List[Dict[str, str]]:
"""
Detect short links in text.
Args:
text: Input text to search
Returns:
List of detected short links with metadata
"""
resolver = ShortLinkResolver()
return resolver.detect_short_links(text)
# CLI interface for testing
def main():
"""CLI for testing the short link resolver."""
import sys
if len(sys.argv) < 2:
print("Usage: python3 link_parser_shortlink.py 'text to test'")
print("\nExamples:")
print(' python3 link_parser_shortlink.py "Check this: https://b23.tv/oLBZZfK"')
print(' python3 link_parser_shortlink.py "https://bit.ly/xxxxx"')
sys.exit(1)
text = ' '.join(sys.argv[1:])
short_links = detect_short_links(text)
import json
print(json.dumps(short_links, ensure_ascii=False, indent=2))
if __name__ == '__main__':
main()
-66
View File
@@ -1,66 +0,0 @@
# KB OCR Processor Skill
**kb-ocr-processor** - Extract text from video frames using OCR.
## Features
- **视频截帧**: 直接从视频文件截取帧图片
- **批量处理**: 支持处理帧目录
- **双引擎**: EasyOCR / Tesseract
- **置信度评分**: 每帧识别结果带置信度
- **文本合并**: 自动合并所有帧的识别结果
## Commands
```bash
# 方式1: 直接从视频截帧并OCR(推荐)
python ocr_processor.py --video ./video.flv --output ocr.json
# 方式2: 处理已截取的帧目录
python ocr_processor.py --input ./frames/ --output ocr.json
# 指定截帧帧率(默认1帧/秒)
python ocr_processor.py --video ./video.flv --fps 2 --output ocr.json
# 同时生成合并的文本文件
python ocr_processor.py --video ./video.flv --output ocr.json --merge
# 使用 Tesseract 引擎
python ocr_processor.py --input ./frames/ --engine tesseract --output ocr.json
```
## Workflow 中的使用
```bash
# 步骤5: OCR处理(视频无字幕时)
python ocr_processor.py --video ${CONTENT_VIDEO_PATH} --output ${CONTENT_DIR}/ocr.json --merge
```
## Output
```json
{
"success": true,
"engine": "easyocr",
"total_frames": 619,
"results": [
{
"source": "0001.jpg",
"text": "字幕文字",
"confidence": 0.95
}
]
}
```
## Installation
```bash
cp -r kb-ocr-processor/ ~/skills/
pip install -r requirements.txt
# 依赖说明
# - easyocr: 主要OCR引擎(自动下载模型)
# - pytesseract: Tesseract引擎
# - ffmpeg: 视频截帧(apt install ffmpeg)
```
-360
View File
@@ -1,360 +0,0 @@
#!/usr/bin/env python3
"""
OCR Processor - 视频帧文字识别工具
支持 EasyOCR 和 Tesseract 引擎
"""
import argparse
import json
import os
import glob
from pathlib import Path
from typing import List, Dict, Optional
import easyocr
import pytesseract
from PIL import Image
class OCRProcessor:
"""OCR 文字识别处理器"""
def __init__(self, engine: str = "easyocr", languages: List[str] = ["ch_sim", "en"]):
"""
初始化 OCR 处理器
Args:
engine: OCR 引擎 ("easyocr" 或 "tesseract")
languages: 语言列表
"""
self.engine = engine
self.languages = languages
self.reader = None
if engine == "easyocr":
self.reader = easyocr.Reader(languages, gpu=False)
def process_image(self, image_path: str) -> Dict:
"""
处理单张图片进行 OCR 识别
Args:
image_path: 图片路径
Returns:
包含 text 和 confidence 的字典
"""
if not os.path.exists(image_path):
return {"text": "", "confidence": 0.0, "error": "File not found"}
try:
if self.engine == "easyocr":
return self._process_with_easyocr(image_path)
elif self.engine == "tesseract":
return self._process_with_tesseract(image_path)
else:
return {"text": "", "confidence": 0.0, "error": f"Unknown engine: {self.engine}"}
except Exception as e:
return {"text": "", "confidence": 0.0, "error": str(e)}
def _process_with_easyocr(self, image_path: str) -> Dict:
"""使用 EasyOCR 处理图片"""
results = self.reader.readtext(image_path)
if not results:
return {"text": "", "confidence": 0.0}
texts = []
confidences = []
for bbox, text, conf in results:
texts.append(text)
confidences.append(conf)
merged_text = "\n".join(texts)
avg_confidence = sum(confidences) / len(confidences) if confidences else 0.0
return {
"text": merged_text,
"confidence": round(avg_confidence, 4),
"details": [
{"text": text, "confidence": round(conf, 4)}
for text, conf in zip(texts, confidences)
]
}
def _process_with_tesseract(self, image_path: str) -> Dict:
"""使用 Tesseract 处理图片"""
image = Image.open(image_path)
text = pytesseract.image_to_string(image, lang="+".join(self.languages))
# Tesseract 不直接返回置信度,返回 0.0
return {
"text": text.strip(),
"confidence": 0.0
}
def process_frames(self, input_dir: str, fps: float = 0.1, progress_file: Optional[str] = None) -> List[Dict]:
"""
处理目录下所有帧图片
Args:
input_dir: 帧图片目录
fps: 采样帧率
progress_file: 可选,进度文件路径(会实时更新)
Returns:
OCR 结果列表
"""
if not os.path.isdir(input_dir):
print(f"Error: Directory not found: {input_dir}")
return []
# 支持的图片格式
image_extensions = ["*.jpg", "*.jpeg", "*.png", "*.bmp", "*.webp"]
image_files = []
for ext in image_extensions:
image_files.extend(glob.glob(os.path.join(input_dir, ext)))
image_files.extend(glob.glob(os.path.join(input_dir, "**", ext), recursive=True))
# 按文件名排序
image_files = sorted(image_files)
if not image_files:
print(f"No images found in {input_dir}")
return []
total = len(image_files)
print(f"Found {total} images in {input_dir}")
print(f"Processing with {self.engine} engine...")
def update_progress(current, status="running", last_result=None):
"""更新进度文件"""
if progress_file:
import json
import time
prog = {
"current": current,
"total": total,
"percent": round(current / total * 100, 1) if total > 0 else 0,
"status": status,
"last_result": last_result,
"updated_at": time.strftime("%Y-%m-%d %H:%M:%S")
}
with open(progress_file, 'w', encoding='utf-8') as f:
json.dump(prog, f, ensure_ascii=False, indent=2)
# 初始化进度
update_progress(0, "started")
results = []
for i, image_path in enumerate(image_files, 1):
basename = os.path.basename(image_path)
print(f"Processing [{i}/{total}]: {basename}")
ocr_result = self.process_image(image_path)
ocr_result["source"] = basename
# 记录结果摘要
text_count = len(ocr_result.get("text", "").split()) if ocr_result.get("text") else 0
last_result = f"✓ {text_count} 处文字" if ocr_result.get("text") else "✗ 无文字"
results.append(ocr_result)
# 实时更新进度
update_progress(i, "running", last_result)
# 完成
update_progress(total, "completed", f"完成 {total} 张")
print(f"\n✅ Processed {total} frames")
return results
def merge_texts(self, results: List[Dict]) -> str:
"""
合并多个 OCR 结果的文本
Args:
results: OCR 结果列表
Returns:
合并后的文本
"""
texts = []
for result in results:
if result.get("text"):
texts.append(result["text"])
return "\n\n---\n\n".join(texts)
def save_results(self, results: List[Dict], output_path: str, merge: bool = False):
"""
保存 OCR 结果到 JSON 文件
Args:
results: OCR 结果列表
output_path: 输出文件路径
merge: 是否同时保存合并后的文本文件
"""
output_data = {
"engine": self.engine,
"total_frames": len(results),
"results": results
}
# 保存 JSON
with open(output_path, "w", encoding="utf-8") as f:
json.dump(output_data, f, ensure_ascii=False, indent=2)
print(f"Results saved to {output_path}")
# 如果需要,同时保存合并的文本
if merge:
text_path = output_path.replace(".json", ".txt")
merged_text = self.merge_texts(results)
with open(text_path, "w", encoding="utf-8") as f:
f.write(merged_text)
print(f"Merged text saved to {text_path}")
def extract_frames(video_path: str, output_dir: str, fps: float = 1.0) -> str:
"""
使用 ffmpeg 从视频中截取帧图片
Args:
video_path: 视频文件路径
output_dir: 输出目录
fps: 每秒截取帧数 (默认: 1)
Returns:
帧图片目录路径
"""
import subprocess
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
# ffmpeg 命令: -i 输入 -vf fps=fps 输出格式
cmd = [
"ffmpeg", "-y",
"-i", video_path,
"-vf", f"fps={fps}",
"-q:v", "2", # 质量等级 (2=高质量)
str(output_dir / "%04d.jpg")
]
print(f"🔄 截取视频帧: {video_path}")
print(f" 输出目录: {output_dir}")
try:
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
# 统计截取的帧数
frame_count = len(list(output_dir.glob("*.jpg")))
print(f"✅ 截取完成: {frame_count} 帧")
return str(output_dir)
else:
print(f"❌ ffmpeg 错误: {result.stderr}")
return ""
except FileNotFoundError:
print("❌ 错误: 未找到 ffmpeg,请先安装: apt install ffmpeg")
return ""
def main():
"""命令行入口"""
parser = argparse.ArgumentParser(
description="OCR 视频帧文字识别工具",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
# 使用 EasyOCR 处理帧目录
python ocr_processor.py --input ./frames/ --output ./ocr.json
# 直接从视频截帧并 OCR
python ocr_processor.py --video ./video.flv --output ./ocr.json
# 使用 Tesseract 处理
python ocr_processor.py --input ./frames/ --output ./ocr.json --engine tesseract
# 指定截帧帧率
python ocr_processor.py --video ./video.flv --output ./ocr.json --fps 2
# 同时生成合并的文本文件
python ocr_processor.py --input ./frames/ --output ./ocr.json --merge
"""
)
# 输入源:帧目录 或 视频文件(二选一)
input_group = parser.add_mutually_exclusive_group(required=True)
input_group.add_argument("--input", "-i", help="输入帧图片目录")
input_group.add_argument("--video", "-v", help="输入视频文件(会自动截帧)")
parser.add_argument("--output", "-o", required=True, help="输出 JSON 文件路径")
parser.add_argument("--fps", "-f", type=float, default=1.0, help="截帧帧率 (默认: 1帧/秒)")
parser.add_argument("--engine", "-e", choices=["easyocr", "tesseract"],
default="easyocr", help="OCR 引擎 (默认: easyocr)")
parser.add_argument("--merge", action="store_true", help="同时保存合并的文本文件")
parser.add_argument("--languages", "-l", nargs="+",
default=["ch_sim", "en"], help="语言列表")
parser.add_argument("--progress", "-p", help="可选:进度文件路径(会实时更新)")
args = parser.parse_args()
# 如果指定了视频文件,先截帧
input_dir = args.input
if args.video:
if not os.path.exists(args.video):
print(json.dumps({
"success": False,
"error": f"视频文件不存在: {args.video}"
}, ensure_ascii=False, indent=2))
return 1
# 创建临时帧目录
frames_dir = Path(args.video).stem + "_frames"
input_dir = extract_frames(args.video, frames_dir, fps=args.fps)
if not input_dir:
print(json.dumps({
"success": False,
"error": "截帧失败"
}, ensure_ascii=False, indent=2))
return 1
# 初始化处理器
processor = OCRProcessor(engine=args.engine, languages=args.languages)
# 处理帧
results = processor.process_frames(input_dir, fps=args.fps, progress_file=args.progress)
if results:
# 保存结果
processor.save_results(results, args.output, merge=args.merge)
# 统一输出格式
total_texts = sum(1 for r in results if r.get("text"))
avg_conf = sum(r.get("confidence", 0) for r in results) / len(results) if results else 0
output = {
"success": True,
"filename": Path(args.output).name,
"filepath": str(Path(args.output).resolve()),
"message": f"OCR处理完成",
"stats": {
"total_frames": len(results),
"frames_with_text": total_texts,
"avg_confidence": round(avg_conf, 4)
}
}
print(json.dumps(output, ensure_ascii=False, indent=2))
return 0
else:
print(json.dumps({
"success": False,
"error": "未处理到任何结果"
}, ensure_ascii=False, indent=2))
return 1
if __name__ == "__main__":
exit(main())
-3
View File
@@ -1,3 +0,0 @@
easyocr>=1.7.0
pytesseract>=0.3.10
Pillow>=9.0.0
-9
View File
@@ -1,9 +0,0 @@
# KB Whisper Diarization 环境变量
# Whisper.cpp 安装路径
WHISPER_CPP_PATH=~/whisper.cpp
# HuggingFace Token(必需,用于下载 pyannote 模型)
# 获取方式: https://huggingface.co/settings/tokens
# 注意: 需要先申请 pyannote/speaker-diarization 模型访问权限
HF_TOKEN=hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
-151
View File
@@ -1,151 +0,0 @@
# KB Whisper Diarization Skill
**kb-whisper-diarization** - 使用 Whisper + pyannote.audio 实现语音识别和说话者分离。
## 什么时候使用
当用户需要:
- 区分多个说话者(如访谈、对话)
- 获取带说话者标签的字幕
- 分析对话中谁说了什么
## 环境变量
在使用之前,必须检查环境变量是否已经设置。
| 变量名 | 用途 | 示例 |
|--------|------|------|
| `WHISPER_CPP_PATH` | whisper.cpp 安装路径 | `~/whisper.cpp` |
| `HF_TOKEN` | HuggingFace Token | `hf_xxx...` |
### 获取 HuggingFace Token(必需)
pyannote.audio 需要从 HuggingFace 下载模型,按以下步骤操作:
**步骤 1: 注册/登录 HuggingFace**
- 访问 https://huggingface.co
- 注册或登录账号
**步骤 2: 申请模型访问权限**
需要申请以下两个模型的访问权限(点击页面上的 "Access repository" 按钮):
- https://huggingface.co/pyannote/speaker-diarization
- https://huggingface.co/pyannote/segmentation
> ⚠️ 申请后需要等待几分钟到几小时才能获得访问权限
**步骤 3: 创建 Access Token**
1. 访问 https://huggingface.co/settings/tokens
2. 点击 "New token"
3. 选择 "Read" 权限
4. 复制生成的 token(以 `hf_` 开头)
**步骤 4: 设置环境变量**
```bash
# 添加到 ~/.kb_env 文件
echo 'export HF_TOKEN="hf_xxxxxxxxxxxxxxxxxxxxxx"' >> ~/.kb_env
source ~/.kb_env
# 验证
echo $HF_TOKEN
```
## 使用方法
```bash
# 基本用法(带说话者分离)
python whisper_diarization.py --input video.mp4
# 说话者分离
python whisper_diarization.py --input interview.mp4 --diarize
# 指定说话者数量
python whisper_diarization.py --input meeting.mp4 --diarize --num-speakers 2
# 指定语言
python whisper_diarization.py --input video.mp4 --language zh
```
## 命令参数
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `--input` | 输入音视频文件路径 | 必需 |
| `--output` | 输出目录 | 文件所在目录 |
| `--diarize` | 启用说话者分离 | False |
| `--num-speakers` | 估计说话者数量 | None(自动检测) |
| `--language` | 语言 | 自动检测 |
| `--model` | Whisper 模型 | base |
| `--background` | 后台异步执行 | False |
## 模型选项
- **Whisper 模型**: tiny, base, small, medium, large
- **说话者分离**: pyannote/speaker-diarization (需申请权限)
## 输出文件
对于输入 `interview.mp4`,将生成:
- `interview.mp4.json` - 完整结果(包含说话者标签)
- `interview.mp4.txt` - 纯文本
- `interview.mp4.srt` - 字幕格式(带说话者标签)
## 输出格式示例
### JSON 格式
```json
{
"success": true,
"input_file": "interview.mp4",
"diarization": true,
"speakers": ["SPEAKER_00", "SPEAKER_01"],
"segments": [
{
"id": 0,
"speaker": "SPEAKER_00",
"start": 0.0,
"end": 5.2,
"text": "大家好,欢迎来到访谈节目"
},
{
"id": 1,
"speaker": "SPEAKER_01",
"start": 5.2,
"end": 10.8,
"text": "谢谢邀请,我很荣幸"
}
]
}
```
### SRT 字幕格式
```
1
00:00:00,000 --> 00:00:05,200
[SPEAKER_00] 大家好,欢迎来到访谈节目
2
00:00:05,200 --> 00:00:10,800
[SPEAKER_01] 谢谢邀请,我很荣幸
```
## 安装
```bash
# 1. 安装依赖
pip install pyannote.audio torch
# 2. 设置 HuggingFace Token
export HF_TOKEN="hf_xxx..."
# 3. 首次运行会自动下载模型
python whisper_diarization.py --input test.mp4
```
## 注意事项
- **说话者分离需要申请权限**: https://huggingface.co/pyannote/speaker-diarization
- **处理时间较长**: 说话者分离会增加处理时间
- **对立体声效果更好**: 多声道音频的分离效果更佳
-8
View File
@@ -1,8 +0,0 @@
# KB Whisper Diarization 依赖
# 核心依赖
torch>=2.0.0
pyannote.audio>=1.0.0
# 可选: GPU 加速
# torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
@@ -1,339 +0,0 @@
#!/usr/bin/env python3
"""
KB Whisper Diarization - 语音识别 + 说话者分离
使用 Whisper 进行语音识别,pyannote.audio 进行说话者分离
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
import re
def parse_arguments():
"""解析命令行参数"""
parser = argparse.ArgumentParser(
description='KB Whisper Diarization - 语音识别 + 说话者分离',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
python whisper_diarization.py --input video.mp4
python whisper_diarization.py --input interview.mp4 --diarize
python whisper_diarization.py --input meeting.mp4 --diarize --num-speakers 2
"""
)
parser.add_argument('--input', required=True, help='输入音视频文件路径')
parser.add_argument('--output', help='输出目录 (默认: 文件所在目录)')
parser.add_argument('--diarize', action='store_true', help='启用说话者分离')
parser.add_argument('--num-speakers', type=int, help='估计说话者数量')
parser.add_argument('--language', help='指定语言 (默认: 自动检测)')
parser.add_argument('--model', default='base', choices=['tiny', 'base', 'small', 'medium', 'large'],
help='Whisper 模型 (默认: base)')
parser.add_argument('--background', action='store_true', help='后台异步执行')
return parser.parse_args()
def get_whisper_path():
"""获取 whisper.cpp 路径"""
path = os.environ.get('WHISPER_CPP_PATH')
if path:
path = os.path.expanduser(path)
if os.path.exists(path):
return path
return os.path.expanduser('~/whisper.cpp')
def check_whisper_model(whisper_path, model):
"""检查 Whisper 模型"""
model_path = os.path.join(whisper_path, 'models', f'ggml-{model}.bin')
if not os.path.exists(model_path):
raise FileNotFoundError(f"Whisper 模型不存在: {model_path}")
return model_path
def convert_to_wav(input_path, output_dir):
"""转换为 16kHz WAV"""
if input_path.lower().endswith('.wav'):
try:
result = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries', 'stream=sample_rate',
'-of', 'default=noprint_wrappers=1:nokey=1', input_path],
capture_output=True, text=True
)
if result.returncode == 0 and int(result.stdout.strip()) == 16000:
return input_path
except Exception:
pass
output_path = os.path.join(output_dir, f"{Path(input_path).stem}_audio.wav")
print(f"转换音频: {input_path} -> {output_path}")
subprocess.run([
'ffmpeg', '-y', '-i', input_path,
'-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1',
'-loglevel', 'error', output_path
], check=True, capture_output=True)
print("✓ 音频转换完成")
return output_path
def run_whisper(whisper_path, audio_path, model, output_dir, language=None):
"""运行 Whisper 识别"""
basename = Path(audio_path).stem.replace('_audio', '')
output_file = os.path.join(output_dir, basename)
cmd = [
os.path.join(whisper_path, 'whisper-cli'),
'-m', os.path.join(whisper_path, 'models', f'ggml-{model}.bin'),
'-f', audio_path,
'-of', output_file,
'-l', language or 'auto',
'--output-txt', '--output-srt', '--output-json',
]
print(f"执行 Whisper: {' '.join(cmd[:6])} ...")
result = subprocess.run(
cmd, capture_output=True, text=True, cwd=whisper_path, timeout=3600
)
if result.returncode != 0 and not os.path.exists(f"{output_file}.txt"):
raise RuntimeError(f"Whisper 失败: {result.stderr}")
return basename
def run_diarization(audio_path, num_speakers=None):
"""运行 pyannote.audio 说话者分离"""
print("\n🎙️ 步骤2: 说话者分离...")
try:
import torch
from pyannote.audio import Pipeline
except ImportError as e:
print(f"✗ 缺少依赖: {e}")
print("请安装: pip install pyannote.audio torch")
raise
# 检查 HuggingFace Token
hf_token = os.environ.get('HF_TOKEN')
if not hf_token:
raise RuntimeError("需要设置 HF_TOKEN 环境变量\n"
"获取方式: https://huggingface.co/settings/tokens")
# 新版 pyannote 需要先登录
from huggingface_hub import login
print("登录 HuggingFace...")
login(token=hf_token)
print("加载 pyannote 流水线...")
pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization")
# 设置设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")
pipeline.to(device)
# 运行分离
print("执行说话者分离...")
diarization = pipeline(
audio_path,
num_speakers=num_speakers
)
# 解析结果
speakers = {}
for turn, _, speaker in diarization.itertracks(yield_label=True):
if speaker not in speakers:
speakers[speaker] = []
speakers[speaker].append({
"start": round(turn.start, 2),
"end": round(turn.end, 2)
})
print(f"✓ 检测到 {len(speakers)} 个说话者: {list(speakers.keys())}")
return speakers
def merge_results(whisper_txt, whisper_srt, diarization, output_dir, basename):
"""合并 Whisper 和说话者分离结果"""
print("\n📝 步骤3: 合并结果...")
# 读取 Whisper 文本
text_segments = []
if os.path.exists(whisper_txt):
with open(whisper_txt, 'r', encoding='utf-8') as f:
content = f.read()
# 简单按行分段
for i, line in enumerate(content.strip().split('\n')):
if line.strip():
text_segments.append({
"id": i,
"text": line.strip(),
"start": i * 3.0, # 估算
"end": (i + 1) * 3.0
})
# 读取 SRT 获取精确时间
srt_segments = []
if os.path.exists(whisper_srt):
with open(whisper_srt, 'r', encoding='utf-8') as f:
content = f.read()
# 解析 SRT
pattern = r'(\d+)\n(\d{2}:\d{2}:\d{2},\d{3}) --> (\d{2}:\d{2}:\d{2},\d{3})\n(.+?)(?=\n\n|\Z)'
matches = re.findall(pattern, content, re.DOTALL)
def parse_time(t):
h, m, s_ms = t.split(':')
s, ms = s_ms.split(',')
return float(h) * 3600 + float(m) * 60 + float(s) + float(ms) / 1000
for idx, num, start, end, text in matches:
srt_segments.append({
"id": int(idx),
"start": parse_time(start),
"end": parse_time(end),
"text": text.strip()
})
# 为每个 Whisper 段落分配说话者
merged_segments = []
for seg in srt_segments:
speaker = assign_speaker(seg['start'], seg['end'], diarization)
merged_segments.append({
"id": len(merged_segments),
"speaker": speaker,
"start": seg['start'],
"end": seg['end'],
"text": seg['text']
})
# 生成带说话者的 SRT
output_srt = os.path.join(output_dir, f"{basename}_diarized.srt")
with open(output_srt, 'w', encoding='utf-8') as f:
for seg in merged_segments:
start = format_time(seg['start'])
end = format_time(seg['end'])
f.write(f"{seg['id'] + 1}\n")
f.write(f"{start} --> {end}\n")
f.write(f"[{seg['speaker']}] {seg['text']}\n\n")
# 生成合并的 TXT
output_txt = os.path.join(output_dir, f"{basename}_diarized.txt")
with open(output_txt, 'w', encoding='utf-8') as f:
for seg in merged_segments:
f.write(f"[{seg['speaker']}] {seg['text']}\n")
# 生成 JSON
output_json = os.path.join(output_dir, f"{basename}_diarized.json")
json_data = {
"success": True,
"tool": "whisper-diarization",
"diarization": True,
"speakers": list(diarization.keys()),
"input_file": basename,
"output_files": {
"txt": f"{basename}_diarized.txt",
"srt": f"{basename}_diarized.srt",
"json": f"{basename}_diarized.json"
},
"segments": merged_segments
}
with open(output_json, 'w', encoding='utf-8') as f:
json.dump(json_data, f, ensure_ascii=False, indent=2)
print(f"✓ 生成文件:")
print(f" - {output_srt}")
print(f" - {output_txt}")
print(f" - {output_json}")
return json_data
def assign_speaker(start, end, diarization):
"""将时间段分配给说话者"""
best_speaker = "UNKNOWN"
max_overlap = 0
for speaker, segments in diarization.items():
for seg in segments:
overlap = min(end, seg['end']) - max(start, seg['start'])
if overlap > max_overlap:
max_overlap = overlap
best_speaker = speaker
return best_speaker
def format_time(seconds):
"""秒数转换为 SRT 时间格式"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
ms = int((seconds % 1) * 1000)
return f"{hours:02d}:{minutes:02d}:{secs:02d},{ms:03d}"
def main():
args = parse_arguments()
input_file = os.path.abspath(args.input)
if not os.path.exists(input_file):
print(f"✗ 文件不存在: {input_file}")
sys.exit(1)
output_dir = os.path.abspath(args.output) if args.output else os.path.dirname(input_file)
os.makedirs(output_dir, exist_ok=True)
basename = Path(input_file).stem
print(f"\n📁 输入: {input_file}")
print(f"📁 输出: {output_dir}")
print(f"🎙️ 说话者分离: {'是' if args.diarize else '否'}")
try:
# 步骤1: Whisper 识别
print(f"\n🎙️ 步骤1: 语音识别...")
whisper_path = get_whisper_path()
check_whisper_model(whisper_path, args.model)
audio_path = convert_to_wav(input_file, output_dir)
run_whisper(whisper_path, audio_path, args.model, output_dir, args.language)
# 步骤2: 说话者分离
if args.diarize:
diarization = run_diarization(audio_path, args.num_speakers)
else:
diarization = {}
# 步骤3: 合并结果
whisper_txt = os.path.join(output_dir, f"{basename}.txt")
whisper_srt = os.path.join(output_dir, f"{basename}.srt")
result = merge_results(whisper_txt, whisper_srt, diarization, output_dir, basename)
# 清理临时音频
if audio_path != input_file and audio_path.endswith('_audio.wav'):
try:
os.remove(audio_path)
except Exception:
pass
print(f"\n✅ 处理完成!")
print(f" 说话者数量: {len(diarization) if args.diarize else 1}")
print(f" 段落数量: {len(result['segments'])}")
except Exception as e:
print(f"\n✗ 错误: {e}")
sys.exit(1)
if __name__ == '__main__':
main()
-5
View File
@@ -1,5 +0,0 @@
# KB Whisper 环境变量配置示例
# 复制此文件为 .env 并填写实际值
# Whisper.cpp 安装路径
WHISPER_CPP_PATH=~/whisper.cpp
-211
View File
@@ -1,211 +0,0 @@
# KB Whisper - 本地语音识别
基于 Whisper.cpp 的本地语音识别工具,支持将音视频文件转换为文本、字幕和 JSON 格式。
## 功能特性
- 🎯 **本地执行** - 完全离线运行,保护隐私
- 🚀 **多模型支持** - 从 tiny 到 large,根据需求选择速度/精度
- 📝 **多格式输出** - JSON、SRT、TXT 三种格式
- ⚡ **异步执行** - 支持后台运行,不阻塞主代理
- 🎬 **广泛兼容** - 支持常见音视频格式(MP3、MP4、WAV 等)
## 安装前提
### 1. 安装 whisper.cpp
```bash
# 克隆仓库
git clone https://github.com/ggerganov/whisper.cpp.git ~/whisper.cpp
# 编译
cd ~/whisper.cpp
make
# 下载模型(至少需要一个模型)
./models/download-ggml-model.sh tiny
./models/download-ggml-model.sh base # 可选
```
### 2. 配置环境变量
在 `~/.clawdbot/clawdbot.json` 中添加:
```json5
{
skills: {
entries: {
"kb-whisper": {
env: {
WHISPER_CPP_PATH: "~/whisper.cpp"
}
}
}
}
}
```
或通过命令行配置:
```bash
clawdbot skills config kb-whisper WHISPER_CPP_PATH "~/whisper.cpp"
```
## 使用方法
### 基本用法
```bash
# 使用默认 tiny 模型处理
python whisper.py --input video.mp4
# 使用 base 模型(推荐,平衡速度和精度)
python whisper.py --input video.mp4 --model base
# 后台异步执行
python whisper.py --input video.mp4 --background
```
### 指定输出目录
```bash
python whisper.py --input video.mp4 --output ./transcripts
```
### 处理音频文件
```bash
python whisper.py --input podcast.mp3 --model base
```
## 输出文件
对于输入 `video.mp4`,将生成以下文件:
```
output/
├── video.json # 详细结果(含时间戳)
├── video.srt # 字幕格式
└── video.txt # 纯文本
```
### JSON 格式
```json
{
"success": true,
"input_file": "video.mp4",
"model": "base",
"output_files": {
"json": "video.json",
"srt": "video.srt",
"txt": "video.txt"
},
"duration_seconds": 120.5,
"segments": [
{
"id": 0,
"start": 0.0,
"end": 3.2,
"text": "这是第一段字幕"
},
{
"id": 1,
"start": 3.2,
"end": 6.8,
"text": "这是第二段字幕"
}
]
}
```
## 模型对比
| 模型 | 速度 | 精度 | 适用场景 |
|------|------|------|----------|
| tiny | ⚡⚡⚡ | ⭐ | 快速预览、实时转录 |
| base | ⚡⚡ | ⭐⭐⭐ | 日常使用(推荐) |
| small | ⚡ | ⭐⭐⭐⭐ | 高质量需求 |
| medium | ⭐ | ⭐⭐⭐⭐⭐ | 专业用途 |
| large | 🐢 | ⭐⭐⭐⭐⭐ | 最高精度要求 |
## 完整示例
### 示例 1: 转录播客
```bash
python whisper.py --input podcast.mp3 --model base --background
```
### 示例 2: 生成视频字幕
```bash
python whisper.py --input tutorial.mp4 --model small
```
### 示例 3: 批量处理(配合脚本)
```bash
#!/bin/bash
for video in videos/*.mp4; do
python whisper.py --input "$video" --model base
done
```
## 在 Workflow 中使用
在知识库处理流程中,如果视频没有字幕:
```bash
# 步骤4: 下载视频后,如果没有字幕
# 步骤5: 使用 Whisper 生成字幕
python whisper.py --input ${video_path} --model base --background
# 等待 Whisper 完成,然后继续后续步骤
```
## 技术细节
- **引擎**: Whisper.cpp(C++ 实现,性能优异)
- **音频采样率**: 自动转换为 16kHz
- **语言**: 自动检测(支持 99 种语言)
- **GPU 支持**: whisper.cpp 支持 CUDA 加速(需编译时启用)
## 故障排查
### 问题: 找不到 whisper.cpp
```bash
# 检查环境变量
echo $WHISPER_CPP_PATH
# 手动测试
~/whisper.cpp/main --help
```
### 问题: 模型未下载
```bash
cd ~/whisper.cpp
./models/download-ggml-model.sh tiny
```
### 问题: 音频格式不支持
```bash
# 转换音频格式
ffmpeg -i input.avi -acodec libmp3lame output.mp3
python whisper.py --input output.mp3
```
## 性能参考
在 CPU 环境下(Intel i7):
- **tiny 模型**: ~1x 实时速度(1分钟音频 ~1分钟处理)
- **base 模型**: ~0.5x 实时速度(1分钟音频 ~2分钟处理)
- **small 模型**: ~0.25x 实时速度(1分钟音频 ~4分钟处理)
## 许可证
本技能使用 MIT 许可证,Whisper.cpp 使用 MIT 许可证。
-124
View File
@@ -1,124 +0,0 @@
---
name: kb-whisper
description: 本地语音识别,支持音视频转文字(基于 Whisper.cpp)
---
# KB Whisper Skill
**kb-whisper** - 本地语音识别,使用 Whisper.cpp 将音视频文件转换为文本、字幕和 JSON 格式输出。
## 什么时候使用
当用户需要:
- 将视频/音频文件转换为文字
- 生成视频字幕(SRT 格式)
- 获取带时间戳的详细识别结果
## 环境变量
在使用之前,必须检查环境变量是否已经设置。
| 变量名 | 用途 | 示例 |
|--------|------|------|
| `WHISPER_CPP_PATH` | whisper.cpp 安装路径 | `~/whisper.cpp` |
## 使用方法
```bash
# 基本用法
python whisper.py --input video.mp4
# 指定模型(默认 tiny)
python whisper.py --input video.mp4 --model base
# 指定输出目录
python whisper.py --input video.mp4 --output ./output
# 异步执行(推荐)
python whisper.py --input video.mp4 --background
```
## 命令参数
| 参数 | 说明 | 默认值 |
|------|------|--------|
| `--input` | 输入音视频文件路径 | 必需 |
| `--model` | Whisper 模型类型 | `tiny` |
| `--output` | 输出目录 | 文件所在目录 |
| `--background` | 后台异步执行 | False |
## 模型选项
- `tiny` - 最小模型,速度快,精度较低(推荐快速识别)
- `base` - 基础模型,平衡速度和精度(推荐)
- `small` - 小型模型,精度更高,速度较慢
- `medium` - 中型模型,高精度
- `large` - 大型模型,最高精度,速度慢
## 输出文件
对于输入 `video.mp4`,将生成:
- `video.json` - 详细结果(包含时间戳、分段信息)
- `video.srt` - 字幕格式文件
- `video.txt` - 纯文本内容
## 输出格式示例
### JSON 格式
```json
{
"success": true,
"input_file": "video.mp4",
"model": "tiny",
"output_files": {
"json": "video.json",
"srt": "video.srt",
"txt": "video.txt"
},
"duration_seconds": 120.5,
"segments": [
{
"id": 0,
"start": 0.0,
"end": 3.2,
"text": "这是第一段字幕"
}
]
}
```
## Workflow 中的使用
```bash
# 步骤4之后:如果已有音频但无字幕,可以使用 whisper
python whisper.py --input ${video_path} --model base --background
```
## 安装
```bash
# 1. 克隆 whisper.cpp
git clone https://github.com/ggerganov/whisper.cpp.git ~/whisper.cpp
# 2. 编译
cd ~/whisper.cpp
make
# 3. 下载模型(tiny)
./models/download-ggml-model.sh tiny
# 4. 下载更多模型(可选)
# ./models/download-ggml-model.sh base
# ./models/download-ggml-model.sh small
# 5. 复制技能到 skills 目录
cp -r kb-whisper/ ~/skills/
```
## 注意事项
- **异步执行**: 使用 `--background` 参数时,命令会在后台运行,不阻塞主代理
- **模型选择**: tiny 模型适合快速识别,base 模型适合一般用途,large 模型适合高精度需求
- **文件格式**: 支持 MP3、MP4、WAV、FLAC、OGG、M4A 等常见格式
Binary file not shown.
-273
View File
@@ -1,273 +0,0 @@
# KB Whisper 使用示例
## 快速开始
### 1. 基础使用
```bash
# 使用默认 tiny 模型转录视频
python whisper.py --input video.mp4
# 使用 base 模型(推荐)
python whisper.py --input video.mp4 --model base
```
### 2. 异步执行
```bash
# 后台处理,不阻塞主代理
python whisper.py --input video.mp4 --model base --background
```
### 3. 指定输出目录
```bash
python whisper.py --input video.mp4 --output ./transcripts
```
### 4. 指定语言
```bash
# 指定中文
python whisper.py --input video.mp4 --language zh
# 指定英文
python whisper.py --input video.mp4 --language en
# 指定日语
python whisper.py --input video.mp4 --language ja
```
## 实际应用场景
### 场景 1: 转录播客
```bash
python whisper.py --input podcast.mp3 --model base --background
```
**输出**:
- `podcast.json` - 带时间戳的详细结果
- `podcast.srt` - 字幕格式
- `podcast.txt` - 纯文本
### 场景 2: 生成视频字幕
```bash
python whisper.py --input tutorial.mp4 --model small
```
**用途**:
- 为教程视频生成字幕
- 导出 SRT 文件到视频编辑软件
- 生成多语言字幕版本
### 场景 3: 会议录音转文字
```bash
python whisper.py --input meeting.mp3 --model base --output ./meetings
```
**输出示例**:
```
meetings/
├── meeting.json # 完整 JSON(含时间戳)
├── meeting.srt # 字幕格式
└── meeting.txt # 纯文本
```
### 场景 4: 批量处理
创建脚本 `batch_transcribe.sh`:
```bash
#!/bin/bash
OUTPUT_DIR="./transcripts"
mkdir -p "$OUTPUT_DIR"
for video in videos/*.mp4; do
echo "处理: $video"
python whisper.py --input "$video" --output "$OUTPUT_DIR" --model base
done
echo "批量处理完成!"
```
```bash
chmod +x batch_transcribe.sh
./batch_transcribe.sh
```
## 在 Workflow 中使用
### 知识库处理流程
在 `kb-workflow` 技能中,如果视频没有字幕:
```bash
# 步骤 4: 下载视频后
# 检查是否有字幕文件
if [ ! -f "${subtitle_file}" ]; then
echo "无字幕文件,使用 Whisper 生成..."
# 使用 Whisper 生成字幕(后台执行)
python /path/to/whisper.py \
--input "${video_path}" \
--model base \
--output "${content_dir}" \
--background
# 等待 Whisper 完成
sleep 5 # 等待后台任务启动
while ! [ -f "${content_dir}/$(basename ${video_path} .mp4).srt" ]; do
sleep 2
done
echo "Whisper 处理完成"
fi
```
## 输出文件使用
### JSON 格式
```python
import json
# 读取 JSON 结果
with open('video.json', 'r', encoding='utf-8') as f:
data = json.load(f)
# 访问分段
for segment in data['segments']:
print(f"[{segment['start']:.2f}-{segment['end']:.2f}] {segment['text']}")
```
### SRT 格式
```bash
# 直接在视频播放器中使用
mpv video.mp4 --sub-file=video.srt
# 或转换字幕格式
ffmpeg -i video.srt video.ass
```
### TXT 格式
```bash
# 搜索关键词
grep "重要信息" video.txt
# 统计字数
wc -m video.txt
```
## 性能优化建议
### 1. 根据内容选择模型
```bash
# 快速预览(短视频、测试)
python whisper.py --input video.mp4 --model tiny
# 日常使用(播客、会议)
python whisper.py --input video.mp4 --model base
# 高质量需求(正式内容、多语言)
python whisper.py --input video.mp4 --model small
```
### 2. 使用后台执行
```bash
# 处理长文件时使用后台模式
python whisper.py --input long_video.mp4 --model base --background
```
### 3. 指定语言提升准确度
```bash
# 如果知道内容语言,指定后准确度更高
python whisper.py --input video.mp4 --language zh
```
## 故障排查
### 问题 1: whisper.cpp 未找到
```bash
# 检查环境变量
echo $WHISPER_CPP_PATH
# 或配置
clawdbot skills config kb-whisper WHISPER_CPP_PATH "~/whisper.cpp"
```
### 问题 2: 模型未下载
```bash
cd ~/whisper.cpp
./models/download-ggml-model.sh base
```
### 问题 3: 音频格式不支持
```bash
# 转换格式
ffmpeg -i input.avi -acodec libmp3lame output.mp3
python whisper.py --input output.mp3
```
### 问题 4: 后台任务未完成
```bash
# 查看日志
cat video_whisper.log
# 检查进程
ps aux | grep whisper
```
## 集成到其他工具
### 集成到 Python 脚本
```python
import subprocess
import json
def transcribe_audio(audio_path, model='base'):
"""调用 whisper.py 转录音频"""
cmd = [
'python3', '/path/to/whisper.py',
'--input', audio_path,
'--model', model
]
result = subprocess.run(cmd, capture_output=True, text=True)
return json.loads(result.stdout)
# 使用
result = transcribe_audio('podcast.mp3', 'base')
print(f"识别到 {result['total_segments']} 个片段")
```
### 集成到 Shell 脚本
```bash
#!/bin/bash
transcribe() {
local file=$1
local model=${2:-base}
echo "转录: $file"
python3 whisper.py --input "$file" --model "$model" --background
}
# 批量处理
transcribe "video1.mp4"
transcribe "video2.mp4" "tiny"
```
-403
View File
@@ -1,403 +0,0 @@
#!/usr/bin/env python3
"""
KB Whisper - 本地语音识别工具
基于 Whisper.cpp 的音视频转文字
功能:
- 自动将视频/音频转换为 16kHz WAV 格式
- 支持指定语言(默认中文)
- 输出标准 JSON 格式供 KB Workflow 使用
"""
import argparse
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
import re
def parse_arguments():
"""解析命令行参数"""
parser = argparse.ArgumentParser(
description='KB Whisper - 本地语音识别工具',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
python whisper.py --input video.mp4
python whisper.py --input video.mp4 --model base
python whisper.py --input video.mp4 --output ./output --background
"""
)
parser.add_argument(
'--input',
required=True,
help='输入音视频文件路径'
)
parser.add_argument(
'--model',
default='base',
choices=['tiny', 'base', 'small', 'medium', 'large'],
help='Whisper 模型类型 (默认: base)'
)
parser.add_argument(
'--output',
help='输出目录 (默认: 输入文件所在目录)'
)
parser.add_argument(
'--background',
action='store_true',
help='后台异步执行'
)
parser.add_argument(
'--language',
default='zh',
help='指定语言 (默认: zh),例如: en, zh, ja'
)
return parser.parse_args()
def get_whisper_cpp_path():
"""获取 whisper.cpp 路径"""
# 优先从环境变量获取
path = os.environ.get('WHISPER_CPP_PATH')
if path:
path = os.path.expanduser(path)
if os.path.exists(path):
return path
# 尝试常见位置
common_paths = [
os.path.expanduser('~/whisper.cpp'),
'/usr/local/bin/whisper',
'/opt/whisper.cpp',
]
for cp in common_paths:
if os.path.exists(cp):
return cp
raise FileNotFoundError(
"未找到 whisper.cpp。请设置环境变量 WHISPER_CPP_PATH "
"或安装到 ~/whisper.cpp\n"
"安装方法: git clone https://github.com/ggerganov/whisper.cpp.git ~/whisper.cpp && cd ~/whisper.cpp && make"
)
def check_model(whisper_path, model):
"""检查模型是否存在"""
model_path = os.path.join(whisper_path, 'models', f'ggml-{model}.bin')
if not os.path.exists(model_path):
raise FileNotFoundError(
f"模型文件不存在: {model_path}\n"
f"请运行以下命令下载模型:\n"
f"cd {whisper_path} && ./models/download-ggml-model.sh {model}"
)
return model_path
def convert_to_wav(input_path, output_dir):
"""将音视频转换为 16kHz WAV 格式
Whisper 需要 WAV 格式的输入文件。
"""
input_path = os.path.abspath(input_path)
# 如果已经是 WAV 且采样率正确,直接返回
if input_path.lower().endswith('.wav'):
try:
result = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries',
'stream=sample_rate', '-of', 'default=noprint_wrappers=1:nokey=1',
input_path],
capture_output=True, text=True
)
if result.returncode == 0 and int(result.stdout.strip()) == 16000:
print(f"输入文件已是 16kHz WAV 格式,直接使用")
return input_path
except Exception:
pass
# 转换
output_path = os.path.join(output_dir, f"{Path(input_path).stem}_audio.wav")
print(f"转换音频格式: {input_path} -> {output_path}")
try:
subprocess.run([
'ffmpeg', '-y', '-i', input_path,
'-acodec', 'pcm_s16le',
'-ar', '16000',
'-ac', '1',
'-loglevel', 'error',
output_path
], check=True, capture_output=True)
print("音频转换完成")
return output_path
except subprocess.CalledProcessError as e:
print(f"音频转换失败: {e}")
raise RuntimeError(f"无法转换音频格式: {e}")
except FileNotFoundError:
raise RuntimeError("未找到 ffmpeg,请安装: sudo apt-get install ffmpeg")
def run_whisper(whisper_path, audio_path, model, output_dir, language='zh'):
"""运行 Whisper 语音识别
Args:
whisper_path: whisper.cpp 目录
audio_path: WAV 音频文件路径
model: 模型名称
output_dir: 输出目录
language: 语言代码
"""
# 输出文件名(使用原始输入文件名,不是 audio.wav)
# 从 output_dir 和 audio_path 推断原始文件名
original_stem = Path(audio_path).stem.replace('_audio', '')
output_basename = original_stem
# 构建命令
cmd = [
os.path.join(whisper_path, 'whisper-cli'),
'-m', os.path.join(whisper_path, 'models', f'ggml-{model}.bin'),
'-f', audio_path,
'-of', os.path.join(output_dir, output_basename),
'-l', language,
'--output-txt',
'--output-srt',
'--output-json',
]
print(f"执行命令: {' '.join(cmd)}")
# 运行
result = subprocess.run(
cmd,
capture_output=True,
text=True,
cwd=whisper_path,
timeout=3600 # 1小时超时
)
if result.returncode != 0:
# 检查是否有警告但仍成功
if "error" in result.stderr.lower() and not os.path.exists(
os.path.join(output_dir, f"{output_basename}.txt")
):
raise RuntimeError(
f"Whisper 执行失败:\n"
f"标准输出: {result.stdout}\n"
f"错误输出: {result.stderr}"
)
return result
def parse_srt_for_duration(srt_file):
"""从 SRT 文件解析时长"""
if not os.path.exists(srt_file):
return 0
with open(srt_file, 'r', encoding='utf-8') as f:
content = f.read()
# 找到最后一个时间戳
pattern = r'(\d{2}:\d{2}:\d{2}),\d{3} --> \d{2}:\d{2}:\d{2},\d{3}'
matches = re.findall(pattern, content)
if not matches:
return 0
# 解析最后一个时间戳
last_time = matches[-1]
h, m, s = last_time.split(':')
return int(h) * 3600 + int(m) * 60 + int(s)
def create_workflow_json(input_file, output_dir, model, language, duration):
"""创建标准格式的 JSON 输出,供 KB Workflow 使用"""
input_path = Path(input_file)
basename = input_path.stem
txt_file = os.path.join(output_dir, f"{basename}.txt")
srt_file = os.path.join(output_dir, f"{basename}.srt")
json_file = os.path.join(output_dir, f"{basename}.json")
# 读取文本内容
text_content = ""
if os.path.exists(txt_file):
with open(txt_file, 'r', encoding='utf-8') as f:
text_content = f.read().strip()
# 解析 SRT 获取分段
segments = []
if os.path.exists(srt_file):
with open(srt_file, 'r', encoding='utf-8') as f:
content = f.read()
# SRT 格式: 序号 -> 时间 -> 文本
pattern = r'(\d+)\n(\d{2}:\d{2}:\d{2},\d{3}) --> (\d{2}:\d{2}:\d{2},\d{3})\n(.+?)(?=\n\n|\Z)'
matches = re.findall(pattern, content, re.DOTALL)
def parse_time(t):
h, m, s_ms = t.split(':')
s, ms = s_ms.split(',')
return float(h) * 3600 + float(m) * 60 + float(s) + float(ms) / 1000
for idx, (num, start, end, text) in enumerate(matches):
segments.append({
"id": idx,
"start": round(parse_time(start), 2),
"end": round(parse_time(end), 2),
"text": text.strip()
})
# 构建 JSON
workflow_json = {
"success": True,
"tool": "whisper",
"model": model,
"language": language,
"input_file": str(input_path),
"output_files": {
"txt": f"{basename}.txt",
"srt": f"{basename}.srt",
"json": f"{basename}.json"
},
"duration_seconds": duration,
"segments": segments,
"text": text_content
}
# 写入 JSON
with open(json_file, 'w', encoding='utf-8') as f:
json.dump(workflow_json, f, ensure_ascii=False, indent=2)
return workflow_json
def main():
"""主函数"""
args = parse_arguments()
# 检查输入文件
input_file = os.path.abspath(args.input)
if not os.path.exists(input_file):
print(f"错误: 输入文件不存在: {input_file}", file=sys.stderr)
sys.exit(1)
# 获取输出目录
if args.output:
output_dir = os.path.abspath(args.output)
else:
output_dir = os.path.dirname(input_file)
os.makedirs(output_dir, exist_ok=True)
# 获取 whisper.cpp 路径
try:
whisper_path = get_whisper_cpp_path()
print(f"✓ Whisper.cpp 路径: {whisper_path}")
except FileNotFoundError as e:
print(f"✗ 错误: {e}", file=sys.stderr)
sys.exit(1)
# 检查模型
try:
check_model(whisper_path, args.model)
print(f"✓ 使用模型: {args.model}")
except FileNotFoundError as e:
print(f"✗ 错误: {e}", file=sys.stderr)
sys.exit(1)
# 后台执行
if args.background:
print("启动后台处理...")
log_file = os.path.join(output_dir, f"{Path(input_file).stem}_whisper.log")
with open(log_file, 'w') as log:
subprocess.Popen(
[sys.executable, __file__,
'--input', input_file,
'--model', args.model,
'--output', output_dir,
'--language', args.language],
stdout=log,
stderr=log,
start_new_session=True
)
print(f"✓ 后台任务已启动")
print(f" 日志: {log_file}")
print(f" 输出: {output_dir}")
result = {
"success": True,
"background": True,
"status": "running",
"log_file": log_file,
"output_dir": output_dir
}
print(json.dumps(result, ensure_ascii=False, indent=2))
return
# 前台执行
print(f"\n📁 处理文件: {input_file}")
print(f"📁 输出目录: {output_dir}")
try:
# 1. 转换为 WAV
print(f"\n🔄 步骤1: 转换音频格式...")
audio_path = convert_to_wav(input_file, output_dir)
# 2. 运行 Whisper
print(f"\n🎙️ 步骤2: 语音识别 (语言: {args.language})...")
run_whisper(whisper_path, audio_path, args.model, output_dir, args.language)
# 3. 清理临时 WAV 文件
if audio_path != input_file and audio_path.endswith('_audio.wav'):
try:
os.remove(audio_path)
print("✓ 清理临时文件")
except Exception:
pass
# 4. 获取时长并创建 JSON
print(f"\n📝 步骤3: 生成输出文件...")
basename = Path(input_file).stem
srt_file = os.path.join(output_dir, f"{basename}.srt")
duration = parse_srt_for_duration(srt_file)
workflow_json = create_workflow_json(
input_file, output_dir, args.model, args.language, duration
)
# 输出结果
print(f"\n✅ 处理完成!")
print(f" 时长: {duration} 秒")
print(f" 分段数: {len(workflow_json['segments'])}")
print(f"\n📄 输出文件:")
for fmt in ['txt', 'srt', 'json']:
fpath = os.path.join(output_dir, f"{basename}.{fmt}")
if os.path.exists(fpath):
size = os.path.getsize(fpath)
print(f" • {fmt.upper()}: {fpath} ({size:,} bytes)")
print(f"\n📋 Workflow JSON:")
print(json.dumps({
"success": workflow_json["success"],
"tool": "whisper",
"text_length": len(workflow_json["text"]),
"segments_count": len(workflow_json["segments"])
}, ensure_ascii=False, indent=2))
except Exception as e:
print(f"\n✗ 错误: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == '__main__':
main()
-86
View File
@@ -1,86 +0,0 @@
# KB Zhihu Content Skill
**kb-zhihu-content** - Download Zhihu videos and extract content.
## Features
- Video download using **yt-dlp**
- Subtitle extraction (returns no subtitle, triggers Whisper)
- **Read article content from MediaCrawler data** (NEW!)
- Supports `ZHIHU_COOKIE` environment variable
## Environment Variables
| Variable | Required | Description |
|----------|----------|-------------|
| `ZHIHU_COOKIE` | No | 知乎登录 Cookie (for member-only videos) |
## Commands
```bash
# Download video
python zhihu_content.py --id "12345678" --type video --output "/path/to/downloads/"
# Extract subtitle (returns no subtitle, triggers Whisper)
python zhihu_content.py --id "12345678" --type subtitle --output "/path/to/subtitles/"
# 从 MediaCrawler 读取文章内容 (⚡ NEW!)
python zhihu_content.py --type article-from-media --input "~/work/MediaCrawler/data/zhihu/json/detail_contents_2026-02-27.json" --output "~/work/knowledge-base/data/zhihu/"
```
## MediaCrawler 数据读取
**新功能**:从 MediaCrawler 爬取的数据中提取知乎文章内容
### 使用方法
```bash
# 从 MediaCrawler JSON 文件读取文章内容
python zhihu_content.py --type article-from-media \
--input "~/work/MediaCrawler/data/zhihu/json/detail_contents_2026-02-27.json" \
--output "~/work/knowledge-base/data/zhihu/"
# 查找最新的爬取数据
python zhihu_content.py --type article-from-media \
--output "~/work/knowledge-base/data/zhihu/"
```
### 参数说明
| 参数 | 说明 |
|------|------|
| `--type article-from-media` | 从 MediaCrawler 数据读取 |
| `--input` | MediaCrawler 输出文件路径(可选,不指定则自动查找最新) |
| `--output` | 输出目录 |
### 输出格式
```json
{
"success": true,
"type": "article",
"filename": "zhihu_article_如何高效的爬取微信公众号.md",
"filepath": "/home/xxx/work/knowledge-base/data/zhihu/zhihu_article_如何高效的爬取微信公众号.md",
"message": "文章内容已导出"
}
```
## 原有功能
### Video Download
Video download uses **yt-dlp**:
1. Parses `ZHIHU_COOKIE` environment variable (optional)
2. Creates temporary Netscape-format cookie file
3. Runs `yt-dlp -f best -o <output> <url>`
4. Cleans up temporary file
**Note**: 知乎视频通常没有官方字幕,将自动触发 Whisper 语音识别
## Installation
```bash
# Already included in kb-skills repo
# Ensure yt-dlp is installed
pip install yt-dlp aiohttp
```
-365
View File
@@ -1,365 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
知乎内容获取工具
支持:
- zvideo: 知乎视频下载
- article-from-media: 从 MediaCrawler 数据读取知乎文章
- 从页面提取视频播放地址
"""
import argparse
import asyncio
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Optional, Dict, Any, List
# MediaCrawler 目录
MEDIACRAWLER_DIR = Path.home() / "work" / "MediaCrawler"
class ZhihuContent:
"""知乎内容获取器"""
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
cookies = {}
for item in self.cookie.split(';'):
item = item.strip()
if '=' in item:
key, value = item.split('=', 1)
cookies[key.strip()] = value.strip()
cookie_content = """# Netscape HTTP Cookie File
# Generated by zhihu_content.py
"""
for name, value in cookies.items():
cookie_content += f".zhihu.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, video_id: str, quality: int = 720) -> Dict[str, Any]:
"""下载知乎视频(使用 yt-dlp)"""
result = {"success": False, "video_id": video_id, "message": "", "filepath": None, "metadata": {}}
try:
print(f"获取知乎视频信息: {video_id}")
cookie_file = self._create_cookie_file()
url = f"https://www.zhihu.com/zvideo/{video_id}"
# 先获取视频标题
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"zhihu_video_{video_id}"
except Exception as e:
title = f"zhihu_video_{video_id}"
# 构建 yt-dlp 命令
safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '-', '_')).strip()[:50]
print(f"下载视频: {title}")
output_template = str(self.output_dir / f"{safe_title}.%(ext)s")
# 构建下载命令
cmd = [
"yt-dlp",
"-f", "bestvideo+bestaudio/best",
"-o", output_template,
"--no-download-archive",
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, "video_id": video_id}
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, video_id: str) -> Dict[str, Any]:
"""
提取字幕
知乎视频通常没有官方字幕,需要使用 Whisper 进行语音识别。
此函数返回 has_subtitle=False,让工作流继续执行 Whisper 步骤。
"""
result = {"success": True, "video_id": video_id, "has_subtitle": False, "message": "知乎视频无官方字幕,将使用 Whisper 进行语音识别"}
return result
def read_article_from_mediacrawler(self, input_file: str = None, output_dir: str = None) -> Dict[str, Any]:
"""
从 MediaCrawler 爬取的数据中读取知乎文章内容
Args:
input_file: MediaCrawler 输出文件路径,不指定则自动查找最新
output_dir: 输出目录
Returns:
处理结果字典
"""
# 确定输出目录
if output_dir:
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
# 查找数据文件
data_dir = MEDIACRAWLER_DIR / "data" / "zhihu" / "json"
if input_file:
# 使用指定的文件
json_file = Path(input_file)
if not json_file.exists():
return {"success": False, "message": f"文件不存在: {input_file}"}
else:
# 查找最新的数据文件
json_files = sorted(data_dir.glob("detail_contents_*.json"), reverse=True)
if not json_files:
return {"success": False, "message": "未找到 MediaCrawler 爬取的数据,请先运行爬虫"}
json_file = json_files[0]
try:
with open(json_file, encoding="utf-8") as f:
data = json.load(f)
if not data:
return {"success": False, "message": "数据文件为空"}
# 提取第一篇文章
item = data[0]
# 准备数据
title = item.get("title", "无标题")
content = item.get("content_text", "")
url = item.get("content_url", "")
author = item.get("user_nickname", "未知")
created_ts = item.get("created_time", 0)
created = datetime.fromtimestamp(created_ts).strftime("%Y-%m-%d %H:%M:%S") if created_ts else ""
voteup = item.get("voteup_count", 0)
comments = item.get("comment_count", 0)
# 生成文件名
safe_title = "".join(c for c in title if c.isalnum() or c in " -_")[:50]
filename = f"zhihu_article_{safe_title}.md"
output_path = self.output_dir / filename
# 写入 Markdown 文件
markdown_content = f"""---
title: "{title}"
author: "{author}"
url: "{url}"
created: "{created}"
voteup: {voteup}
comments: {comments}
platform: zhihu
type: article
---
# {title}
**作者**: {author}
**来源**: [知乎]({url})
**发布时间**: {created}
**点赞**: {voteup}
**评论**: {comments}
---
{content}
"""
with open(output_path, "w", encoding="utf-8") as f:
f.write(markdown_content)
return {
"success": True,
"type": "article",
"title": title,
"filename": filename,
"filepath": str(output_path),
"content_length": len(content),
"message": f"文章内容已导出: {title}"
}
except Exception as e:
return {"success": False, "message": f"读取失败: {str(e)}"}
async def main():
parser = argparse.ArgumentParser(description='知乎内容获取工具')
parser.add_argument('--id', '-i', help='知乎视频ID (用于 video/subtitle 类型)')
parser.add_argument('--type', '-t', required=True,
choices=['video', 'subtitle', 'article-from-media', 'zhuanlan', 'article'],
help='内容类型: video=视频下载, subtitle=字幕提取, article-from-media=从MediaCrawler读取文章, zhuanlan/article=爬取知乎专栏文章')
parser.add_argument('--output', '-o', default='./output', help='输出目录')
parser.add_argument('--quality', '-q', type=int, default=720, help='视频质量')
parser.add_argument('--input', '-in', help='MediaCrawler 数据文件路径 (用于 article-from-media 类型)')
args = parser.parse_args()
cookie = os.environ.get('ZHIHU_COOKIE')
downloader = ZhihuContent(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 == 'article-from-media':
# 同步调用,不需要 asyncio
result = downloader.read_article_from_mediacrawler(
input_file=args.input,
output_dir=args.output
)
elif args.type in ('zhuanlan', 'article'):
# 知乎专栏文章:先爬取再读取
# 需要从 --id 或构造完整 URL
if args.id:
# 如果传入的是完整 URL
if args.id.startswith('http'):
article_url = args.id
else:
# 如果只传入 ID,构造 URL
article_url = f"https://zhuanlan.zhihu.com/p/{args.id}"
else:
result = {"success": False, "message": "zhuanlan/article 类型需要 --id 参数(文章ID或完整URL)"}
print(json.dumps(result, ensure_ascii=False, indent=2))
return result
# 调用 zhihu_article_crawler.py 爬取
crawler_script = Path(__file__).parent.parent / "kb-zhihu-metadata" / "zhihu_article_crawler.py"
if not crawler_script.exists():
# 尝试另一种路径
crawler_script = Path.home() / ".openclaw" / "agents" / "knowledge-base" / "workspace" / "skills" / "kb-zhihu-metadata" / "zhihu_article_crawler.py"
if crawler_script.exists():
import subprocess
cmd = ["python3", str(crawler_script), "--url", article_url, "--output", args.output]
print(f"调用爬虫: {' '.join(cmd)}")
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0:
result = {"success": False, "message": f"爬取失败: {proc.stderr[:200]}"}
else:
# 解析爬虫输出获取文件路径
try:
crawler_output = json.loads(proc.stdout.split("--- JSON 输出 ---")[-1].strip())
if crawler_output.get("success"):
# 直接使用爬虫输出的文件
result = {
"success": True,
"type": "article",
"title": crawler_output.get("title"),
"filename": Path(crawler_output["filepath"]).name,
"filepath": crawler_output["filepath"],
"content_length": crawler_output.get("content_length"),
"message": f"知乎专栏文章已爬取: {crawler_output.get('title')}"
}
else:
result = {"success": False, "message": crawler_output.get("error", "爬取失败")}
except Exception as e:
result = {"success": False, "message": f"解析爬虫输出失败: {str(e)}"}
else:
result = {"success": False, "message": f"未找到 zhihu_article_crawler.py: {crawler_script}"}
else:
result = {"success": False, "message": f"不支持的类型: {args.type}"}
# 统一输出格式
if result.get("success"):
output = {
"success": True,
"type": args.type,
"has_subtitle": result.get("has_subtitle", False),
"filename": result.get('filename') or (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', '处理成功')
}
if result.get('title'):
output['title'] = result['title']
if result.get('content_length'):
output['content_length'] = result['content_length']
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)