diff --git a/skills/kb-workflow/kb_workflow.py b/skills/kb-workflow/kb_workflow.py index b6e636d..181abec 100644 --- a/skills/kb-workflow/kb_workflow.py +++ b/skills/kb-workflow/kb_workflow.py @@ -3,19 +3,41 @@ """ KB Workflow - Knowledge Base Workflow Orchestration -工作流编排技能，整合以下功能： -- 扫描收件箱目录 (pending, metadata, content, analyzed) -- 调用技能2-4处理内容 -- 归档处理完成的条目 -- 订阅更新检测 -- 生成汇报 +知识库工作流编排技能。 + +## 文件存放规则 + +### 知识库目录结构 + +``` +knowledge-base/ +├── pending/ # 用户放入待处理内容 (*.md) +├── metadata/ # 链接解析结果 (*.link.json) +├── content/ # 内容提取产物 +│ ├── json/ # 元数据 (*.meta.json) +│ ├── subtitles/ # 字幕文件 (*.srt/*.lrc) +│ └── downloads/ # 视频文件 (*.mp4/*.flv) +├── frames/ # 视频截帧 (*.jpg) +├── analyzed/ # 分析报告 (*.md) +├── archive/ # 最终归档 (YYYY-MM-DD/) +└── reports/ # 汇报文件 (*.json) +``` + +### 文件保留规则 + +| 文件类型 | 保留位置 | 说明 | +|---------|---------|------| +| 原始输入 | `pending/` | 用户放入 | +| 链接信息 | `metadata/*.link.json` | ✅ 保留 | +| 元数据 | `content/json/*.meta.json` | ✅ 保留 | +| 视频文件 | `content/downloads/` | ✅ 保留 | +| 字幕文件 | `content/subtitles/` | ✅ 保留 | +| 截帧文件 | `frames/` | ✅ 保留 | +| 分析报告 | `analyzed/*.md` | ✅ 保留 | +| 归档文件 | `archive/YYYY-MM-DD/` | ✅ 保留 | Usage: - python kb_workflow.py --trigger manual - python kb_workflow.py --trigger scheduled - python kb_workflow.py --status - python kb_workflow.py --report - python kb_workflow.py --scope pending --limit 5 + python kb_workflow.py --trigger manual --scope pending --limit 5 """ import os @@ -23,12 +45,12 @@ import sys import json import argparse import logging +import subprocess from datetime import datetime, timedelta from pathlib import Path from typing import Dict, List, Optional, Any -from enum import Enum +import shutil -# Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' @@ -36,638 +58,596 @@ logging.basicConfig( logger = logging.getLogger('KBWorkflow') -class TriggerType(Enum): - MANUAL = "manual" - SCHEDULED = "scheduled" - WEBHOOK = "webhook" - - -class Scope(Enum): - ALL = "all" - PENDING = "pending" - METADATA = "metadata" - CONTENT = "content" - ANALYZED = "analyzed" - ARCHIVE = "archive" - - class KBWorkflow: """Knowledge Base Workflow Orchestrator""" - # Base directories for the inbox structure - INBOX_DIRS = ['pending', 'metadata', 'content', 'analyzed'] + # 工作流阶段目录 + STAGES = ['pending', 'metadata', 'content', 'analyzed'] def __init__(self, base_path: str = None, - archive_path: str = None, + skills_path: str = None, output_format: str = "json"): """ Initialize KB Workflow Args: - base_path: Base path for knowledge base (default: workspace/knowledge-base) - archive_path: Path for archived items (default: archive/) + 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__).parent / "knowledge-base" - self.archive_path = Path(archive_path) if archive_path else self.base_path / "archive" + 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 - # Ensure directories exist + # 创建完整的目录结构 self._ensure_directories() - # Initialize subsystems - self._init_subsystems() + logger.info(f"KB Workflow initialized") + logger.info(f"Base: {self.base_path}") def _ensure_directories(self): - """Create necessary directories""" - # Create main inbox directories - for subdir in self.INBOX_DIRS: + """创建完整的目录结构""" + # 主阶段目录 + for subdir in self.STAGES: dir_path = self.base_path / subdir if not dir_path.exists(): dir_path.mkdir(parents=True, exist_ok=True) - logger.info(f"Created directory: {dir_path}") - # Create archive directory - if not self.archive_path.exists(): - self.archive_path.mkdir(parents=True, exist_ok=True) - logger.info(f"Created archive directory: {self.archive_path}") + # 内容子目录 + content_subdirs = ['json', 'downloads', 'subtitles'] + for subdir in content_subdirs: + dir_path = self.base_path / 'content' / subdir + if not dir_path.exists(): + dir_path.mkdir(parents=True, exist_ok=True) + + # 截帧目录 + frames_dir = self.base_path / 'frames' + if not frames_dir.exists(): + frames_dir.mkdir(parents=True, exist_ok=True) + + # 归档目录 + archive_dir = self.base_path / 'archive' + 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") - def _init_subsystems(self): - """Initialize subsystem references""" - # Placeholder for skill integrations - self.subscriptions_file = self.base_path / "subscriptions.json" - self.report_file = self.base_path / "reports" / f"report_{datetime.now().strftime('%Y%m%d')}.json" - - # ==================== Inbox Scanning ==================== - - def scan_inbox(self, scope: str = "all", limit: int = None) -> Dict[str, List[Dict]]: - """ - Scan inbox directories for items - - Args: - scope: Which directories to scan (all, pending, metadata, content, analyzed) - limit: Maximum number of items to return per directory - - Returns: - Dictionary with directory names as keys and item lists as values - """ - result = {} - - if scope == "all": - dirs_to_scan = self.INBOX_DIRS - elif scope in self.INBOX_DIRS: - dirs_to_scan = [scope] - else: - logger.warning(f"Unknown scope: {scope}, scanning all") - dirs_to_scan = self.INBOX_DIRS - - for dir_name in dirs_to_scan: - dir_path = self.base_path / dir_name - items = self._scan_directory(dir_path) - - if limit: - items = items[:limit] - - result[dir_name] = items - logger.info(f"Scanned {dir_name}: {len(items)} items") - - return result - - def _scan_directory(self, path: Path) -> List[Dict]: - """ - Scan a directory for items - - Args: - path: Directory path to scan - - Returns: - List of item dictionaries - """ - items = [] - - if not path.exists(): - return items - - for item_path in path.iterdir(): - if item_path.is_file(): - item_info = { - 'name': item_path.name, - 'path': str(item_path), - 'extension': item_path.suffix, - 'size': item_path.stat().st_size, - 'modified': datetime.fromtimestamp(item_path.stat().st_mtime).isoformat(), - 'type': self._classify_file_type(item_path) - } - items.append(item_info) - elif item_path.is_dir(): - # Treat subdirectories as items too - item_info = { - 'name': item_path.name, - 'path': str(item_path), - 'type': 'directory', - 'modified': datetime.fromtimestamp(item_path.stat().st_mtime).isoformat() - } - items.append(item_info) - - return items - - def _classify_file_type(self, path: Path) -> str: - """Classify file type by extension""" - suffix = path.suffix.lower() - - type_mapping = { - '.md': 'markdown', - '.txt': 'text', - '.json': 'json', - '.yaml': 'yaml', - '.yml': 'yaml', - '.pdf': 'pdf', - '.jpg': 'image', - '.jpeg': 'image', - '.png': 'image', - '.mp4': 'video', - '.mp3': 'audio', - '.zip': 'archive', - '.url': 'link' - } - - return type_mapping.get(suffix, 'unknown') - - def get_status(self) -> Dict[str, Any]: - """ - Get current workflow status - - Returns: - Status dictionary - """ - inbox_status = {} - total_items = 0 - - for dir_name in self.INBOX_DIRS: - dir_path = self.base_path / dir_name - items = self._scan_directory(dir_path) - inbox_status[dir_name] = { - 'count': len(items), - 'path': str(dir_path) - } - total_items += len(items) - - # Check archive - archive_items = self._scan_directory(self.archive_path) - + def get_directory_structure(self) -> Dict[str, str]: + """获取目录结构说明""" return { - 'status': 'ready', - 'timestamp': datetime.now().isoformat(), - 'inbox': inbox_status, - 'archive': { - 'count': len(archive_items), - 'path': str(self.archive_path) - }, - 'total_items': total_items + '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)' } - # ==================== Item Processing ==================== + # ==================== 阶段1: 链接解析 ==================== - def process_item(self, item: Dict, target_stage: str = None) -> Dict[str, Any]: - """ - Process a single item through the workflow - - This method calls skills 2-4 to process items: - - Skill 2: Link parsing / URL extraction - - Skill 3: Metadata extraction - - Skill 4: Content analysis - - Args: - item: Item dictionary with path and metadata - target_stage: Target stage to process to + def _extract_link(self, file_path: Path) -> Optional[Dict[str, Any]]: + """阶段1: 解析链接""" + try: + content = file_path.read_text(encoding='utf-8') - Returns: - Processing result dictionary - """ - result = { - 'item': item, - 'success': False, - 'stages_completed': [], - 'output_path': None, - 'error': None + import re + url_pattern = r'https?://[^\s<>"]+' + urls = re.findall(url_pattern, content) + + if not urls: + return None + + url = urls[0] + result = { + "platform": self._detect_platform(url), + "type": self._detect_type(url), + "id": self._extract_id(url), + "url": url, + "original_content": content + } + + logger.info(f"Parsed: {result['platform']}/{result['type']} - {result['id']}") + return result + + except Exception as e: + logger.error(f"Link extraction failed: {e}") + return None + + def _detect_platform(self, url: str) -> str: + """检测平台""" + if 'bilibili.com' in url or 'b23.tv' in url: + 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: + 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""" + 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: - item_path = Path(item['path']) - - # Determine processing stage based on current location - current_stage = self._get_stage_from_path(item_path) - - # Skill 2: Link/URL Parsing - if current_stage in ['pending', None] or target_stage == 'metadata': - logger.info(f"Stage 1 → 2: Processing link/URL for {item['name']}") - # TODO: Integrate with link-parser skill - result['stages_completed'].append('link_parsing') - - # Skill 3: Metadata Extraction - if current_stage in ['metadata', 'pending'] or target_stage == 'content': - logger.info(f"Stage 2 → 3: Extracting metadata for {item['name']}") - # TODO: Integrate with zhihu-metadata skill - result['stages_completed'].append('metadata_extraction') - - # Skill 4: Content Analysis - if current_stage in ['content', 'metadata'] or target_stage == 'analyzed': - logger.info(f"Stage 3 → 4: Analyzing content for {item['name']}") - # TODO: Integrate with bilibili-analyzer skill - result['stages_completed'].append('content_analysis') - - # Move to next stage if successful - if result['stages_completed']: - self._move_to_next_stage(item_path, result['stages_completed'][-1]) - result['success'] = True - + 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: - result['error'] = str(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 + 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}") + return file_path + + def _move_to_stage(self, file_path: Path, from_stage: str, to_stage: str) -> Path: + """移动文件到指定阶段""" + new_path = self.base_path / to_stage / file_path.name + if file_path.exists(): + file_path.rename(new_path) + logger.info(f"Moved: {file_path.name} → {to_stage}/") + return new_path + + def process_item(self, item: Dict) -> Dict[str, Any]: + """处理单个项目""" + result = { + "item": item, + "success": False, + "stages_completed": [], + "files_created": [], + "error": None + } + + file_path = Path(item['path']) + base_name = file_path.stem + + try: + logger.info(f"Processing: {file_path.name}") + + # 阶段1: 链接解析 + link_info = self._extract_link(file_path) + if not link_info: + result["error"] = "无法解析链接" + return result + + result["stages_completed"].append("link_parsing") + self._save_json(link_info, 'metadata', base_name, '.link.json') + + # 阶段2: 元数据提取 + logger.info(f"Stage 2: Extracting metadata...") + metadata = self._extract_metadata(file_path, 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}") + + # 阶段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') + + result["success"] = len(result["stages_completed"]) >= 2 + logger.info(f"Complete: {len(result['stages_completed'])} stages, {len(result['files_created'])} files") + + except Exception as e: + result["error"] = str(e) logger.error(f"Processing failed: {e}") return result - def _get_stage_from_path(self, path: Path) -> Optional[str]: - """Determine which stage a file is in based on its path""" - for stage in self.INBOX_DIRS: - if stage in str(path): - return stage - return None - - def _move_to_next_stage(self, item_path: Path, completed_stage: str) -> None: - """Move item to the next stage directory""" - stage_order = ['pending', 'metadata', 'content', 'analyzed'] - - try: - current_idx = stage_order.index(completed_stage) - if current_idx < len(stage_order) - 1: - next_stage = stage_order[current_idx + 1] - next_path = self.base_path / next_stage / item_path.name - - # Move file - if item_path.exists(): - item_path.rename(next_path) - logger.info(f"Moved {item_path.name} → {next_stage}/") - except ValueError: - logger.warning(f"Unknown stage: {completed_stage}") - def process_all(self, scope: str = "pending", limit: int = None) -> Dict[str, Any]: - """ - Process all items in specified scope - - Args: - scope: Which directories to process - limit: Maximum items to process - - Returns: - Processing summary - """ - summary = { - 'scope': scope, - 'processed': 0, - 'failed': 0, - 'results': [] - } + """处理所有项目""" + summary = {"processed": 0, "failed": 0, "results": []} inbox = self.scan_inbox(scope=scope) - for dir_name, items in inbox.items(): + for stage, items in inbox.items(): if limit and summary['processed'] >= limit: break - + for item in items: if limit and summary['processed'] >= limit: break - - result = self.process_item(item) - summary['results'].append(result) - if result['success']: - summary['processed'] += 1 + result = self.process_item(item) + summary["results"].append(result) + + if result["success"]: + summary["processed"] += 1 else: - summary['failed'] += 1 + summary["failed"] += 1 return summary - # ==================== Archiving ==================== + # ==================== 状态和汇报 ==================== - def archive_item(self, item: Dict, analysis: Dict = None) -> Dict[str, Any]: - """ - Archive a processed item to the archive directory + def scan_inbox(self, scope: str = "all") -> Dict[str, List[Dict]]: + """扫描收件箱""" + result = {} - Args: - item: Item to archive - analysis: Optional analysis metadata - - Returns: - Archiving result - """ - result = { - 'success': False, - 'original_path': item['path'], - 'archive_path': None, - 'error': None - } - - try: - item_path = Path(item['path']) - - if not item_path.exists(): - result['error'] = f"Item not found: {item['path']}" - return result - - # Create archive subdirectory by date - date_dir = self.archive_path / datetime.now().strftime('%Y-%m-%d') - if not date_dir.exists(): - date_dir.mkdir(parents=True, exist_ok=True) - - # Move item to archive - archive_path = date_dir / item_path.name - item_path.rename(archive_path) - - # Save analysis metadata if provided - if analysis: - meta_path = date_dir / f"{item_path.stem}_meta.json" - with open(meta_path, 'w', encoding='utf-8') as f: - json.dump({ - 'archived_at': datetime.now().isoformat(), - 'analysis': analysis - }, f, ensure_ascii=False, indent=2) - - result['success'] = True - result['archive_path'] = str(archive_path) - logger.info(f"Archived: {item['name']} → {archive_path}") - - except Exception as e: - result['error'] = str(e) - logger.error(f"Archive failed: {e}") - - return result - - # ==================== Subscriptions ==================== - - def check_subscriptions(self) -> Dict[str, Any]: - """ - Check for updates from subscribed sources - - Returns: - Subscription check results - """ - result = { - 'checked_at': datetime.now().isoformat(), - 'sources': {}, - 'new_items': 0, - 'errors': [] - } - - # Load subscription list - subscriptions = self._load_subscriptions() - - for sub in subscriptions: - try: - source_name = sub.get('name', 'unknown') - source_type = sub.get('type', 'rss') - - # Placeholder: Check for new content - logger.info(f"Checking subscription: {source_name}") - - # TODO: Implement actual subscription checking - # - RSS feeds - # - YouTube channel updates - # - Bilibili user updates - # - Custom web hooks - - result['sources'][source_name] = { - 'type': source_type, - 'status': 'checked', - 'new_items': 0 - } - - except Exception as e: - result['errors'].append({ - 'source': sub.get('name', 'unknown'), - 'error': str(e) - }) - - return result - - def _load_subscriptions(self) -> List[Dict]: - """Load subscriptions from file""" - if self.subscriptions_file.exists(): - try: - with open(self.subscriptions_file, 'r', encoding='utf-8') as f: - return json.load(f) - except Exception as e: - logger.warning(f"Failed to load subscriptions: {e}") - - return [] - - # ==================== Reporting ==================== - - def generate_report(self, period: str = "daily") -> Dict[str, Any]: - """ - Generate workflow report - - Args: - period: Report period (daily, weekly, monthly) - - Returns: - Report dictionary - """ - # Calculate date range - now = datetime.now() - - if period == "daily": - start_date = now - timedelta(days=1) - elif period == "weekly": - start_date = now - timedelta(weeks=1) - elif period == "monthly": - start_date = now - timedelta(days=30) + if scope == "all": + stages = self.STAGES + elif scope in self.STAGES: + stages = [scope] else: - start_date = now - timedelta(days=1) + stages = self.STAGES - # Gather statistics - status = self.get_status() + for stage in stages: + dir_path = self.base_path / stage + items = [] + + if dir_path.exists(): + for item_path in dir_path.iterdir(): + if item_path.is_file(): + items.append({ + 'name': item_path.name, + 'path': str(item_path), + 'stage': stage, + 'modified': datetime.fromtimestamp( + item_path.stat().st_mtime + ).isoformat() + }) + + result[stage] = items + logger.info(f"Scanned {stage}: {len(items)} items") + + return result + + def get_status(self) -> Dict[str, Any]: + """获取状态""" inbox = self.scan_inbox(scope="all") - # Build report - report = { - 'generated_at': now.isoformat(), - 'period': period, - 'start_date': start_date.isoformat(), - 'summary': { - 'total_items_inbox': status['total_items'], - 'total_archived': status['archive']['count'], - 'items_by_stage': status['inbox'] - }, - 'processing_stats': { - 'pending': len(inbox.get('pending', [])), - 'metadata': len(inbox.get('metadata', [])), - 'content': len(inbox.get('content', [])), - 'analyzed': len(inbox.get('analyzed', [])) - }, - 'recommendations': self._generate_recommendations(status, inbox) + # 统计各类文件 + 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 + + 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() } - - # Save report - if self.report_file: - self.report_file.parent.mkdir(parents=True, exist_ok=True) - with open(self.report_file, 'w', encoding='utf-8') as f: - json.dump(report, f, ensure_ascii=False, indent=2) - logger.info(f"Report saved: {self.report_file}") - - return report - def _generate_recommendations(self, status: Dict, inbox: Dict) -> List[str]: - """Generate processing recommendations""" - recommendations = [] + 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) - pending_count = status['inbox'].get('pending', {}).get('count', 0) - analyzed_count = status['inbox'].get('analyzed', {}).get('count', 0) + result = {"archived": 0, "errors": []} + cutoff = datetime.now() - timedelta(days=days) - if pending_count > 10: - recommendations.append(f"⚠️ 待处理内容较多 ({pending_count} 项)，建议优先处理") - - if analyzed_count > 20: - recommendations.append(f"📦 已分析内容较多 ({analyzed_count} 项)，建议归档") - - if status['total_items'] == 0: - recommendations.append("📭 收件箱为空，暂无待处理内容") - - return recommendations - - # ==================== CLI Interface ==================== - - def run(self, trigger: str = "manual", scope: str = "all", - limit: int = None, report: bool = False): - """ - Main entry point for running the workflow - - Args: - trigger: Trigger type (manual, scheduled) - scope: Processing scope - limit: Maximum items to process - report: Generate report after processing - """ - logger.info(f"KB Workflow started: trigger={trigger}, scope={scope}") - - # Handle different triggers - if trigger == "status": - result = self.get_status() - elif trigger == "report": - result = self.generate_report() - elif trigger == "subscriptions": - result = self.check_subscriptions() - else: - # Process items - result = self.process_all(scope=scope, limit=limit) + for stage in self.STAGES: + stage_path = self.base_path / stage - # Generate report if requested - if report: - report_result = self.generate_report() - result['report'] = report_result + 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)) + logger.info(f"Archived {result['archived']} items") return result +# ==================== CLI ==================== + def main(): - """CLI entry point""" parser = argparse.ArgumentParser( description='KB Workflow - Knowledge Base Workflow Orchestration', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" -Examples: - # Get workflow status +## 文件存放规则 + +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 - - # Process pending items manually - python kb_workflow.py --trigger manual --scope pending - - # Process up to 5 items - python kb_workflow.py --trigger manual --limit 5 - - # Generate daily report - python kb_workflow.py --trigger scheduled --report - - # Check subscriptions - python kb_workflow.py --trigger scheduled --scope subscriptions + python kb_workflow.py --trigger manual --scope pending --limit 5 """ ) - parser.add_argument( - '--trigger', - type=str, - default='manual', - choices=['manual', 'scheduled', 'webhook', 'status', 'report', 'subscriptions'], - help='Workflow trigger type (default: manual)' - ) - - parser.add_argument( - '--scope', - type=str, - default='all', - choices=['all', 'pending', 'metadata', 'content', 'analyzed', 'archive'], - help='Processing scope (default: all)' - ) - - parser.add_argument( - '--limit', - type=int, - default=None, - help='Maximum items to process' - ) - - parser.add_argument( - '--report', - action='store_true', - help='Generate report after processing' - ) - - parser.add_argument( - '--output', - type=str, - choices=['json', 'text'], - default='json', - help='Output format (default: json)' - ) - - parser.add_argument( - '--base-path', - type=str, - default=None, - help='Base path for knowledge base' - ) - - parser.add_argument( - '--verbose', - action='store_true', - help='Enable verbose logging' - ) + parser.add_argument('--trigger', type=str, default='status', + choices=['status', 'manual', 'archive', 'report'], + help='触发类型') + parser.add_argument('--scope', type=str, default='pending', + choices=['all', 'pending', 'metadata', 'content', 'analyzed'], + help='处理范围') + parser.add_argument('--limit', type=int, default=None, + help='处理数量限制') + parser.add_argument('--days', type=int, default=30, + help='归档天数') args = parser.parse_args() - if args.verbose: - logging.getLogger().setLevel(logging.DEBUG) + workflow = KBWorkflow() - # Initialize workflow - workflow = KBWorkflow(base_path=args.base_path) - - # Run workflow - result = workflow.run( - trigger=args.trigger, - scope=args.scope, - limit=args.limit, - report=args.report - ) - - # Output result - if args.output == 'json': - print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) + if args.trigger == 'status': + result = workflow.get_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: - print(result) + result = {"status": "unknown"} - return 0 if result else 1 + print(json.dumps(result, ensure_ascii=False, indent=2, default=str)) + return 0 if __name__ == '__main__':