feat(kb-workflow): 优化无字幕视频处理逻辑

- kb-bilibili-content: extract_subtitle 返回 success=True 即使没有字幕
- 添加 download_subtitle 方法,支持下载官方字幕并转为 SRT 格式
- kb-workflow: 更新工作流逻辑,没有字幕是正常的,继续执行 Whisper
- 添加条件分支:有字幕→深度分析,无字幕→Whisper
This commit is contained in:
2026-02-13 13:36:42 +08:00
parent b02cd969b8
commit a03223ac80
2 changed files with 178 additions and 31 deletions
+126 -7
View File
@@ -160,8 +160,13 @@ class BilibiliContent:
return result
async def extract_subtitle(self, bvid: str) -> Dict[str, Any]:
"""提取字幕"""
result = {"success": False, "bvid": bvid, "subtitles": [], "message": ""}
"""
提取字幕
注意:没有找到官方字幕是正常的!很多视频没有自动生成的字幕。
返回 success=True 即使没有字幕,让工作流继续执行后续步骤(如 Whisper)。
"""
result = {"success": False, "bvid": bvid, "subtitles": [], "has_subtitle": False, "message": ""}
try:
import aiohttp
@@ -174,7 +179,9 @@ class BilibiliContent:
pages = data.get('data', [])
if not pages:
result["message"] = "未找到分P"
result["message"] = "未找到分P(这是正常的)"
# 仍然返回 success=True,让工作流继续
result["success"] = True
return result
cid = pages[0].get('cid', '')
@@ -186,16 +193,124 @@ class BilibiliContent:
subtitles = sub_data.get('data', {}).get('subtitle', {}).get('list', [])
if not subtitles:
result["message"] = "未找到字幕"
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)} 个字幕"
result["message"] = f"✅ 找到 {len(subtitles)} 个字幕,将使用官方字幕进行深度分析"
print(result["message"])
except Exception as e:
result["message"] = f"错误: {str(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
@@ -252,7 +367,7 @@ class BilibiliContent:
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', 'article'], help='内容类型')
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')
@@ -265,6 +380,9 @@ async def main():
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', ''))
@@ -273,6 +391,7 @@ async def main():
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', '处理成功')
+52 -24
View File
@@ -94,36 +94,39 @@ python3 /path/to/zhihu_metadata.py --id "[ID]" --type question --output "${metad
1. 获取目录路径:
```bash
python3 kb_workflow.py --content_downloads
python3 kb_workflow.py --content_subtitles
python3 kb_workflow.py --content
```
2. 启动子代理调用对应平台的 content 技能:
```bash
# Bilibili 视频
python3 /path/to/bilibili_content.py --id "[ID]" --type video --output [content_dir]
# Bilibili 字幕下载(自动检测是否有官方字幕)
python3 /path/to/bilibili_content.py --id "[ID]" --type subtitle-download --output [content_dir]
# Bilibili 字幕
python3 /path/to/bilibili_content.py --id "[ID]" --type subtitle --output [subtitles_dir]
# 返回 JSON 包含 has_subtitle 字段
```
3. 解析输出 JSON,提取 video_filename 和 subtitle_filename(如有)
3. 解析输出 JSON:
- **has_subtitle = true**: 有官方字幕,直接进入步骤 6
- **has_subtitle = false**: 没有官方字幕,继续步骤 5
### 步骤 5: 语音识别转文字(禁止跳过)
> ⚠️ **重要提示**:没有找到官方字幕是**正常的**!很多视频没有自动生成的字幕。不要因为没有字幕就停止工作流,继续执行 Whisper 步骤即可。
> ⚠️ **重要**:此步骤为耗时步骤,使用 Whisper 进行语音识别。
> - **严禁跳过**:即使等待时间较长,也必须等待识别完成
> - **定时汇报**:每2-3分钟向用户汇报进度
> - **超时时间**:30分钟
> - **耐心等待**:不要中途启动其他步骤,必须串行执行
### 步骤 5: 语音识别转文字(仅当没有官方字幕时执行)
如果步骤4没有获取到字幕,需要使用 Whisper 进行语音识别:
> ⚠️ **重要**:此步骤仅在**没有官方字幕**时执行。如果有官方字幕,**跳过此步骤**。
此步骤使用 Whisper 进行语音识别:
- **严禁跳过**:必须等待识别完成
- **定时汇报**:每2-3分钟向用户汇报进度
- **超时时间**:30分钟
- **耐心等待**:不要中途启动其他步骤,必须串行执行
1. 获取目录路径:
```bash
python3 kb_workflow.py --content
```
2. 启动子代理调用 kb-whisper(传入完整路径):
2. 启动子代理调用 kb-whisper(传入视频路径):
```bash
# 必须传入 --output 完整路径
python3 /path/to/whisper.py --input [video_path] --model base --output [content_dir]
@@ -142,12 +145,15 @@ python3 kb_workflow.py --analyzed
2. 准备 content JSON 文件(包含 subtitle/description)
3. 启动子代理调用 kb-deep-analyzer:
3. 如果有官方字幕,使用官方字幕的 SRT 文件
4. 如果没有官方字幕,使用 Whisper 识别的结果
5. 启动子代理调用 kb-deep-analyzer:
```bash
python3 /path/to/deep_analyzer.py --metadata [meta_path] --content [content_path] --output [analyzed_dir]/[filename].md
```
4. 解析输出 JSON,获取分析结果
6. 解析输出 JSON,获取分析结果
### 步骤 7: 归档
@@ -167,14 +173,36 @@ python3 kb_workflow.py --archive-file "[分析报告文件路径]"
## 子技能清单
| 步骤 | 子技能 | 命令示例 | 输出 |
|------|--------|----------|------|
| 2 | kb-link-parser | `--url "BVxxx"` | `{p}_{id}.link.json` |
| 3 | kb-bilibili-metadata | `--id "BVxxx" --type video` | `{p}_{id}.meta.json` |
| 3 | kb-zhihu-metadata | `--id "xxx" --type question` | `{p}_{id}.meta.json` |
| 4 | kb-bilibili-content | `--id "BVxxx" --type video` | `{p}_{id}.{ext}` |
| 5 | kb-whisper | `--input xxx.mp4 --model base` | `{p}_{id}.json`, `{p}_{id}.txt` |
| 6 | kb-deep-analyzer | `--metadata x.meta.json --content x.json --output x.md` | `{p}_{id}.md` |
| 步骤 | 子技能 | 命令示例 | 输出 | 备注 |
|------|--------|----------|------|------|
| 2 | kb-link-parser | `--url "BVxxx"` | `{p}_{id}.link.json` | |
| 3 | kb-bilibili-metadata | `--id "BVxxx" --type video` | `{p}_{id}.meta.json` | |
| 3 | kb-zhihu-metadata | `--id "xxx" --type question` | `{p}_{id}.meta.json` | |
| 4 | kb-bilibili-content | `--id "BVxxx" --type subtitle-download` | `{p}_{id}.srt` | 如果没有字幕是正常的 |
| 4 | kb-bilibili-content | `--id "BVxxx" --type video` | `{p}_{id}.mp4` | 如果需要下载视频 |
| 5 | kb-whisper | `--input xxx.mp4 --model base` | `{p}_{id}.json`, `{p}_{id}.txt` | 仅当没有官方字幕时执行 |
| 6 | kb-deep-analyzer | `--metadata x.meta.json --content x.json --output x.md` | `{p}_{id}.md` | |
## 条件分支逻辑
```
步骤3: 获取元数据
↓
步骤4: 下载字幕 (--type subtitle-download)
↓
┌─────────────────────────────────────┐
│ has_subtitle = true? │
├─────────────────────────────────────┤
│ 是 → 直接进入步骤6(深度分析) │
│ 否 → 进入步骤5(Whisper语音识别) │
└─────────────────────────────────────┘
↓
步骤6: 深度分析
```
> ⚠️ **重要**:没有找到官方字幕是**正常的**!很多视频没有自动生成的字幕。
> - 如果 `has_subtitle = false`,继续执行 Whisper 步骤
> - 如果 `has_subtitle = true`,跳过 Whisper 步骤,直接使用官方字幕
## 目录结构