#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ KB Workflow - Knowledge Base Workflow Orchestrator 知识库工作流辅助工具函数。 此模块提供查重、文件操作、Git 管理等工具函数， 供主代理在执行 SKILL.md 中定义的 workflow 时调用。 """ import os import sys import json import argparse import subprocess import shutil import re from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any # ==================== 工具函数 ==================== def extract_platform(filename: str) -> str: """从文件名提取平台（第一个下划线之前）""" match = re.match(r'^([a-zA-Z0-9]+)_(.+)$', filename) if match: return match.group(1) return "unknown" def extract_base_name(filename: str) -> str: """从文件名提取基础名称（去掉扩展名）""" return Path(filename).stem def get_knowledge_base_path() -> Path: """获取知识库根目录""" base_path_str = os.environ.get('KB_BASE_PATH', '') if base_path_str: return Path(base_path_str).expanduser() return Path.home() / "work" / "knowledge-base" def ensure_directories(base_path: Path = None) -> Dict[str, Path]: """确保目录存在""" if base_path is None: base_path = get_knowledge_base_path() dirs = { "base": base_path, "pending": base_path / "pending", "metadata": base_path / "metadata", "content": base_path / "content", "content_downloads": base_path / "content" / "downloads", "content_subtitles": base_path / "content" / "subtitles", "analyzed": base_path / "analyzed", "archive": base_path / "archive", "archive_subscriptions": base_path / "archive" / "subscriptions", } for name, dir_path in dirs.items(): if isinstance(dir_path, Path): dir_path.mkdir(parents=True, exist_ok=True) return dirs def check_duplicate(base_path: Path = None, url: str = None, platform: str = None, content_id: str = None) -> Dict[str, Any]: """查重检查""" if base_path is None: base_path = get_knowledge_base_path() metadata_dir = base_path / "metadata" archive_dir = base_path / "archive" # TODO 直接使用url在知识库中查询metadata中的origin_url项 result = {"duplicate": False, "status": None, "files": []} return result def archive_and_git( analyzed_file: str, base_path: Path = None, git_root: Path = None, git_user: str = None, git_email: str = None, git_remote: str = None ) -> Dict[str, Any]: """归档并提交Git""" if base_path is None: base_path = get_knowledge_base_path() if git_root is None: git_root = base_path / "archive" if git_user is None: git_user = os.environ.get('KB_GIT_USER', '') if git_email is None: git_email = os.environ.get('KB_GIT_EMAIL', '') if git_remote is None: git_remote = os.environ.get('KB_GIT_REMOTE', '') result = { "success": False, "archive_path": "", "git_commit": False, "git_push": None, "error": None } try: # 1. 创建归档目录 date_str = datetime.now().strftime('%Y-%m-%d') date_dir = base_path / "archive" / date_str date_dir.mkdir(parents=True, exist_ok=True) # 2. 复制分析报告 src_file = Path(analyzed_file) archive_file = date_dir / src_file.name shutil.copy(src_file, archive_file) result["archive_path"] = str(archive_file) # 3. 复制元数据 meta_file = base_path / "metadata" / f"{src_file.stem}.meta.json" if meta_file.exists(): shutil.copy(meta_file, date_dir / meta_file.name) # 4. Git 操作 git_dir = git_root / ".git" if not git_dir.exists(): subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True) if git_user: subprocess.run(['git', 'config', 'user.name', git_user], cwd=str(git_root)) if git_email: subprocess.run(['git', 'config', 'user.email', git_email], cwd=str(git_root)) if (git_root / ".git").exists(): subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True) commit_msg = f"Archive: {src_file.name} - {datetime.now().isoformat()}" commit_result = subprocess.run( ['git', 'commit', '-m', commit_msg], cwd=str(git_root), capture_output=True, text=True ) result["git_commit"] = commit_result.returncode == 0 if git_remote and commit_result.returncode == 0: push_result = subprocess.run( ['git', 'push', 'origin', 'master'], cwd=str(git_root), capture_output=True, text=True, timeout=60 ) result["git_push"] = {"success": push_result.returncode == 0} result["success"] = True except Exception as e: result["error"] = str(e) return result def generate_summary(url: str, stages: List[Dict], final_file: str, archive_result: Dict = None) -> str: """生成处理完成汇报""" # 提取视频标题（如果有） title_match = re.search(r'com/video/([^?]+)', url) title = title_match.group(1) if title_match else "未知" summary = f""" ✅ **处理完成！** **视频:** {title} **处理阶段:** """ for stage in stages: stage_name = stage.get('stage', 'unknown') emoji = "✓" if stage.get('success') else "○" summary += f"{emoji} {stage_name}\n" if final_file: summary += f"\n**最终文件:** `{final_file}`" if archive_result: if archive_result.get('git_commit'): summary += f"\n**Git:** 已提交" if archive_result.get('git_push', {}).get('success'): summary += " ✓" elif archive_result.get('git_push'): summary += " ✗" return summary # ==================== CLI ==================== def main(): """CLI入口 - 用于命令行测试""" parser = argparse.ArgumentParser( description='KB Workflow - Knowledge Base Workflow Orchestrator', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 使用示例: # 查重 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 "analyzed/bilibili_BVxxx.md" """ ) parser.add_argument('--check', action='store_true', help='查重') 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_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 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 parser.print_help() if __name__ == '__main__': main()