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

Merge pull request 'fix/ci-deploy-reset' (#16) from fix/ci-deploy-reset into main
Deploy kb-skills / deploy (push) Successful in 1s
Details

Browse Source 
Reviewed-on: http://192.168.1.22:8418/starmen1265/kb-skills/pulls/16
This commit was merged in pull request #16.
This commit is contained in:
starmen1265
2026-03-16 15:38:48 +08:00
parent ed1de1a5b1 f98d649b5c
commit 46aab1d9f4
3 changed files with 98 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.planning/debug/resolved/kb-workflow-legacy-code-removal.md
+74
View File
@@ -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
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()