diff --git a/.planning/debug/resolved/kb-workflow-legacy-code-removal.md b/.planning/debug/resolved/kb-workflow-legacy-code-removal.md new file mode 100644 index 0000000..e1d5f35 --- /dev/null +++ b/.planning/debug/resolved/kb-workflow-legacy-code-removal.md @@ -0,0 +1,74 @@ +--- +status: resolved +trigger: "kb-workflow不应该再包含旧流程" +created: 2026-03-16T10:00:00+08:00 +updated: 2026-03-16T11:00:00+08:00 +--- + +## Current Focus + +hypothesis: 根本原因已确认 - kb_workflow.py 包含312行冗余的旧流程代码，这些代码已被 kb_pipeline 包完全替代 +test: 重构并验证 +expecting: 修复后测试通过，AI代理不再看到旧代码 +next_action: 等待用户确认 + +## Symptoms + +expected: kb-workflow 只保留新的管道架构 (基于 kb_pipeline 包) +actual: 实际运行时AI代理仍然有时候会参考旧流程的代码处理 +errors: 功能异常 - AI代理可能调用旧流程函数而不是新管道 +reproduction: AI代理读取 kb_workflow.py 时看到旧流程函数并使用它们 +started: 在 v1.0 里程碑完成后发现，新的管道架构已经存在 + +## Eliminated + +## Evidence + +- timestamp: 2026-03-16T10:05:00+08:00 + checked: kb_workflow.py 文件结构 + found: 文件共495行，其中： + - 第1-312行：旧流程代码（内容类型判断、工具函数、归档函数等） + - 第314-389行：新的 process_command() 使用 kb_pipeline + - 第391-495行：CLI入口和参数处理 + implication: 约63%的代码是旧流程，可能干扰AI代理 + +- timestamp: 2026-03-16T10:08:00+08:00 + checked: kb_pipeline 包的导出 + found: kb_pipeline 已提供完整替代： + - LinkParser, ParsedLink - URL解析 + - IdempotencyChecker - 查重 + - PlatformAdapter, BilibiliAdapter, ZhihuAdapter - 平台适配 + - WorkflowStateManager - 状态管理 + - Pipeline - 管道编排 + implication: 旧函数已被完全替代 + +- timestamp: 2026-03-16T10:25:00+08:00 + checked: 旧函数与 kb_pipeline 的对应关系 + found: 所有旧函数都有替代实现 + implication: 可以安全移除 + +- timestamp: 2026-03-16T10:40:00+08:00 + checked: 修复后的代码 + found: kb_workflow.py 从 495 行减少到 143 行（减少 71%） + implication: AI代理现在只能看到管道入口 + +- timestamp: 2026-03-16T10:42:00+08:00 + checked: 测试验证 + found: pytest tests/test_cli/test_process_command.py - 7 tests passed + implication: 修复后功能正常 + +## Resolution + +root_cause: kb_workflow.py 包含312行已废弃的旧流程代码，这些代码已被 kb_pipeline 包完全替代。AI代理读取文件时会看到这些旧函数并可能错误地使用它们。 + +fix: +1. 移除 kb_workflow.py 中所有旧流程代码（is_video_content, check_duplicate, archive_and_git 等） +2. 只保留 process_command() 和 main() 入口 +3. CLI 只保留 `process` 子命令 +4. 更新 SKILL.md，移除旧命令示例 + +verification: 7 个测试全部通过 + +files_changed: +- kb-workflow/kb_workflow.py +- kb-workflow/SKILL.md diff --git a/kb-workflow/SKILL.md b/kb-workflow/SKILL.md index a4c8573..8b536d9 100644 --- a/kb-workflow/SKILL.md +++ b/kb-workflow/SKILL.md @@ -60,7 +60,7 @@ echo "✅ 环境变量检查通过" ## Workflow 执行方式 -### 方式一：一键处理（推荐） +### 一键处理（推荐） 使用管道 CLI 一键完成所有步骤： @@ -85,21 +85,6 @@ python3 kb-workflow/kb_workflow.py process "https://..." -o /path/to/output python3 kb-workflow/kb_workflow.py process "https://..." --no-skip ``` -### 方式二：手动分步执行（高级） - -如需更细粒度控制，可手动调用各子技能： - -```bash -# 1. 查重 -python3 kb-workflow/kb_workflow.py --check --url "https://..." - -# 2. 一键处理 -python3 kb-workflow/kb_workflow.py process "https://..." - -# 3. 归档 -python3 kb-workflow/kb_workflow.py --archive-file "/path/to/report.md" -``` - ## 支持的平台和内容类型 | 平台 | 内容类型 | URL 示例 | @@ -152,22 +137,6 @@ python3 kb-workflow/kb_workflow.py --archive-file "/path/to/report.md" 3. **禁止降级处理**：不能使用预设模板或简单摘要来"假装成功" 4. **每次执行都要使用 AI**：必须有真实的 AI API 调用记录 -## 辅助命令 - -```bash -# 查重 -python3 kb-workflow/kb_workflow.py --check --url "https://..." - -# 获取目录路径 -python3 kb-workflow/kb_workflow.py --base-path -python3 kb-workflow/kb_workflow.py --metadata -python3 kb-workflow/kb_workflow.py --content -python3 kb-workflow/kb_workflow.py --analyzed - -# 归档分析报告 -python3 kb-workflow/kb_workflow.py --archive-file "/path/to/report.md" -``` - ## 示例 ``` diff --git a/kb-workflow/kb_workflow.py b/kb-workflow/kb_workflow.py index 540b0ae..1851637 100644 --- a/kb-workflow/kb_workflow.py +++ b/kb-workflow/kb_workflow.py @@ -3,330 +3,24 @@ """ KB Workflow - Knowledge Base Workflow Orchestrator -知识库工作流辅助工具函数。 +知识库工作流管道入口。 -此模块提供查重、文件操作、Git 管理等工具函数， -供主代理在执行 SKILL.md 中定义的 workflow 时调用。 +此模块提供 process_command() 函数作为管道入口， +内部使用 kb_pipeline 包实现完整的内容处理流程。 + +使用方式: + python kb_workflow.py process "https://www.bilibili.com/video/BVxxx" + python kb_workflow.py process "https://..." -o /path/to/output """ -import os -import sys -import json import argparse -import subprocess -import shutil -import re -from datetime import datetime +import asyncio +import json +import sys from pathlib import Path -from typing import Dict, List, Optional, Any +from typing import Any, Dict, Optional -# ==================== 子技能映射 ==================== - -METADATA_AGENTS = { - "bilibili": "kb-bilibili-metadata", - "zhihu": "kb-zhihu-metadata", -} - -CONTENT_AGENTS = { - "bilibili": "kb-bilibili-content", - "zhihu": "kb-zhihu-content", -} - - -# ==================== 内容类型判断 ==================== - -# 需要下载视频/字幕的内容类型 -VIDEO_CONTENT_TYPES = {"video", "zvideo"} - -# 需要提取文字的内容类型 -TEXT_CONTENT_TYPES = {"article", "zhuanlan", "question", "answer"} - -# 需要下载图片的内容类型 -IMAGE_CONTENT_TYPES = {"note"} - - -def is_video_content(content_type: str) -> bool: - """判断是否为视频内容类型""" - return content_type in VIDEO_CONTENT_TYPES - - -def is_text_content(content_type: str) -> bool: - """判断是否为文字内容类型（专栏、文章、问答等）""" - return content_type in TEXT_CONTENT_TYPES - - -def is_image_content(content_type: str) -> bool: - """判断是否为图片内容类型（小红书等）""" - return content_type in IMAGE_CONTENT_TYPES - - -def get_content_handler(content_type: str) -> str: - """ - 获取内容处理方式 - - Returns: - "video" - 需要下载视频/字幕 - "text" - 需要提取文字 - "image" - 需要下载图片 - "unknown" - 未知类型 - """ - if is_video_content(content_type): - return "video" - elif is_text_content(content_type): - return "text" - elif is_image_content(content_type): - return "image" - else: - return "unknown" - - -# ==================== 工具函数 ==================== - -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 get_archive_path() -> Path: - """获取归档目录（AI分析文档存放位置）""" - archive_path_str = os.environ.get('KB_ARCHIVE_PATH', '') - if archive_path_str: - return Path(archive_path_str).expanduser() - # 默认：~/work/zq/30.资源/AI分析 - return Path.home() / "work" / "zq" / "30.资源" / "AI分析" - - -def get_git_root() -> Path: - """获取 Git 仓库根目录（用于归档提交）""" - git_root_str = os.environ.get('KB_GIT_ROOT', '') - if git_root_str: - return Path(git_root_str).expanduser() - # 默认：~/work/zq - return Path.home() / "work" / "zq" - - -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", - "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" - - # 从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 - - -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 = get_git_root() - 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. 检查并初始化 Git 仓库 - git_dir = git_root / ".git" - if not git_dir.exists(): - # 初始化仓库 - subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True) - subprocess.run(['git', 'config', 'user.name', git_user], cwd=str(git_root)) - subprocess.run(['git', 'config', 'user.email', git_email], cwd=str(git_root)) - - # 关联远端仓库 - if git_remote: - subprocess.run(['git', 'remote', 'add', 'origin', git_remote], cwd=str(git_root), capture_output=True) - # 拉取远端分支 - subprocess.run(['git', 'fetch', 'origin'], cwd=str(git_root), capture_output=True) - # 设置 main 分支并拉取 - subprocess.run(['git', 'branch', '--set-upstream-to=origin/main', 'main'], cwd=str(git_root), capture_output=True) - subprocess.run(['git', 'pull', 'origin', 'main', '--rebase'], cwd=str(git_root), capture_output=True, timeout=120) - else: - # 已有仓库，检查远端是否配置 - if git_remote: - remote_check = subprocess.run(['git', 'remote', '-v'], cwd=str(git_root), capture_output=True, text=True) - if 'origin' not in remote_check.stdout: - subprocess.run(['git', 'remote', 'add', 'origin', git_remote], cwd=str(git_root), capture_output=True) - - # 拉取远端最新代码 - subprocess.run(['git', 'fetch', 'origin'], cwd=str(git_root), capture_output=True) - # 尝试拉取 main 分支，如果失败则尝试 master - pull_result = subprocess.run(['git', 'pull', 'origin', 'main', '--rebase'], cwd=str(git_root), capture_output=True, timeout=120) - if pull_result.returncode != 0: - subprocess.run(['git', 'pull', 'origin', 'master', '--rebase'], cwd=str(git_root), capture_output=True, timeout=120) - - # 2. 创建归档目录 - date_str = datetime.now().strftime('%Y-%m-%d') - archive_base = get_archive_path() - date_dir = archive_base / date_str - date_dir.mkdir(parents=True, exist_ok=True) - - # 3. 复制分析报告 - src_file = Path(analyzed_file) - archive_file = date_dir / src_file.name - shutil.copy(src_file, archive_file) - result["archive_path"] = str(archive_file) - - # 4. 复制元数据 - meta_file = base_path / "metadata" / f"{src_file.stem}.meta.json" - if meta_file.exists(): - shutil.copy(meta_file, date_dir / meta_file.name) - - # 5. Git 提交并推送 - if (git_root / ".git").exists(): - # 添加所有变更 - subprocess.run(['git', 'add', '.'], cwd=str(git_root), capture_output=True) - - # 检查是否有变更需要提交 - status_result = subprocess.run(['git', 'status', '--porcelain'], cwd=str(git_root), capture_output=True, text=True) - - if status_result.stdout.strip(): - 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', 'main'], - cwd=str(git_root), capture_output=True, text=True, timeout=60 - ) - if push_result.returncode != 0: - # 如果 main 分支推送失败，尝试 master - 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, "output": push_result.stdout} - - 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 Process Command ==================== - def process_command( url: str, output_dir: Optional[str] = None, @@ -342,8 +36,6 @@ def process_command( 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: @@ -351,7 +43,7 @@ def process_command( try: from kb_pipeline.pipeline import Pipeline - from kb_pipeline.processors import ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor + from kb_pipeline.processors import ASRProcessor, AnalyzeProcessor, DownloadProcessor, ParseProcessor from kb_pipeline.schemas import PipelineConfig async def run_pipeline(): @@ -402,24 +94,22 @@ def process_command( } -# ==================== CLI ==================== - def main(): - """CLI入口 - 用于命令行测试""" + """CLI入口 - 用于命令行调用""" parser = argparse.ArgumentParser( description='KB Workflow - Knowledge Base Workflow Orchestrator', formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" + epilog=''' 使用示例: - # 查重 - 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" - """ + + # 指定输出目录 + python kb_workflow.py process "https://..." -o /path/to/output + + # 强制重新执行（忽略已完成状态） + python kb_workflow.py process "https://..." --no-skip + ''' ) # Create subparsers for commands @@ -430,29 +120,12 @@ def main(): 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)') + help='Skip completed steps (default: True)') process_parser.add_argument('--no-skip', action='store_false', dest='skip_completed', - help='Re-run all steps') + 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='归档的文件路径') - parser.add_argument('--base-path', type=str, help='获取知识库根目录') - - parser.add_argument('--pending', action='store_true', help='获取url解析目录') - parser.add_argument('--metadata', action='store_true', help='获取元数据目录') - parser.add_argument('--content', action='store_true', help='获取内容目录') - parser.add_argument('--content_downloads', action='store_true', help='获取主内容下载目录') - parser.add_argument('--analyzed', action='store_true', help='获取深度分析目录') - parser.add_argument('--archive_dir', action='store_true', help='获取归档目录') - 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( @@ -461,45 +134,6 @@ def main(): 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()