diff --git a/skills/kb-workflow/kb_workflow.py b/skills/kb-workflow/kb_workflow.py index 9cac8e2..f3cd787 100644 --- a/skills/kb-workflow/kb_workflow.py +++ b/skills/kb-workflow/kb_workflow.py @@ -20,19 +20,6 @@ from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any - -# ==================== 子技能映射 ==================== - -METADATA_AGENTS = { - "bilibili": "kb-bilibili-metadata", - "zhihu": "kb-zhihu-metadata", -} - -CONTENT_AGENTS = { - "bilibili": "kb-bilibili-content", -} - - # ==================== 工具函数 ==================== def extract_platform(filename: str) -> str: @@ -88,29 +75,10 @@ def check_duplicate(base_path: Path = None, url: str = None, platform: str = Non metadata_dir = base_path / "metadata" archive_dir = base_path / "archive" - # 从URL提取ID - if url and not content_id: - bv_match = re.search(r'BV[a-zA-Z0-9]+', url) - content_id = bv_match.group() if bv_match else None + # TODO 直接使用url在知识库中查询metadata中的origin_url项 result = {"duplicate": False, "status": None, "files": []} - - if metadata_dir.exists(): - for f in metadata_dir.iterdir(): - if f.is_file() and content_id and content_id in f.name: - result["duplicate"] = True - result["status"] = "metadata" - result["files"].append(str(f)) - - if archive_dir.exists(): - for date_dir in archive_dir.iterdir(): - if date_dir.is_dir(): - for f in date_dir.iterdir(): - if f.is_file() and content_id and content_id in f.name: - result["duplicate"] = True - result["status"] = "archived" - result["files"].append(str(f)) - + return result @@ -236,8 +204,26 @@ def main(): # 查重 python kb_workflow.py --check --url "https://www.bilibili.com/video/BVxxx" + # 提取平台 + python kb_workflow.py --extract_platform "bilibili_BVxxx.link.json" + + # 链接解析 + python kb_workflow.py --pending --url "https://www.bilibili.com/video/BVxxx" + + # 元数据提取 + python kb_workflow.py --metadata --link_file "pending/bilibili_BVxxx.link.json" + + # 内容下载 + python kb_workflow.py --content --meta_file "metadata/bilibili_BVxxx.meta.json" + + # OCR处理 + python kb_workflow.py --ocr --source "content/subtitles/bilibili_BVxxx.srt" + + # 深度分析 + python kb_workflow.py --analyze --meta_file "metadata/bilibili_BVxxx.meta.json" --text_file "content/bilibili_BVxxx.txt" + # 归档 - python kb_workflow.py --archive "/path/to/analyzed.md" + python kb_workflow.py --archive "analyzed/bilibili_BVxxx.md" """ ) @@ -245,21 +231,157 @@ def main(): parser.add_argument('--url', type=str, help='要检查的URL') parser.add_argument('--archive', type=str, help='要归档的文件路径') parser.add_argument('--base-path', type=str, help='知识库根目录') - + parser.add_argument('--pending', action='store_true', help='执行链接解析') + parser.add_argument('--metadata', action='store_true', help='执行元数据提取') + parser.add_argument('--content', action='store_true', help='执行内容下载') + parser.add_argument('--ocr', action='store_true', help='执行OCR处理') + parser.add_argument('--analyze', action='store_true', help='执行深度分析') + parser.add_argument('--link_file', type=str, help='链接解析结果文件路径') + parser.add_argument('--meta_file', type=str, help='元数据文件路径') + parser.add_argument('--text_file', type=str, help='文字内容文件路径') + parser.add_argument('--source', type=str, help='OCR源文件路径') + parser.add_argument('--extract_platform', type=str, help='提取平台名称') args = parser.parse_args() - base_path = Path(args.base_path) if args.base_path else None + # 获取知识库路径 + base_path_str = os.environ.get('KB_BASE_PATH', '') + base_path = Path(base_path_str).expanduser() if base_path_str else Path.home() / "work" / "knowledge-base" + + # 获取子目录路径 + pending_dir = base_path / "pending" + metadata_dir = base_path / "metadata" + content_dir = base_path / "content" + content_downloads_dir = content_dir / "downloads" + content_subtitles_dir = content_dir / "subtitles" + analyzed_dir = base_path / "analyzed" + archive_dir = base_path / "archive" + + if args.extract_platform: + result = extract_platform(args.extract_platform) + print(result) + return if args.check and args.url: result = check_duplicate(base_path, args.url) print(json.dumps(result, ensure_ascii=False, indent=2)) + return - elif args.archive: + if args.pending and args.url: + # 链接解析 - 调用 kb-link-parser + import subprocess + cmd = ['python3', 'kb-link-parser/link-parser.py', '--url', args.url] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode == 0: + data = json.loads(proc.stdout) + if data.get('success'): + platform = data.get('platform', 'unknown') + content_id = data.get('id', 'unknown') + output_file = pending_dir / f"{platform}_{content_id}.link.json" + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + print(json.dumps({"success": True, "output_file": str(output_file)}, ensure_ascii=False, indent=2)) + else: + print(proc.stdout) + else: + print(json.dumps({"success": False, "error": proc.stderr}, ensure_ascii=False, indent=2)) + return + + if args.metadata and args.link_file: + # 元数据提取 - 调用对应平台的元数据技能 + link_file = Path(args.link_file) + with open(link_file, 'r', encoding='utf-8') as f: + link_data = json.load(f) + platform = link_data.get('platform', '') + content_id = link_data.get('id', '') + + if platform == 'bilibili': + cmd = ['python3', 'kb-bilibili-metadata/bilibili_metadata.py', + '--id', content_id, '--type', 'video', + '--output', str(metadata_dir / f"{platform}_{content_id}.meta.json")] + elif platform == 'zhihu': + cmd = ['python3', 'kb-zhihu-metadata/zhihu_metadata.py', '--url', link_data.get('url', '')] + else: + print(json.dumps({"success": False, "error": f"不支持的平台: {platform}"}, ensure_ascii=False, indent=2)) + return + + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode == 0: + print(proc.stdout) + else: + print(json.dumps({"success": False, "error": proc.stderr}, ensure_ascii=False, indent=2)) + return + + if args.content and args.meta_file: + # 内容下载 - 调用对应平台的内容技能 + meta_file = Path(args.meta_file) + with open(meta_file, 'r', encoding='utf-8') as f: + meta_data = json.load(f) + platform = extract_platform(meta_file.name) + content_id = extract_base_name(meta_file.name) + + if platform == 'bilibili': + # 下载视频 + cmd_video = ['python3', 'kb-bilibili-content/bilibili_content.py', + '--id', content_id, '--type', 'video', '--output', str(content_dir)] + subprocess.run(cmd_video) + # 下载字幕 + cmd_subtitle = ['python3', 'kb-bilibili-content/bilibili_content.py', + '--id', content_id, '--type', 'subtitle', '--output', str(content_dir)] + subprocess.run(cmd_subtitle) + # 查找生成的文件 + video_file = None + subtitle_file = None + for f in content_downloads_dir.iterdir(): + if content_id in f.name: + video_file = str(f) + for f in content_subtitles_dir.iterdir(): + if content_id in f.name: + subtitle_file = str(f) + print(json.dumps({"success": True, "output_file": video_file, "subtitles": subtitle_file}, ensure_ascii=False, indent=2)) + else: + print(json.dumps({"success": False, "error": f"不支持的平台: {platform}"}, ensure_ascii=False, indent=2)) + return + + if args.ocr and args.source: + # OCR处理 - 调用 kb-ocr-processor + source_file = Path(args.source) + platform = extract_platform(source_file.name) + content_id = extract_base_name(source_file.name) + output_file = content_dir / f"{platform}_{content_id}.txt" + + cmd = ['python3', 'kb-ocr-processor/ocr_processor.py', + '--input', str(source_file), '--output', str(output_file)] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode == 0: + print(json.dumps({"success": True, "output_file": str(output_file)}, ensure_ascii=False, indent=2)) + else: + print(json.dumps({"success": False, "error": proc.stderr}, ensure_ascii=False, indent=2)) + return + + if args.analyze and args.meta_file: + # 深度分析 - 调用 kb-deep-analyzer + meta_file = Path(args.meta_file) + platform = extract_platform(meta_file.name) + content_id = extract_base_name(meta_file.name) + output_file = analyzed_dir / f"{platform}_{content_id}.md" + + text_file = args.text_file if args.text_file else str(meta_file) + + cmd = ['python3', 'kb-deep-analyzer/deep_analyzer.py', + '--metadata', str(meta_file), '--content', text_file, '--output', str(output_file)] + proc = subprocess.run(cmd, capture_output=True, text=True) + if proc.returncode == 0: + print(json.dumps({"success": True, "output_file": str(output_file)}, ensure_ascii=False, indent=2)) + else: + print(json.dumps({"success": False, "error": proc.stderr}, ensure_ascii=False, indent=2)) + return + + if args.archive: result = archive_and_git(args.archive, base_path) print(json.dumps(result, ensure_ascii=False, indent=2)) + return - else: - parser.print_help() + parser.print_help() if __name__ == '__main__':