diff --git a/kb-bilibili-content/SKILL.md b/kb-bilibili-content/SKILL.md deleted file mode 100644 index 4701426..0000000 --- a/kb-bilibili-content/SKILL.md +++ /dev/null @@ -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 -f -o ` -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)** diff --git a/kb-bilibili-content/__pycache__/bilibili_content.cpython-312.pyc b/kb-bilibili-content/__pycache__/bilibili_content.cpython-312.pyc deleted file mode 100644 index d8b5ed3..0000000 Binary files a/kb-bilibili-content/__pycache__/bilibili_content.cpython-312.pyc and /dev/null differ diff --git a/kb-bilibili-content/bilibili_content.py b/kb-bilibili-content/bilibili_content.py deleted file mode 100644 index 08bfbef..0000000 --- a/kb-bilibili-content/bilibili_content.py +++ /dev/null @@ -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) diff --git a/kb-bilibili-content/bilibili_extractor.py b/kb-bilibili-content/bilibili_extractor.py deleted file mode 100644 index 971d239..0000000 --- a/kb-bilibili-content/bilibili_extractor.py +++ /dev/null @@ -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() diff --git a/kb-bilibili-content/video_downloader.py b/kb-bilibili-content/video_downloader.py deleted file mode 100644 index f5d25c4..0000000 --- a/kb-bilibili-content/video_downloader.py +++ /dev/null @@ -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❌ 下载失败") diff --git a/kb-funasr/.env.example b/kb-funasr/.env.example deleted file mode 100644 index 1c21fe6..0000000 --- a/kb-funasr/.env.example +++ /dev/null @@ -1,2 +0,0 @@ -# FunASR 模型缓存目录 -FUNASR_CACHE_DIR=~/.cache/modelscope/hub diff --git a/kb-funasr/README.md b/kb-funasr/README.md deleted file mode 100644 index baef2d8..0000000 --- a/kb-funasr/README.md +++ /dev/null @@ -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 许可证。 diff --git a/kb-funasr/SKILL.md b/kb-funasr/SKILL.md deleted file mode 100644 index 886012c..0000000 --- a/kb-funasr/SKILL.md +++ /dev/null @@ -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 等常见格式 -- **首次运行**: 首次运行时会自动下载模型文件，需要稳定的网络连接 diff --git a/kb-funasr/__pycache__/funasr.cpython-312.pyc b/kb-funasr/__pycache__/funasr.cpython-312.pyc deleted file mode 100644 index 17e5daf..0000000 Binary files a/kb-funasr/__pycache__/funasr.cpython-312.pyc and /dev/null differ diff --git a/kb-funasr/funasr_tool.py b/kb-funasr/funasr_tool.py deleted file mode 100644 index b082155..0000000 --- a/kb-funasr/funasr_tool.py +++ /dev/null @@ -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() diff --git a/kb-link-parser/SKILL.md b/kb-link-parser/SKILL.md deleted file mode 100644 index b8c97cc..0000000 --- a/kb-link-parser/SKILL.md +++ /dev/null @@ -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/ -``` diff --git a/kb-link-parser/__pycache__/link_parser_shortlink.cpython-312.pyc b/kb-link-parser/__pycache__/link_parser_shortlink.cpython-312.pyc deleted file mode 100644 index eeed198..0000000 Binary files a/kb-link-parser/__pycache__/link_parser_shortlink.cpython-312.pyc and /dev/null differ diff --git a/kb-link-parser/link-parser.py b/kb-link-parser/link-parser.py deleted file mode 100644 index b8cb635..0000000 --- a/kb-link-parser/link-parser.py +++ /dev/null @@ -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()) diff --git a/kb-link-parser/link_parser_shortlink.py b/kb-link-parser/link_parser_shortlink.py deleted file mode 100644 index 20739e7..0000000 --- a/kb-link-parser/link_parser_shortlink.py +++ /dev/null @@ -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() diff --git a/kb-ocr-processor/SKILL.md b/kb-ocr-processor/SKILL.md deleted file mode 100644 index cfd9ef5..0000000 --- a/kb-ocr-processor/SKILL.md +++ /dev/null @@ -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） -``` diff --git a/kb-ocr-processor/ocr_processor.py b/kb-ocr-processor/ocr_processor.py deleted file mode 100644 index b58d618..0000000 --- a/kb-ocr-processor/ocr_processor.py +++ /dev/null @@ -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()) diff --git a/kb-ocr-processor/requirements.txt b/kb-ocr-processor/requirements.txt deleted file mode 100644 index f3f72c2..0000000 --- a/kb-ocr-processor/requirements.txt +++ /dev/null @@ -1,3 +0,0 @@ -easyocr>=1.7.0 -pytesseract>=0.3.10 -Pillow>=9.0.0 diff --git a/kb-whisper-diarization/.env.example b/kb-whisper-diarization/.env.example deleted file mode 100644 index 108647b..0000000 --- a/kb-whisper-diarization/.env.example +++ /dev/null @@ -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 diff --git a/kb-whisper-diarization/SKILL.md b/kb-whisper-diarization/SKILL.md deleted file mode 100644 index c9fde27..0000000 --- a/kb-whisper-diarization/SKILL.md +++ /dev/null @@ -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 -- **处理时间较长**: 说话者分离会增加处理时间 -- **对立体声效果更好**: 多声道音频的分离效果更佳 diff --git a/kb-whisper-diarization/requirements.txt b/kb-whisper-diarization/requirements.txt deleted file mode 100644 index d465929..0000000 --- a/kb-whisper-diarization/requirements.txt +++ /dev/null @@ -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 diff --git a/kb-whisper-diarization/whisper_diarization.py b/kb-whisper-diarization/whisper_diarization.py deleted file mode 100644 index aa0cb58..0000000 --- a/kb-whisper-diarization/whisper_diarization.py +++ /dev/null @@ -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() diff --git a/kb-whisper/.env.example b/kb-whisper/.env.example deleted file mode 100644 index 8ffc440..0000000 --- a/kb-whisper/.env.example +++ /dev/null @@ -1,5 +0,0 @@ -# KB Whisper 环境变量配置示例 -# 复制此文件为 .env 并填写实际值 - -# Whisper.cpp 安装路径 -WHISPER_CPP_PATH=~/whisper.cpp diff --git a/kb-whisper/README.md b/kb-whisper/README.md deleted file mode 100644 index 7b84ea8..0000000 --- a/kb-whisper/README.md +++ /dev/null @@ -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 许可证。 diff --git a/kb-whisper/SKILL.md b/kb-whisper/SKILL.md deleted file mode 100644 index 9ebb999..0000000 --- a/kb-whisper/SKILL.md +++ /dev/null @@ -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 等常见格式 diff --git a/kb-whisper/__pycache__/whisper.cpython-312.pyc b/kb-whisper/__pycache__/whisper.cpython-312.pyc deleted file mode 100644 index c605169..0000000 Binary files a/kb-whisper/__pycache__/whisper.cpython-312.pyc and /dev/null differ diff --git a/kb-whisper/examples.md b/kb-whisper/examples.md deleted file mode 100644 index 02905d0..0000000 --- a/kb-whisper/examples.md +++ /dev/null @@ -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" -``` diff --git a/kb-whisper/whisper.py b/kb-whisper/whisper.py deleted file mode 100644 index 4d95f3a..0000000 --- a/kb-whisper/whisper.py +++ /dev/null @@ -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() diff --git a/kb-zhihu-content/SKILL.md b/kb-zhihu-content/SKILL.md deleted file mode 100644 index 9009c3e..0000000 --- a/kb-zhihu-content/SKILL.md +++ /dev/null @@ -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 ` -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 -``` diff --git a/kb-zhihu-content/zhihu_content.py b/kb-zhihu-content/zhihu_content.py deleted file mode 100644 index ff9c4b4..0000000 --- a/kb-zhihu-content/zhihu_content.py +++ /dev/null @@ -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)