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 <noreply@anthropic.com>
This commit is contained in:
2026-03-13 23:44:10 +08:00
parent 76aefa23e1
commit 33eeb6bc34
+119 -15
View File
@@ -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()