From 48ace388dafdb90caf07e9785e7b40efd37357ef Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Tue, 10 Feb 2026 23:46:24 +0800 Subject: [PATCH] =?UTF-8?q?Revert=20"feat(kb-workflow):=20=E8=A1=A5?= =?UTF-8?q?=E5=85=A8CLI=E5=8F=82=E6=95=B0=E8=B0=83=E7=94=A8=E9=80=BB?= =?UTF-8?q?=E8=BE=91"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 61dad2414de918d2cf4927c1a1861538c020834d. --- skills/kb-workflow/kb_workflow.py | 202 ++++++------------------------ 1 file changed, 40 insertions(+), 162 deletions(-) diff --git a/skills/kb-workflow/kb_workflow.py b/skills/kb-workflow/kb_workflow.py index f3cd787..9cac8e2 100644 --- a/skills/kb-workflow/kb_workflow.py +++ b/skills/kb-workflow/kb_workflow.py @@ -20,6 +20,19 @@ 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: @@ -75,10 +88,29 @@ def check_duplicate(base_path: Path = None, url: str = None, platform: str = Non metadata_dir = base_path / "metadata" archive_dir = base_path / "archive" - # TODO 直接使用url在知识库中查询metadata中的origin_url项 + # 从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 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 @@ -204,26 +236,8 @@ 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 "analyzed/bilibili_BVxxx.md" + python kb_workflow.py --archive "/path/to/analyzed.md" """ ) @@ -231,157 +245,21 @@ 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_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 + base_path = Path(args.base_path) if args.base_path else None 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: + elif args.archive: result = archive_and_git(args.archive, base_path) print(json.dumps(result, ensure_ascii=False, indent=2)) - return - parser.print_help() + else: + parser.print_help() if __name__ == '__main__':