From 33eeb6bc34a882d323688c3f67e4d9b0a273b032 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:44:10 +0800 Subject: [PATCH] feat(04-05): add CLI 'process' subcommand - Add process_command function that executes full pipeline - Add subparser for 'process' command with URL argument - Add --output-dir and --skip-completed flags - Add --no-skip flag to re-run all steps - Maintain backward compatibility with legacy arguments Co-Authored-By: Claude Opus 4.6 --- kb-workflow/kb_workflow.py | 134 ++++++++++++++++++++++++++++++++----- 1 file changed, 119 insertions(+), 15 deletions(-) diff --git a/kb-workflow/kb_workflow.py b/kb-workflow/kb_workflow.py index 40fd971..540b0ae 100644 --- a/kb-workflow/kb_workflow.py +++ b/kb-workflow/kb_workflow.py @@ -325,6 +325,83 @@ def generate_summary(url: str, stages: List[Dict], final_file: str, archive_resu return summary +# ==================== CLI Process Command ==================== + +def process_command( + url: str, + output_dir: Optional[str] = None, + skip_completed: bool = True +) -> Dict[str, Any]: + """Execute full pipeline for a URL + + Args: + url: URL to process + output_dir: Optional output directory for content + skip_completed: Skip already completed steps (default True) + + Returns: + Dictionary with success status, content_dir, and results + """ + import asyncio + + # Add project root to path for kb_pipeline imports + project_root = Path(__file__).parent.parent + if str(project_root) not in sys.path: + sys.path.insert(0, str(project_root)) + + try: + from kb_pipeline.pipeline import Pipeline + from kb_pipeline.processors import ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor + from kb_pipeline.schemas import PipelineConfig + + async def run_pipeline(): + # Determine output directory + out_dir = Path(output_dir) if output_dir else Path.cwd() + + # Create processors + processors = [ + ParseProcessor(), + DownloadProcessor(), + ASRProcessor(), + AnalyzeProcessor(), + ] + + # Create pipeline config + config = PipelineConfig( + steps=["parse", "download", "asr", "analyze"], + output_dir=out_dir, + skip_completed=skip_completed, + ) + + # Create and run pipeline + pipeline = Pipeline( + processors=processors, + config=config, + output_dir=out_dir, + ) + + result = await pipeline.run(url) + return result + + # Run async pipeline + result = asyncio.run(run_pipeline()) + + return { + "success": result.success, + "error": result.error_message, + "content_dir": str(result.content_dir) if result.content_dir else None, + "results": result.results, + } + + except Exception as e: + return { + "success": False, + "error": str(e), + "content_dir": None, + "results": {}, + } + + # ==================== CLI ==================== def main(): @@ -336,12 +413,28 @@ def main(): 使用示例: # 查重 python kb_workflow.py --check --url "https://www.bilibili.com/video/BVxxx" - + # 归档 python kb_workflow.py --archive "/path/to/analyzed.md" + + # 处理URL (通过完整管道) + python kb_workflow.py process "https://www.bilibili.com/video/BVxxx" """ ) - + + # Create subparsers for commands + subparsers = parser.add_subparsers(dest='command', help='Available commands') + + # Process subcommand + process_parser = subparsers.add_parser('process', help='Process URL through full pipeline') + process_parser.add_argument('url', type=str, help='URL to process') + process_parser.add_argument('--output-dir', '-o', type=str, help='Output directory for content') + process_parser.add_argument('--skip-completed', action='store_true', default=True, + help='Skip completed steps (default: True)') + process_parser.add_argument('--no-skip', action='store_false', dest='skip_completed', + help='Re-run all steps') + + # Legacy arguments (for backward compatibility) parser.add_argument('--check', action='store_true', help='查重') parser.add_argument('--url', type=str, help='要检查的URL') parser.add_argument('--archive-file', type=str, help='归档的文件路径') @@ -356,46 +449,57 @@ def main(): parser.add_argument('--archive_subscriptions', action='store_true', help='获取订阅信息目录') parser.add_argument('--extract_platform', type=str, help='要提取平台名称的文件名') args = parser.parse_args() - + base_path = Path(args.base_path) if args.base_path else None dirs = ensure_directories(base_path) - + + # Handle process subcommand + if args.command == 'process': + result = process_command( + url=args.url, + output_dir=args.output_dir, + skip_completed=args.skip_completed + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return + + # Legacy argument handling if args.check and args.url: result = check_duplicate(base_path, args.url) print(json.dumps(result, ensure_ascii=False, indent=2)) - + elif args.archive_file: result = archive_and_git(args.archive_file, base_path) print(json.dumps(result, ensure_ascii=False, indent=2)) - + elif args.pending: print(json.dumps({"path": str(dirs["pending"])}, ensure_ascii=False, indent=2)) - + elif args.metadata: print(json.dumps({"path": str(dirs["metadata"])}, ensure_ascii=False, indent=2)) - + elif args.content: print(json.dumps({"path": str(dirs["content"])}, ensure_ascii=False, indent=2)) - + elif args.content_downloads: print(json.dumps({"path": str(dirs["content_downloads"])}, ensure_ascii=False, indent=2)) - + elif args.analyzed: print(json.dumps({"path": str(dirs["analyzed"])}, ensure_ascii=False, indent=2)) - + elif args.archive_dir: print(json.dumps({"path": str(get_archive_path())}, ensure_ascii=False, indent=2)) - + elif args.archive_subscriptions: print(json.dumps({"path": str(dirs["archive_subscriptions"])}, ensure_ascii=False, indent=2)) - + elif args.extract_platform: platform = extract_platform(args.extract_platform) print(json.dumps({"platform": platform}, ensure_ascii=False, indent=2)) - + elif args.base_path: print(json.dumps({"path": str(dirs["base"])}, ensure_ascii=False, indent=2)) - + else: parser.print_help()