starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(kb-workflow): remove legacy code - use kb_pipeline only

Browse Source 
Root cause: kb_workflow.py contained 312 lines of deprecated legacy
code that was confusing AI agents. The legacy functions have been
completely replaced by the kb_pipeline package.

Changes:
- Removed all legacy functions (is_video_content, check_duplicate, etc.)
- Kept only process_command() and main() entry points
- Updated SKILL.md to remove old command examples
- File reduced from 495 to 143 lines (71% reduction)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
starmen1265
2026-03-16 15:27:31 +08:00
parent f05c93d2b1
commit 0f0011b4a6
2 changed files with 24 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kb-workflow/SKILL.md
+1 -32
View File
@@ -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"
```
## 示例
```
kb-workflow/kb_workflow.py
+23 -389
View File
@@ -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()