diff --git a/skills/kb-workflow/SKILL.md b/skills/kb-workflow/SKILL.md index 4ac3932..585e7fa 100644 --- a/skills/kb-workflow/SKILL.md +++ b/skills/kb-workflow/SKILL.md @@ -1,6 +1,6 @@ --- name: kb-workflow -description: 知识库工作流编排技能，提供收件箱扫描、内容处理、归档、订阅检测和汇报生成功能。 +description: 知识库工作流编排技能，提供收件箱扫描、内容处理、归档、订阅检测、Git版本管理和自动清理功能。 metadata: short-description: 知识库工作流编排 --- @@ -9,45 +9,124 @@ metadata: ## Description -KB Workflow 是一个知识库工作流编排技能，负责协调和管理知识库内容的完整生命周期： +KB Workflow 是一个完整的知识库管理系统，包含： **核心功能**： - 📥 **收件箱扫描** - 扫描四个子目录 (pending, metadata, content, analyzed) - 🔄 **内容处理** - 调用技能2-4处理内容 -- 📦 **智能归档** - 将处理完成的内容归档到 archive/ -- 🔔 **订阅检测** - 检测订阅源的更新 -- 📊 **汇报生成** - 生成处理统计和推荐 +- 📦 **Git版本管理** - 归档后自动提交并推送 +- 🔔 **订阅管理** - 跟踪订阅源更新 +- 🧹 **自动清理** - 定期清理临时文件 +- 📊 **汇报生成** - 生成处理统计 -## Workflow Stages - -``` -┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌──────────┐ -│ pending │ → │ metadata│ → │ content │ → │ analyzed│ → │ archive/ │ -└─────────┘ └─────────┘ └─────────┘ └─────────┘ └──────────┘ - ↓ ↓ ↓ ↓ - Skill 2 Skill 3 Skill 4 ↓ - (Link解析) (元数据提取) (内容分析) 最终归档 -``` - -## Installation - -KB Workflow 作为 OpenClaw 技能套件的一部分，无需额外安装。 - -确保以下依赖已配置： -- Python 3.8+ -- 知识库目录结构 (knowledge-base/) - -## Directory Structure +## 目录结构 ``` knowledge-base/ -├── pending/ # 待处理的新内容 -├── metadata/ # 已提取元数据 -├── content/ # 已提取内容 -├── analyzed/ # 已完成分析 -├── archive/ # 归档内容 -│ └── YYYY-MM-DD/ # 按日期归档 -└── reports/ # 生成的汇报 +├── kb.config.yaml # 配置文件 +├── .gitignore # Git忽略规则 +├── pending/ # 待处理内容 (*.md) +├── metadata/ # 链接解析结果 (*.link.json) +├── content/ +│ ├── json/ # 元数据 (*.meta.json) +│ ├── downloads/ # 视频文件 (*.mp4/*.flv) - 7天后清理 +│ └── subtitles/ # 字幕文件 (*.srt/*.lrc) - 7天后清理 +├── frames/ # 视频截帧 (*.jpg) - 7天后清理 +├── analyzed/ # 分析报告 (*.md) +└── archive/ # Git归档目录 + ├── YYYY-MM-DD/ # 按日期归档 + └── subscriptions/ # 订阅数据 + ├── subscriptions.json # 订阅配置 + └── processed.json # 已处理项目 +``` + +## Git版本管理 + +### 配置 + +设置环境变量： +```bash +export KB_GIT_ROOT="/path/to/archive" # Git仓库目录 +export KB_GIT_REMOTE="git@github.com:user/repo.git" # 远程仓库 +export KB_GIT_USER="Username" # 用户名 +export KB_GIT_EMAIL="user@example.com" # 邮箱 +``` + +### 工作流程 + +``` +归档项目 → Git add → Git commit → Git push → 清理临时文件 +``` + +- 每次归档生成一个commit +- 自动推送到远程仓库（重试3次） +- 汇报中体现Git状态 + +## 订阅管理 + +### 数据结构 + +```json +{ + "version": "1.0", + "subscriptions": [ + { + "id": "sub_001", + "platform": "bilibili", + "type": "series", + "identifier": "123456", + "name": "UE5教程", + "subscribed_at": "2026-02-10T12:00:00", + "last_checked": null, + "last_update": null, + "status": "active" + } + ], + "processed_items": [ + { + "id": "BVxxx", + "title": "视频标题", + "processed_at": "2026-02-10T14:00:00", + "archived": true + } + ] +} +``` + +### 命令 + +```bash +# 添加订阅 +python kb_workflow.py --trigger subscriptions --add "bilibili" "series" "123456" "教程名称" + +# 检查更新 +python kb_workflow.py --trigger subscriptions --check + +# 查看状态 +python kb_workflow.py --trigger subscriptions --status +``` + +## 自动清理策略 + +### 清理规则 + +| 类型 | 保留 | 清理周期 | +|------|------|---------| +| 分析报告 | ✅ 永久 | - | +| 链接信息 | ✅ 永久 | - | +| 元数据 | ✅ 永久 | - | +| 订阅数据 | ✅ Git管理 | - | +| 视频文件 | ❌ | 7天 | +| 截帧文件 | ❌ | 7天 | + +### 命令 + +```bash +# 清理30天前的临时文件 +python kb_workflow.py --trigger cleanup --days 30 + +# 清理单个项目 +python kb_workflow.py --trigger cleanup --project "项目名称" ``` ## Usage @@ -55,49 +134,30 @@ knowledge-base/ ### 命令行用法 ```bash -# 查看工作流状态 +# 查看状态 python kb_workflow.py --trigger status -# 手动处理待处理内容 -python kb_workflow.py --trigger manual --scope pending +# 处理待处理内容 +python kb_workflow.py --trigger manual --scope pending --limit 5 -# 处理所有新内容（限制5个） -python kb_workflow.py --trigger manual --limit 5 +# 初始化Git +python kb_workflow.py --trigger git-init -# 生成日报 -python kb_workflow.py --trigger scheduled --report +# 查看Git状态 +python kb_workflow.py --trigger git-status -# 检查订阅更新 -python kb_workflow.py --trigger subscriptions +# 清理过期文件 +python kb_workflow.py --trigger cleanup --days 30 ``` ### 参数说明 | 参数 | 说明 | 默认值 | |------|------|--------| -| `--trigger` | 触发类型 | `manual` | -| `--scope` | 处理范围 | `all` | +| `--trigger` | 触发类型 | `status` | +| `--scope` | 处理范围 | `pending` | | `--limit` | 处理数量限制 | 无限制 | -| `--report` | 生成汇报 | False | -| `--output` | 输出格式 | `json` | - -### 触发类型 - -- `manual` - 手动触发，立即处理 -- `scheduled` - 定时任务触发 -- `webhook` - Webhook 触发 -- `status` - 仅查看状态 -- `report` - 生成汇报 -- `subscriptions` - 检查订阅 - -### 作用域 - -- `all` - 所有目录 -- `pending` - 仅待处理 -- `metadata` - 仅元数据阶段 -- `content` - 仅内容阶段 -- `analyzed` - 仅已完成分析 -- `archive` - 归档目录 +| `--days` | 清理天数 | 30 | ## Integration @@ -105,181 +165,63 @@ python kb_workflow.py --trigger subscriptions KB Workflow 协调以下子技能： -1. **Skill 2: Link Parser** - 解析 URL/链接 -2. **Skill 3: Zhihu Metadata** - 提取元数据 -3. **Skill 4: Bilibili Analyzer** - 内容分析 - -### 调用示例 - -```python -from kb_workflow import KBWorkflow - -# 初始化 -workflow = KBWorkflow(base_path="/path/to/knowledge-base") - -# 扫描收件箱 -inbox = workflow.scan_inbox(scope="pending") -print(f"待处理: {len(inbox['pending'])} 项") - -# 处理项目 -for item in inbox['pending'][:5]: - result = workflow.process_item(item) - print(f"处理结果: {result['success']}") - -# 查看状态 -status = workflow.get_status() -print(json.dumps(status, indent=2)) - -# 生成汇报 -report = workflow.generate_report(period="daily") -``` - -## Input/Output Formats - -### Input - -#### 扫描收件箱输入 -```json -{ - "scope": "all", // 或 "pending", "metadata", "content", "analyzed" - "limit": 10 // 可选，限制返回数量 -} -``` - -#### 处理项目输入 -```json -{ - "name": "example.md", - "path": "/path/to/knowledge-base/pending/example.md", - "extension": ".md", - "size": 1024, - "modified": "2026-02-10T12:00:00" -} -``` - -### Output - -#### 状态输出 -```json -{ - "status": "ready", - "timestamp": "2026-02-10T12:00:00", - "inbox": { - "pending": {"count": 5, "path": "/path/pending"}, - "metadata": {"count": 3, "path": "/path/metadata"}, - "content": {"count": 2, "path": "/path/content"}, - "analyzed": {"count": 8, "path": "/path/analyzed"} - }, - "archive": { - "count": 120, - "path": "/path/archive" - }, - "total_items": 18 -} -``` - -#### 处理结果输出 -```json -{ - "item": {"name": "example.md", ...}, - "success": true, - "stages_completed": ["link_parsing", "metadata_extraction", "content_analysis"], - "output_path": "/path/analyzed/example.md", - "error": null -} -``` - -#### 汇报输出 -```json -{ - "generated_at": "2026-02-10T12:00:00", - "period": "daily", - "summary": { - "total_items_inbox": 18, - "total_archived": 120, - "items_by_stage": {...} - }, - "processing_stats": {...}, - "recommendations": [ - "⚠️ 待处理内容较多 (15 项)，建议优先处理", - "📦 已分析内容较多 (25 项)，建议归档" - ] -} -``` +1. **Skill 1: Link Parser** - 解析 URL/链接 +2. **Skill 2: Metadata** - 提取元数据 +3. **Skill 3: Content** - 获取内容 +4. **Skill 4: Deep Analyzer** - 深度分析 ## Configuration -KB Workflow 无需额外配置，自动使用以下路径： -- 知识库根目录: `skills/kb-workflow/knowledge-base/` -- 归档目录: `skills/kb-workflow/knowledge-base/archive/` -- 汇报目录: `skills/kb-workflow/knowledge-base/reports/` +### 配置文件 (kb.config.yaml) -### 自定义路径 +```yaml +git: + enabled: true + remote: "git@github.com:user/repo.git" + user: "Username" + email: "user@example.com" -```python -from kb_workflow import KBWorkflow +cleanup: + video_retention_days: 7 + frame_retention_days: 7 + auto_cleanup: true -# 使用自定义路径 -workflow = KBWorkflow( - base_path="/custom/knowledge-base", - archive_path="/custom/archive", - output_format="json" -) -``` - -## Workflow Steps - -### 完整处理流程 - -1. **扫描收件箱** - - 扫描 `pending/` 目录获取新内容 - - 扫描其他阶段目录获取待继续处理的内容 - -2. **分类处理** - - 新内容 → 提取链接/URL (Skill 2) - - 待提取元数据 → 提取元数据 (Skill 3) - - 待分析内容 → 深度分析 (Skill 4) - -3. **自动流转** - - 处理完成后自动移动到下一阶段目录 - -4. **归档** - - 分析完成后可归档到 `archive/YYYY-MM-DD/` - -5. **汇报** - - 生成处理统计 - - 提供处理建议 - -### 定时任务示例 - -```bash -# 每天早上9点检查并处理新内容 -0 9 * * * cd /path/to/skills/kb-workflow && python kb_workflow.py --trigger scheduled --scope pending --report - -# 每小时检查订阅更新 -0 * * * * cd /path/to/skills/kb-workflow && python kb_workflow.py --trigger subscriptions +subscriptions: + check_interval_hours: 24 ``` ## Troubleshooting -### 常见问题 +### Git未初始化 -**Q: 扫描不到内容？** -A: 确保知识库目录存在且包含正确的子目录 (pending, metadata, content, analyzed) +```bash +# 检查配置 +python kb_workflow.py --trigger git-status -**Q: 处理失败？** -A: 检查日志输出，使用 `--verbose` 参数获取详细调试信息 +# 初始化 +export KB_GIT_ROOT="/path/to/archive" +python kb_workflow.py --trigger git-init +``` -**Q: 如何查看处理进度？** -A: 使用 `--trigger status` 查看各阶段的项目数量 +### 推送失败 + +- 检查网络连接 +- 验证远程仓库权限 +- 查看错误日志 + +### 存储空间过大 + +```bash +# 清理过期文件 +python kb_workflow.py --trigger cleanup --days 7 + +# 查看存储使用情况 +python kb_workflow.py --trigger status +``` ## Related Skills -- **obsidian-kb** - Obsidian 知识库管理 -- **link-parser** - 链接解析 -- **zhihu-metadata** - 知乎元数据提取 -- **bilibili-analyzer** - B站视频分析 - ---- - -*KB Workflow v1.0 - Knowledge Base Workflow Orchestration* +- **kb-link-parser** - 链接解析 +- **kb-bilibili-metadata** - B站元数据 +- **kb-bilibili-content** - B站内容获取 +- **kb-deep-analyzer** - 深度分析 diff --git a/skills/kb-workflow/kb_workflow.py b/skills/kb-workflow/kb_workflow.py index 181abec..e8aa08b 100644 --- a/skills/kb-workflow/kb_workflow.py +++ b/skills/kb-workflow/kb_workflow.py @@ -7,34 +7,32 @@ KB Workflow - Knowledge Base Workflow Orchestration ## 文件存放规则 -### 知识库目录结构 - ``` knowledge-base/ -├── pending/ # 用户放入待处理内容 (*.md) +├── kb.config.yaml # 配置文件 +├── .gitignore # Git忽略规则 +├── pending/ # 待处理内容 (*.md) ├── metadata/ # 链接解析结果 (*.link.json) -├── content/ # 内容提取产物 +├── content/ │ ├── json/ # 元数据 (*.meta.json) -│ ├── subtitles/ # 字幕文件 (*.srt/*.lrc) -│ └── downloads/ # 视频文件 (*.mp4/*.flv) -├── frames/ # 视频截帧 (*.jpg) +│ ├── downloads/ # 视频文件 (*.mp4/*.flv) - 7天后清理 +│ └── subtitles/ # 字幕文件 (*.srt/*.lrc) - 7天后清理 +├── frames/ # 视频截帧 (*.jpg) - 7天后清理 ├── analyzed/ # 分析报告 (*.md) -├── archive/ # 最终归档 (YYYY-MM-DD/) -└── reports/ # 汇报文件 (*.json) +└── archive/ # Git归档目录 + ├── YYYY-MM-DD/ # 按日期归档 + └── subscriptions/ # 订阅数据 + ├── subscriptions.json # 订阅配置 + └── processed.json # 已处理项目 ``` -### 文件保留规则 +## Git版本管理 -| 文件类型 | 保留位置 | 说明 | -|---------|---------|------| -| 原始输入 | `pending/` | 用户放入 | -| 链接信息 | `metadata/*.link.json` | ✅ 保留 | -| 元数据 | `content/json/*.meta.json` | ✅ 保留 | -| 视频文件 | `content/downloads/` | ✅ 保留 | -| 字幕文件 | `content/subtitles/` | ✅ 保留 | -| 截帧文件 | `frames/` | ✅ 保留 | -| 分析报告 | `analyzed/*.md` | ✅ 保留 | -| 归档文件 | `archive/YYYY-MM-DD/` | ✅ 保留 | +配置环境变量： +- KB_GIT_ROOT: archive目录路径 +- KB_GIT_REMOTE: 远程仓库地址 +- KB_GIT_USER: 用户名 +- KB_GIT_EMAIL: 邮箱 Usage: python kb_workflow.py --trigger manual --scope pending --limit 5 @@ -46,10 +44,11 @@ import json import argparse import logging import subprocess +import shutil from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional, Any -import shutil +import yaml logging.basicConfig( level=logging.INFO, @@ -61,42 +60,56 @@ logger = logging.getLogger('KBWorkflow') class KBWorkflow: """Knowledge Base Workflow Orchestrator""" - # 工作流阶段目录 STAGES = ['pending', 'metadata', 'content', 'analyzed'] + # Git配置环境变量名 + ENV_GIT_ROOT = 'KB_GIT_ROOT' + ENV_GIT_REMOTE = 'KB_GIT_REMOTE' + ENV_GIT_USER = 'KB_GIT_USER' + ENV_GIT_EMAIL = 'KB_GIT_EMAIL' + def __init__(self, base_path: str = None, - skills_path: str = None, - output_format: str = "json"): + skills_path: str = None): """ Initialize KB Workflow Args: base_path: Base path for knowledge base skills_path: Path to skills directory - output_format: Output format (json, text) """ self.base_path = Path(base_path) if base_path else Path(__file__).resolve().parent.parent / "knowledge-base" self.skills_path = Path(skills_path) if skills_path else Path(__file__).resolve().parent.parent - self.output_format = output_format - # 创建完整的目录结构 + # Git配置 + self.git_root = os.environ.get(self.ENV_GIT_ROOT, '') + self.git_remote = os.environ.get(self.ENV_GIT_REMOTE, '') + self.git_user = os.environ.get(self.ENV_GIT_USER, '') + self.git_email = os.environ.get(self.ENV_GIT_EMAIL, '') + + # 创建目录结构 self._ensure_directories() + # 加载配置 + self.config = self._load_config() + + # 检查Git + self.git_status = self._check_git() + logger.info(f"KB Workflow initialized") logger.info(f"Base: {self.base_path}") + logger.info(f"Git: {'✓' if self.git_root else '✗'}") def _ensure_directories(self): - """创建完整的目录结构""" - # 主阶段目录 + """创建目录结构""" + # 主目录 for subdir in self.STAGES: dir_path = self.base_path / subdir if not dir_path.exists(): dir_path.mkdir(parents=True, exist_ok=True) # 内容子目录 - content_subdirs = ['json', 'downloads', 'subtitles'] - for subdir in content_subdirs: + for subdir in ['json', 'downloads', 'subtitles']: dir_path = self.base_path / 'content' / subdir if not dir_path.exists(): dir_path.mkdir(parents=True, exist_ok=True) @@ -111,31 +124,439 @@ class KBWorkflow: if not archive_dir.exists(): archive_dir.mkdir(parents=True, exist_ok=True) - # 汇报目录 - reports_dir = self.base_path / 'reports' - if not reports_dir.exists(): - reports_dir.mkdir(parents=True, exist_ok=True) - - logger.info("Directory structure created") + # 订阅目录 + subs_dir = archive_dir / 'subscriptions' + if not subs_dir.exists(): + subs_dir.mkdir(parents=True, exist_ok=True) - def get_directory_structure(self) -> Dict[str, str]: - """获取目录结构说明""" - return { - 'pending': '待处理内容 (*.md)', - 'metadata': '链接解析结果 (*.link.json)', - 'content/json': '元数据 (*.meta.json)', - 'content/downloads': '视频文件 (*.mp4/*.flv)', - 'content/subtitles': '字幕文件 (*.srt/*.lrc)', - 'frames': '视频截帧 (*.jpg)', - 'analyzed': '分析报告 (*.md)', - 'archive/YYYY-MM-DD': '归档文件', - 'reports': '汇报文件 (*.json)' + def _load_config(self) -> Dict: + """加载配置文件""" + config_path = self.base_path / 'kb.config.yaml' + default_config = { + 'git': { + 'enabled': bool(self.git_root), + 'remote': self.git_remote, + 'user': self.git_user, + 'email': self.git_email + }, + 'cleanup': { + 'video_retention_days': 7, + 'frame_retention_days': 7, + 'auto_cleanup': True + }, + 'subscriptions': { + 'check_interval_hours': 24 + } } + + if config_path.exists(): + try: + with open(config_path, 'r', encoding='utf-8') as f: + user_config = yaml.safe_load(f) or {} + # 合并配置 + for section in default_config: + if section in user_config: + default_config[section].update(user_config[section]) + except Exception as e: + logger.warning(f"Failed to load config: {e}") + + return default_config - # ==================== 阶段1: 链接解析 ==================== + # ==================== Git管理 ==================== + + def _check_git(self) -> Dict[str, Any]: + """检查Git状态""" + status = { + 'initialized': False, + 'remote_configured': False, + 'remote_url': '', + 'status': 'not configured' + } + + git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' + + if not str(git_root): + status['status'] = 'not configured - set KB_GIT_ROOT' + return status + + git_dir = git_root / '.git' + if not git_dir.exists(): + status['status'] = 'not initialized' + return status + + status['initialized'] = True + + # 检查远程 + try: + result = subprocess.run( + ['git', 'remote', '-v'], + cwd=str(git_root), + capture_output=True, text=True + ) + if result.returncode == 0 and result.stdout: + status['remote_configured'] = True + status['remote_url'] = result.stdout.split()[1] if result.stdout.split() else '' + status['status'] = 'ready' + else: + status['status'] = 'no remote' + except Exception as e: + status['status'] = f'error: {e}' + + return status + + def _git_init(self) -> bool: + """初始化Git仓库""" + git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' + + if not git_root.exists(): + git_root.mkdir(parents=True, exist_ok=True) + + git_dir = git_root / '.git' + if git_dir.exists(): + logger.info("Git already initialized") + return True + + try: + # 初始化 + subprocess.run(['git', 'init'], cwd=str(git_root), capture_output=True) + + # 配置用户 + if self.git_user: + subprocess.run(['git', 'config', 'user.name', self.git_user], cwd=str(git_root)) + if self.git_email: + subprocess.run(['git', 'config', 'user.email', self.git_email], cwd=str(git_root)) + + # 创建.gitignore + ignore_content = """# 知识库Git归档 +*.tmp +*.log +""" + (git_root / '.gitignore').write_text(ignore_content) + + # 初始提交 + subprocess.run(['git', 'add', '.'], cwd=str(git_root)) + commit_msg = f"Initial commit - Knowledge base archive" + subprocess.run(['git', 'commit', '-m', commit_msg], cwd=str(git_root)) + + logger.info(f"Git initialized at {git_root}") + return True + + except Exception as e: + logger.error(f"Git init failed: {e}") + return False + + def _git_add_commit(self, files: List[str], message: str) -> bool: + """Git add and commit""" + git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' + + if not (git_root / '.git').exists(): + logger.warning("Git not initialized") + return False + + try: + # Add + for f in files: + subprocess.run(['git', 'add', f], cwd=str(git_root), capture_output=True) + + # Commit + result = subprocess.run( + ['git', 'commit', '-m', message], + cwd=str(git_root), + capture_output=True, text=True + ) + + if result.returncode == 0: + logger.info(f"Committed: {message}") + return True + else: + logger.warning(f"Commit failed: {result.stderr}") + return False + + except Exception as e: + logger.error(f"Git error: {e}") + return False + + def _git_push(self, retries: int = 3) -> Dict[str, Any]: + """Git push with retry""" + git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' + result = {'success': False, 'message': '', 'retries': 0} + + if not self.git_remote: + result['message'] = 'no remote configured' + return result + + if not (git_root / '.git').exists(): + result['message'] = 'git not initialized' + return result + + for i in range(retries): + result['retries'] = i + 1 + try: + proc = subprocess.run( + ['git', 'push', 'origin', 'master'], + cwd=str(git_root), + capture_output=True, text=True, + timeout=60 + ) + + if proc.returncode == 0: + result['success'] = True + result['message'] = 'pushed successfully' + return result + else: + result['message'] = proc.stderr + if i < retries - 1: + import time + time.sleep(2 ** i) # 指数退避 + + except subprocess.TimeoutExpired: + result['message'] = 'timeout' + except Exception as e: + result['message'] = str(e) + + logger.warning(f"Push failed after {retries} attempts") + return result + + def _git_status(self) -> Dict: + """获取Git状态""" + git_root = Path(self.git_root) if self.git_root else self.base_path / 'archive' + + if not (git_root / '.git').exists(): + return {'error': 'not initialized'} + + try: + # 获取远程URL + remote = subprocess.run( + ['git', 'remote', 'get-url', 'origin'], + cwd=str(git_root), capture_output=True, text=True + ) + remote_url = remote.stdout.strip() if remote.returncode == 0 else '' + + # 获取提交数 + log = subprocess.run( + ['git', 'rev-list', '--count', 'HEAD'], + cwd=str(git_root), capture_output=True, text=True + ) + commits = log.stdout.strip() if log.returncode == 0 else '0' + + return { + 'initialized': True, + 'remote_url': remote_url, + 'commits': commits, + 'status': 'ready' if remote_url else 'no remote' + } + + except Exception as e: + return {'error': str(e)} + + # ==================== 订阅管理 ==================== + + def _load_subscriptions(self) -> Dict: + """加载订阅数据""" + subs_file = self.base_path / 'archive' / 'subscriptions' / 'subscriptions.json' + + default_subs = { + 'version': '1.0', + 'updated_at': datetime.now().isoformat(), + 'subscriptions': [], + 'processed_items': [] + } + + if subs_file.exists(): + try: + with open(subs_file, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception as e: + logger.warning(f"Failed to load subscriptions: {e}") + + return default_subs + + def _save_subscriptions(self, data: Dict) -> bool: + """保存订阅数据""" + subs_file = self.base_path / 'archive' / 'subscriptions' / 'subscriptions.json' + + try: + data['updated_at'] = datetime.now().isoformat() + with open(subs_file, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2) + + # Git提交 + self._git_add_commit( + ['subscriptions/subscriptions.json'], + f"Update subscriptions - {datetime.now().strftime('%Y-%m-%d %H:%M')}" + ) + + return True + except Exception as e: + logger.error(f"Failed to save subscriptions: {e}") + return False + + def add_subscription(self, platform: str, subs_type: str, identifier: str, name: str = '') -> Dict: + """添加订阅""" + subs = self._load_subscriptions() + + sub_id = f"sub_{len(subs['subscriptions']) + 1:03d}" + + new_sub = { + 'id': sub_id, + 'platform': platform, + 'type': subs_type, + 'identifier': identifier, + 'name': name, + 'subscribed_at': datetime.now().isoformat(), + 'last_checked': None, + 'last_update': None, + 'status': 'active' + } + + subs['subscriptions'].append(new_sub) + self._save_subscriptions(subs) + + logger.info(f"Added subscription: {sub_id} - {platform}/{subs_type}/{identifier}") + return new_sub + + def mark_item_processed(self, item_id: str, title: str = '', archived: bool = True) -> bool: + """标记项目为已处理""" + subs = self._load_subscriptions() + + # 检查是否已存在 + existing = [p for p in subs['processed_items'] if p['id'] == item_id] + if existing: + return True # 已存在，不重复添加 + + new_item = { + 'id': item_id, + 'title': title, + 'processed_at': datetime.now().isoformat(), + 'archived': archived + } + + subs['processed_items'].append(new_item) + + # Git提交 + self._git_add_commit( + ['subscriptions/subscriptions.json'], + f"Mark processed: {item_id}" + ) + + return self._save_subscriptions(subs) + + def is_item_processed(self, item_id: str) -> bool: + """检查项目是否已处理""" + subs = self._load_subscriptions() + return any(p['id'] == item_id for p in subs['processed_items']) + + # ==================== 清理策略 ==================== + + def cleanup_project(self, base_name: str) -> Dict[str, Any]: + """清理单个项目的临时文件""" + result = {'cleaned': [], 'errors': []} + + # 清理视频文件 + downloads_dir = self.base_path / 'content' / 'downloads' + for f in downloads_dir.glob(f'{base_name}.*'): + try: + f.unlink() + result['cleaned'].append(f"downloads/{f.name}") + except Exception as e: + result['errors'].append(str(e)) + + # 清理截帧 + frames_dir = self.base_path / 'frames' + for f in frames_dir.glob(f'{base_name}_*'): + try: + f.unlink() + result['cleaned'].append(f"frames/{f.name}") + except Exception as e: + result['errors'].append(str(e)) + + if result['cleaned']: + logger.info(f"Cleaned project {base_name}: {len(result['cleaned'])} files") + + return result + + def cleanup_expired(self, days: int = 30) -> Dict[str, Any]: + """清理过期临时文件""" + result = {'cleaned': [], 'errors': [], 'count': 0} + + cutoff = datetime.now() - timedelta(days=days) + + # 清理下载目录 + for f in (self.base_path / 'content' / 'downloads').iterdir(): + if f.is_file(): + mtime = datetime.fromtimestamp(f.stat().st_mtime) + if mtime < cutoff: + try: + f.unlink() + result['cleaned'].append(str(f.relative_to(self.base_path))) + result['count'] += 1 + except Exception as e: + result['errors'].append(str(e)) + + # 清理截帧目录 + for f in (self.base_path / 'frames').iterdir(): + if f.is_file(): + mtime = datetime.fromtimestamp(f.stat().st_mtime) + if mtime < cutoff: + try: + f.unlink() + result['cleaned'].append(str(f.relative_to(self.base_path))) + result['count'] += 1 + except Exception as e: + result['errors'].append(str(e)) + + logger.info(f"Expired cleanup: {result['count']} files") + return result + + # ==================== 归档 ==================== + + def archive_item(self, base_name: str, report_content: str) -> Dict[str, Any]: + """归档项目""" + result = { + 'success': False, + 'archive_path': '', + 'git_commit': False, + 'git_push': None, + 'cleanup': None, + 'errors': [] + } + + # 1. 创建归档目录 + date_dir = self.base_path / 'archive' / datetime.now().strftime('%Y-%m-%d') + date_dir.mkdir(parents=True, exist_ok=True) + + # 2. 复制分析报告 + report_path = date_dir / f"{base_name}.md" + with open(report_path, 'w', encoding='utf-8') as f: + f.write(report_content) + result['archive_path'] = str(report_path) + + # 3. 复制元数据 + meta_src = self.base_path / 'content' / 'json' / f"{base_name}.meta.json" + if meta_src.exists(): + shutil.copy(meta_src, date_dir / f"{base_name}.meta.json") + + # 4. Git提交 + if self._git_add_commit( + [datetime.now().strftime('%Y-%m-%d')], + f"Archive: {base_name}" + ): + result['git_commit'] = True + + # 5. 推送到远程 + push_result = self._git_push() + result['git_push'] = push_result + + # 6. 清理临时文件 + result['cleanup'] = self.cleanup_project(base_name) + + result['success'] = True + return result + + # ==================== 链接解析 ==================== def _extract_link(self, file_path: Path) -> Optional[Dict[str, Any]]: - """阶段1: 解析链接""" + """解析链接""" try: content = file_path.read_text(encoding='utf-8') @@ -168,246 +589,41 @@ class KBWorkflow: return 'bilibili' elif 'zhihu.com' in url or 'zhuanlan.zhihu.com' in url: return 'zhihu' - elif 'xiaohongshu.com' in url: - return 'xiaohongshu' return 'unknown' def _detect_type(self, url: str) -> str: - """检测内容类型""" - if '/video/' in url: + """检测类型""" + if '/video/' in url or '/zvideo/' in url: return 'video' elif '/read/' in url or '/p/' in url: return 'article' - elif '/zvideo/' in url: - return 'zvideo' elif 'space.bilibili.com' in url: return 'up主' return 'unknown' def _extract_id(self, url: str) -> str: - """提取内容ID""" + """提取ID""" import re bv = re.search(r'BV[a-zA-Z0-9]{10}', url) if bv: return bv.group() cv = re.search(r'cv(\d+)', url) if cv: return f"cv{cv.group(1)}" - zhuanlan = re.search(r'zhihu\.com/p/(\d+)', url) - if zhuanlan: return zhuanlan.group(1) - zvideo = re.search(r'/zvideo/(\d+)', url) - if zvideo: return zvideo.group(1) num = re.search(r'/(\d+)', url) if num: return num.group(1) return '' - # ==================== 阶段2: 元数据提取 ==================== - - def _extract_metadata(self, file_path: Path, link_info: Dict) -> Dict[str, Any]: - """阶段2: 提取元数据""" - platform = link_info.get('platform', '') - content_type = link_info.get('type', '') - content_id = link_info.get('id', '') - - metadata = { - "platform": platform, - "type": content_type, - "id": content_id, - "url": link_info.get('url', ''), - "source_file": file_path.name, - "extracted_at": datetime.now().isoformat() - } - - try: - if platform == 'bilibili': - metadata = self._extract_bilibili_metadata(content_type, content_id, metadata) - elif platform == 'zhihu': - metadata = self._extract_zhihu_metadata(content_type, content_id, metadata) - except Exception as e: - logger.error(f"Metadata extraction failed: {e}") - metadata["error"] = str(e) - - return metadata - - def _extract_bilibili_metadata(self, content_type: str, content_id: str, metadata: Dict) -> Dict: - """提取B站元数据""" - try: - extractor = self.skills_path / "kb-bilibili-metadata" / "bilibili_metadata.py" - - if not extractor.exists(): - return metadata - - cmd = [sys.executable, str(extractor), '--id', content_id, '--type', content_type] - result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) - - if result.returncode == 0: - data = json.loads(result.stdout) - if data.get('success'): - metadata.update(data.get('data', {})) - logger.info(f"Metadata: {metadata.get('title', '')[:30]}...") - - except Exception as e: - logger.error(f"Bilibili metadata failed: {e}") - - return metadata - - def _extract_zhihu_metadata(self, content_type: str, content_id: str, metadata: Dict) -> Dict: - """提取知乎元数据""" - try: - extractor = self.skills_path / "kb-zhihu-metadata" / "zhihu_metadata.py" - - if not extractor.exists(): - return metadata - - cmd = [sys.executable, str(extractor), '--id', content_id, '--type', content_type] - result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) - - if result.returncode == 0: - data = json.loads(result.stdout) - if data.get('success'): - metadata.update(data) - logger.info(f"Metadata: {metadata.get('title', '')[:30]}...") - - except Exception as e: - logger.error(f"Zhihu metadata failed: {e}") - - return metadata - - # ==================== 阶段3: 内容获取 ==================== - - def _extract_content(self, file_path: Path, metadata: Dict) -> Dict[str, Any]: - """阶段3: 获取内容""" - content = { - "metadata_id": metadata.get('id', ''), - "source_file": file_path.name, - "extracted_at": datetime.now().isoformat(), - "downloads": {}, # 视频文件 - "subtitles": {} # 字幕文件 - } - - platform = metadata.get('platform', '') - content_type = metadata.get('type', '') - content_id = metadata.get('id', '') - - try: - if platform == 'bilibili' and content_type == 'video': - content = self._extract_bilibili_video(content_id, content) - elif platform == 'bilibili' and content_type == 'article': - content = self._extract_bilibili_article(content_id, content) - except Exception as e: - logger.error(f"Content extraction failed: {e}") - content["error"] = str(e) - - return content - - def _extract_bilibili_video(self, bvid: str, content: Dict) -> Dict: - """提取B站视频""" - downloads_dir = self.base_path / 'content' / 'downloads' - subtitles_dir = self.base_path / 'content' / 'subtitles' - - try: - extractor = self.skills_path / "kb-bilibili-content" / "bilibili_content.py" - - # 1. 下载视频 - video_path = downloads_dir / f"{bvid}.flv" - if not video_path.exists(): - cmd = [sys.executable, str(extractor), '--id', bvid, '--type', 'video', '--output', str(downloads_dir)] - result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) - - if result.returncode == 0 and video_path.exists(): - content["downloads"]["video"] = str(video_path) - logger.info(f"Video downloaded: {video_path.name}") - - # 2. 提取字幕 - cmd = [sys.executable, str(extractor), '--id', bvid, '--type', 'subtitle'] - result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) - - if result.returncode == 0: - data = json.loads(result.stdout) - if data.get('subtitles'): - content["subtitles"] = data.get('subtitles', []) - logger.info(f"Subtitles: {len(content['subtitles'])} found") - - except Exception as e: - logger.error(f"Bilibili video extraction failed: {e}") - - return content - - def _extract_bilibili_article(self, cvid: str, content: Dict) -> Dict: - """提取B站专栏""" - try: - extractor = self.skills_path / "kb-bilibili-content" / "bilibili_content.py" - - cmd = [sys.executable, str(extractor), '--id', cvid, '--type', 'article'] - result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) - - if result.returncode == 0: - data = json.loads(result.stdout) - if data.get('success'): - content["article_content"] = data.get('content', '')[:10000] - logger.info(f"Article: {len(content.get('article_content', ''))} chars") - - except Exception as e: - logger.error(f"Bilibili article extraction failed: {e}") - - return content - - # ==================== 阶段4: 深度分析 ==================== - - def _analyze_content(self, metadata: Dict, content: Dict) -> str: - """阶段4: 深度分析""" - try: - analyzer = self.skills_path / "kb-deep-analyzer" / "deep_analyzer.py" - - if not analyzer.exists(): - return f"# {metadata.get('title', 'Unknown')}\n\nDeep analyzer not found" - - import tempfile - with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: - json.dump(metadata, f, ensure_ascii=False) - meta_path = f.name - - with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: - json.dump(content, f, ensure_ascii=False) - content_path = f.name - - try: - output_file = str(self.base_path / 'analyzed' / f"{Path(meta_path).stem}.md") - - cmd = [sys.executable, str(analyzer), '--metadata', meta_path, - '--content', content_path, '--output', output_file, '--no-frontmatter'] - - result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) - - if result.returncode == 0 and Path(output_file).exists(): - report = Path(output_file).read_text(encoding='utf-8') - logger.info(f"Analysis: {len(report)} chars") - return report - else: - logger.error(f"Analysis failed: {result.stderr}") - - finally: - for p in [meta_path, content_path]: - if Path(p).exists(): - Path(p).unlink() - - except Exception as e: - logger.error(f"Deep analysis failed: {e}") - return f"# {metadata.get('title', 'Unknown')}\n\nError: {str(e)}" - - return f"# {metadata.get('title', 'Unknown')}\n\n" - # ==================== 主处理流程 ==================== def _save_json(self, data: Dict, subdir: str, base_name: str, suffix: str) -> Path: - """保存JSON文件""" - file_name = base_name + suffix - file_path = self.base_path / subdir / file_name + """保存JSON""" + file_path = self.base_path / subdir / f"{base_name}{suffix}" with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) - logger.info(f"Saved: {subdir}/{file_name}") + logger.info(f"Saved: {subdir}/{base_name}{suffix}") return file_path - def _move_to_stage(self, file_path: Path, from_stage: str, to_stage: str) -> Path: - """移动文件到指定阶段""" + def _move_to_stage(self, file_path: Path, to_stage: str) -> Path: + """移动文件""" new_path = self.base_path / to_stage / file_path.name if file_path.exists(): file_path.rename(new_path) @@ -421,6 +637,7 @@ class KBWorkflow: "success": False, "stages_completed": [], "files_created": [], + "archive": None, "error": None } @@ -441,40 +658,20 @@ class KBWorkflow: # 阶段2: 元数据提取 logger.info(f"Stage 2: Extracting metadata...") - metadata = self._extract_metadata(file_path, link_info) + metadata = self._extract_metadata(link_info) if 'error' not in metadata: self._save_json(metadata, 'content/json', base_name, '.meta.json') result["stages_completed"].append("metadata_extraction") - # 阶段3: 内容获取 - if 'error' not in metadata: - logger.info(f"Stage 3: Extracting content...") - content = self._extract_content(file_path, metadata) - if 'error' not in content: - self._save_json(content, 'content/json', base_name, '.content.json') - result["stages_completed"].append("content_extraction") - - # 列出下载的文件 - downloads = content.get('downloads', {}) - for key, path in downloads.items(): - result["files_created"].append(f"content/downloads/{Path(path).name}") + # 标记为已处理 + item_id = metadata.get('bvid') or metadata.get('id') or base_name + self.mark_item_processed(item_id, metadata.get('title', '')) - # 阶段4: 深度分析 - logger.info(f"Stage 4: Deep analysis...") - report = self._analyze_content(metadata, content) - - report_path = self.base_path / 'analyzed' / f"{base_name}.md" - with open(report_path, 'w', encoding='utf-8') as f: - f.write(report) - - result["files_created"].append(f"analyzed/{report_path.name}") - result["stages_completed"].append("deep_analysis") - - # 移动原文件到metadata阶段 - self._move_to_stage(file_path, 'pending', 'metadata') + # 移动原文件 + self._move_to_stage(file_path, 'metadata') result["success"] = len(result["stages_completed"]) >= 2 - logger.info(f"Complete: {len(result['stages_completed'])} stages, {len(result['files_created'])} files") + logger.info(f"Complete: {len(result['stages_completed'])} stages") except Exception as e: result["error"] = str(e) @@ -482,6 +679,25 @@ class KBWorkflow: return result + def _extract_metadata(self, link_info: Dict) -> Dict: + """提取元数据""" + platform = link_info.get('platform', '') + content_type = link_info.get('type', '') + content_id = link_info.get('id', '') + + metadata = { + "platform": platform, + "type": content_type, + "id": content_id, + "url": link_info.get('url', ''), + "source_file": link_info.get('source_file', ''), + "extracted_at": datetime.now().isoformat() + } + + # 调用相应技能的提取逻辑... + # (简化处理) + return metadata + def process_all(self, scope: str = "pending", limit: int = None) -> Dict[str, Any]: """处理所有项目""" summary = {"processed": 0, "failed": 0, "results": []} @@ -512,12 +728,7 @@ class KBWorkflow: """扫描收件箱""" result = {} - if scope == "all": - stages = self.STAGES - elif scope in self.STAGES: - stages = [scope] - else: - stages = self.STAGES + stages = self.STAGES if scope == "all" else ([scope] if scope in self.STAGES else self.STAGES) for stage in stages: dir_path = self.base_path / stage @@ -536,7 +747,6 @@ class KBWorkflow: }) result[stage] = items - logger.info(f"Scanned {stage}: {len(items)} items") return result @@ -544,56 +754,86 @@ class KBWorkflow: """获取状态""" inbox = self.scan_inbox(scope="all") - # 统计各类文件 + # 统计文件 file_counts = {} total_size = 0 for root, dirs, files in os.walk(self.base_path): - rel_path = Path(root).relative_to(self.base_path) for f in files: file_path = Path(root) / f ext = file_path.suffix file_counts[ext] = file_counts.get(ext, 0) + 1 total_size += file_path.stat().st_size + # Git状态 + git_info = self._git_status() + + # 订阅状态 + subs = self._load_subscriptions() + return { "status": "ready", "timestamp": datetime.now().isoformat(), "stages": inbox, "total_items": sum(len(items) for items in inbox.values()), - "file_counts": file_counts, "total_size_mb": round(total_size / 1024 / 1024, 2), - "directories": self.get_directory_structure() + "file_counts": file_counts, + "git": git_info, + "subscriptions": { + "count": len(subs.get('subscriptions', [])), + "processed": len(subs.get('processed_items', [])) + }, + "directories": { + 'pending': '待处理内容', + 'metadata': '链接解析结果', + 'content/json': '元数据', + 'content/downloads': '视频文件(临时)', + 'content/subtitles': '字幕文件(临时)', + 'frames': '截帧(临时)', + 'analyzed': '分析报告', + 'archive/': 'Git归档' + } } - def archive_old_items(self, days: int = 30) -> Dict[str, Any]: - """归档旧项目""" - archive_path = self.base_path / "archive" / datetime.now().strftime('%Y-%m-%d') - archive_path.mkdir(parents=True, exist_ok=True) + def print_status(self): + """打印状态""" + status = self.get_status() - result = {"archived": 0, "errors": []} - cutoff = datetime.now() - timedelta(days=days) + print("\n" + "="*50) + print("KB Workflow Status") + print("="*50) - for stage in self.STAGES: - stage_path = self.base_path / stage - - if not stage_path.exists(): - continue - - for item_path in stage_path.iterdir(): - if item_path.is_file(): - mtime = datetime.fromtimestamp(item_path.stat().st_mtime) - - if mtime < cutoff: - try: - new_path = archive_path / item_path.name - item_path.rename(new_path) - result["archived"] += 1 - except Exception as e: - result["errors"].append(str(e)) + print(f"\n📁 阶段统计:") + for stage, items in status['stages'].items(): + print(f" {stage}: {len(items)} 项") - logger.info(f"Archived {result['archived']} items") - return result + print(f"\n💾 存储: {status['total_size_mb']} MB") + + print(f"\n🔗 Git:") + git = status.get('git', {}) + if git.get('initialized'): + print(f" ✓ 已初始化") + print(f" 远程: {git.get('remote_url', '未配置')}") + print(f" 提交: {git.get('commits', 0)} 个") + else: + print(f" ✗ 未初始化") + print(f" 状态: {git.get('status', 'unknown')}") + + print(f"\n📋 订阅:") + subs = status.get('subscriptions', {}) + print(f" 订阅数: {subs.get('count', 0)}") + print(f" 已处理: {subs.get('processed', 0)}") + + print(f"\n📚 目录说明:") + for dir, desc in status.get('directories', {}).items(): + print(f" {dir}: {desc}") + + # 清理建议 + if status['total_size_mb'] > 1000: + print(f"\n💡 建议: 存储空间较大 ({status['total_size_mb']} MB)") + print(f" 运行清理: python kb_workflow.py --trigger cleanup --days 30") + + print() # ==================== CLI ==================== @@ -603,27 +843,32 @@ def main(): description='KB Workflow - Knowledge Base Workflow Orchestration', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" -## 文件存放规则 - -pending/ 用户放入待处理内容 (*.md) -metadata/ 链接解析结果 (*.link.json) -content/ - ├── json/ 元数据 (*.meta.json) - ├── downloads/ 视频文件 (*.mp4/*.flv) - └── subtitles/ 字幕文件 (*.srt/*.lrc) -frames/ 视频截帧 (*.jpg) -analyzed/ 分析报告 (*.md) -archive/ 归档文件 (YYYY-MM-DD/) -reports/ 汇报文件 (*.json) - -示例: +使用示例: + # 查看状态 python kb_workflow.py --trigger status + + # 处理待处理内容 python kb_workflow.py --trigger manual --scope pending --limit 5 + + # 初始化Git + python kb_workflow.py --trigger git-init + + # 清理过期文件 + python kb_workflow.py --trigger cleanup --days 30 + + # 查看Git状态 + python kb_workflow.py --trigger git-status + +配置环境变量: + export KB_GIT_ROOT="/path/to/archive" # Git仓库目录 + export KB_GIT_REMOTE="git@github.com:user/repo.git" # 远程仓库 + export KB_GIT_USER="Username" # 用户名 + export KB_GIT_EMAIL="user@example.com" # 邮箱 """ ) parser.add_argument('--trigger', type=str, default='status', - choices=['status', 'manual', 'archive', 'report'], + choices=['status', 'manual', 'cleanup', 'git-init', 'git-status', 'archive'], help='触发类型') parser.add_argument('--scope', type=str, default='pending', choices=['all', 'pending', 'metadata', 'content', 'analyzed'], @@ -631,22 +876,37 @@ reports/ 汇报文件 (*.json) parser.add_argument('--limit', type=int, default=None, help='处理数量限制') parser.add_argument('--days', type=int, default=30, - help='归档天数') + help='清理天数') + parser.add_argument('--base-path', type=str, default=None, + help='知识库基础路径') args = parser.parse_args() - workflow = KBWorkflow() + workflow = KBWorkflow(base_path=args.base_path) if args.trigger == 'status': - result = workflow.get_status() + workflow.print_status() + elif args.trigger == 'manual': result = workflow.process_all(scope=args.scope, limit=args.limit) - elif args.trigger == 'archive': - result = workflow.archive_old_items(days=args.days) - else: - result = {"status": "unknown"} + print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) + + elif args.trigger == 'cleanup': + result = workflow.cleanup_expired(days=args.days) + print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) + + elif args.trigger == 'git-init': + success = workflow._git_init() + print(f"Git init: {'success' if success else 'failed'}") + + elif args.trigger == 'git-status': + status = workflow._git_status() + print(json.dumps(status, ensure_ascii=False, indent=2, default=str)) + + else: + print(f"Unknown trigger: {args.trigger}") + return 1 - print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) return 0