diff --git a/SkillBug.md b/SkillBug.md new file mode 100644 index 0000000..5a37834 --- /dev/null +++ b/SkillBug.md @@ -0,0 +1,24 @@ +# SkillBug.md - 技能问题记录 + +## 2026-02-10 - kb-workflow 技能依赖缺失 ✅ 已解决 + +**技能：** kb-workflow +**问题：** 代码中引用了不存在的模块 + +### 解决方案 +1. **删除 kb_handler.py** - 不在 SKILL.md 要求中，是遗留代码 +2. **删除 obsidian-kb** - 不再需要，恢复它是错误操作 +3. **kb-workflow 应该使用 kb-link-parser** - 按 SKILL.md 规定走标准流程 + +### 当前状态 +- ✅ kb_workflow.py 正常 +- ✅ 使用 kb-link-parser 解析短链 + +### kb-workflow 正确用法 +```bash +# 查看状态 +python kb_workflow.py --trigger status + +# 处理待处理内容 +python kb_workflow.py --trigger manual --scope pending --limit 5 +``` diff --git a/skills/kb-workflow/kb_handler.py b/skills/kb-workflow/kb_handler.py deleted file mode 100755 index 971cd4c..0000000 --- a/skills/kb-workflow/kb_handler.py +++ /dev/null @@ -1,551 +0,0 @@ -#!/usr/bin/env python3 -""" -Auto-trigger handler for obsidian-kb skill. - -Detects video/article links in user messages and automatically processes them. -Supports Bilibili, YouTube, and Zhihu. - -## 触发词 -- "保存到知识库" + 链接 → 添加到收件箱并处理 -- "订阅到知识库" + 链接 → 添加订阅 -""" - -import re -import sys -import os -from typing import List, Dict, Optional, Tuple -from urllib.parse import urlparse - -# Add path for imports -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) -sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'src')) - -# Import short link resolver -from shortlink_resolver import ( - ShortLinkResolver, - detect_and_format, - PlaywrightResolver -) - -# 导入主流程 -try: - from main_pipeline import MainPipeline - MAIN_PIPELINE_AVAILABLE = True -except ImportError: - MAIN_PIPELINE_AVAILABLE = False - -# Pattern definitions for auto-detection -PATTERNS = { - 'bilibili_full': re.compile(r'https?://(?:www\.)?bilibili\.com/video/(BV\w+)'), - 'bilibili_short': re.compile(r'https?://b23\.tv/(\w+)'), - 'bilibili_bv': re.compile(r'\b(BV[a-zA-Z0-9]{10})\b'), - 'youtube_full': re.compile(r'https?://(?:www\.)?youtube\.com/watch\?v=([a-zA-Z0-9_-]{11})'), - 'youtube_short': re.compile(r'https?://youtu\.be/([a-zA-Z0-9_-]{11})'), - 'zhihu_article': re.compile(r'https?://zhuanlan\.zhihu\.com/p/(\d+)'), - 'zhihu_answer': re.compile(r'https?://(?:www\.)?zhihu\.com/question/\d+/answer/(\d+)'), -} - -# 触发词模式 -TRIGGER_PATTERNS = { - 'save': re.compile(r'(保存到知识库|加入知识库|添加到知识库)', re.IGNORECASE), - 'subscribe': re.compile(r'(订阅到知识库|订阅知识库|关注到知识库)', re.IGNORECASE), -} - - -def detect_links(text: str) -> List[Dict[str, str]]: - """ - Detect video/article links in text. - - Args: - text: Input text to search - - Returns: - List of detected links with metadata: - [ - { - 'url': 'https://...', - 'platform': 'bilibili', - 'type': 'video', - 'id': 'BVxxx', - 'is_short': False - }, - ... - ] - """ - detected = [] - seen_ids = set() # Track IDs to avoid duplicates - - for pattern_name, pattern in PATTERNS.items(): - matches = pattern.finditer(text) - - for match in matches: - # Determine platform and type - if 'bilibili' in pattern_name: - platform = 'bilibili' - content_type = 'video' - if 'short' in pattern_name: - is_short = True - url = match.group(0) - content_id = None # Need to resolve short URL - else: - is_short = False - if pattern_name == 'bilibili_bv': - url = f"https://www.bilibili.com/video/{match.group(1)}" - else: - url = match.group(0) - content_id = match.group(1) - - elif 'youtube' in pattern_name: - platform = 'youtube' - content_type = 'video' - is_short = 'short' in pattern_name - url = match.group(0) - content_id = match.group(1) - - elif 'zhihu' in pattern_name: - platform = 'zhihu' - content_type = 'article' - is_short = False - url = match.group(0) - content_id = match.group(1) - - else: - continue - - # Skip duplicates by URL or ID - unique_key = url if (not content_id or is_short) else content_id - if unique_key in seen_ids: - continue - seen_ids.add(unique_key) - - detected.append({ - 'url': url, - 'platform': platform, - 'type': content_type, - 'id': content_id, - 'is_short': is_short, - 'pattern': pattern_name - }) - - return detected - - -def format_preview(detected: List[Dict[str, str]]) -> str: - """ - Format a preview message for detected links. - - Args: - detected: List of detected links - - Returns: - Formatted preview message - """ - if not detected: - return "" - - lines = ["🔍 检测到内容链接:\n"] - - for i, item in enumerate(detected, 1): - platform_emoji = { - 'bilibili': '📺', - 'youtube': '▶️', - 'zhihu': '📝' - }.get(item['platform'], '🔗') - - platform_name = { - 'bilibili': 'B站视频', - 'youtube': 'YouTube视频', - 'zhihu': '知乎文章' - }.get(item['platform'], item['platform']) - - lines.append(f"{platform_emoji} [{i}] {platform_name}") - lines.append(f" URL: {item['url']}") - - if item['is_short']: - lines.append(f" ⚠️ 短链接，需要用户提供完整URL或ID") - elif item['id']: - lines.append(f" ID: {item['id']}") - - lines.append("") - - return "\n".join(lines) - - -def format_short_link_help(short_link: Optional[str] = None) -> str: - """ - Format help message for short links. - - Args: - short_link: Optional short link URL to mention - - Returns: - Help message - """ - return ShortLinkResolver.format_short_link_help(short_link) - - -def get_action_prompt(detected: List[Dict[str, str]]) -> str: - """ - Generate prompt asking user what to do. - - Args: - detected: List of detected links - - Returns: - Action prompt - """ - has_short = any(item['is_short'] for item in detected) - - if has_short: - return f"{format_short_link_help()}\n\n其他链接是否立即处理？(y/n)" - - count = len(detected) - if count == 1: - return "是否添加到收件箱并处理？(y/n)" - else: - return f"是否将上述 {count} 个内容添加到收件箱？(y/n)" - - -def detect_triggers(text: str) -> Optional[str]: - """ - 检测用户消息中的触发词。 - - Args: - text: 用户消息 - - Returns: - 'save' | 'subscribe' | None - """ - # 检测"保存到知识库" - if TRIGGER_PATTERNS['save'].search(text): - return 'save' - - # 检测"订阅到知识库" - if TRIGGER_PATTERNS['subscribe'].search(text): - return 'subscribe' - - return None - - -def handle_save_to_kb(detected: List[Dict[str, str]]) -> str: - """ - 处理"保存到知识库"命令：将链接添加到收件箱并处理。 - - Args: - detected: 检测到的链接列表 - - Returns: - 处理结果消息 - """ - if not detected: - return "❌ 未检测到任何链接，请提供要保存的链接" - - results = [] - for item in detected: - if item['is_short']: - results.append(f"⚠️ 短链接无法处理: {item['url']}") - continue - - # 调用保存处理 - success, msg = process_link(item['url'], item['platform'], auto_process=True) - results.append(msg) - - return "\n".join(results) - - -def handle_subscribe_to_kb(detected: List[Dict[str, str]]) -> str: - """ - 处理"订阅到知识库"命令：将链接添加到订阅列表。 - - Args: - detected: 检测到的链接列表 - - Returns: - 处理结果消息 - """ - if not detected: - return "❌ 未检测到任何链接，请提供要订阅的链接" - - results = [] - for item in detected: - if item['platform'] != 'bilibili': - results.append(f"⚠️ 暂不支持订阅 {item['platform']}: {item['url']}") - continue - - if item['is_short']: - results.append(f"⚠️ 短链接无法订阅: {item['url']}") - continue - - try: - # 调用订阅命令 - import subprocess - cmd = [ - sys.executable, - os.path.join(os.path.dirname(__file__), 'kb_workflow.py'), - '--trigger', 'subscriptions', - '--add', 'bilibili', 'video', item['id'], item.get('url', '') - ] - result = subprocess.run(cmd, capture_output=True, text=True) - if result.returncode == 0: - results.append(f"✅ 已订阅: {item['id']}") - else: - results.append(f"❌ 订阅失败: {result.stderr}") - except Exception as e: - results.append(f"❌ 订阅失败: {e}") - - return "\n".join(results) - - -def should_auto_process(text: str) -> bool: - """ - Check if user wants to auto-process based on their message. - - Args: - text: User's message - - Returns: - True if user indicates they want immediate processing - """ - auto_keywords = ['立即处理', '处理', 'process', '--process', '马上'] - return any(keyword in text.lower() for keyword in auto_keywords) - - -def process_link(url: str, platform: str, auto_process: bool = False) -> Tuple[bool, str]: - """ - Process a single link (extract metadata, add to inbox). - - Args: - url: Content URL - platform: Platform name (bilibili/youtube/zhihu) - auto_process: Whether to immediately process after adding - - Returns: - (success, message) tuple - """ - try: - # Import here to avoid issues if dependencies not available - from sources.bilibili import BilibiliAdapter - from models import Content, ContentStatus - from storage import Storage - - # Only support Bilibili for now - if platform != 'bilibili': - return False, f"暂不支持 {platform}，请手动添加" - - # Handle short links - if 'b23.tv' in url: - return False, "短链接无法自动处理，请提供完整URL或BV号" - - # Create adapter - adapter = BilibiliAdapter() - - # Extract metadata - content = adapter.extract_content(url) - if not content: - return False, "无法提取视频信息" - - # Get transcript - transcript = adapter.get_transcript(content) - if transcript: - content.transcript = transcript - - # Save to inbox - content.status = ContentStatus.CONFIRMED - storage = Storage() - storage.save_content(content) - - # Build success message - msg = f"✅ 已添加到收件箱\n" - msg += f" 标题: {content.title}\n" - msg += f" UP主: {content.author}\n" - msg += f" 时长: {content.duration}秒\n" - - if auto_process: - # Process immediately - try: - from src.pipeline import ProcessingPipeline - pipeline = ProcessingPipeline() - content_text = content.transcript or content.summary or "" - result = pipeline.process_full(content, content_text) - storage.save_content(result) - - msg += f"\n🔄 处理完成\n" - msg += f" 📄 笔记: {result.output_path or 'N/A'}\n" - msg += f" 🏷️ 标签: {', '.join(result.tags)}" - except Exception as e: - msg += f"\n⚠️ 处理失败: {e}" - - return True, msg - - except ImportError as e: - return False, f"依赖模块未安装: {e}" - except Exception as e: - return False, f"处理失败: {e}" - - -def process_video_full(source_id: str, enable_git_sync: bool = False) -> Tuple[bool, str]: - """ - 使用主流程完整处理视频（新接口） - - 完整流程: 提取 → 处理 → 输出 → 同步 - - Args: - source_id: 视频ID（支持BV号、URL） - enable_git_sync: 是否启用Git同步 - - Returns: - (success, message) tuple - """ - if not MAIN_PIPELINE_AVAILABLE: - return False, "主流程模块未可用，请检查依赖" - - try: - # 导入配置 - from src.config import get_config - - # 获取配置 - config = get_config() - - # 创建主流程 - pipeline = MainPipeline(config=config, enable_git_sync=enable_git_sync) - - # 处理视频 - result = pipeline.process_video(source_id) - - # 返回结果 - if result['success']: - return True, result['message'] - else: - return False, result.get('error', 'Unknown error') - - except Exception as e: - return False, f"处理失败: {e}" - - -def handle_message(text: str) -> Optional[str]: - """ - Main entry point: handle incoming message and return response. - - ## 触发条件 - - 检测到"保存到知识库"/"订阅到知识库"触发词 + 链接 → 自动处理 - - 仅检测到链接 → 询问用户是否处理 - - Args: - text: User's message text - - Returns: - Response message, or None if no action needed - """ - # 首先检测触发词 - trigger = detect_triggers(text) - - # 检测链接 - detected = detect_links(text) - - # 如果没有链接，也没有触发词，返回 None - if not detected and not trigger: - return None - - # 如果有触发词（save/subscribe），但没有链接 - if trigger and not detected: - if trigger == 'save': - return "请提供要保存到知识库的链接" - elif trigger == 'subscribe': - return "请提供要订阅到知识库的链接" - - # 处理"保存到知识库"触发 - if trigger == 'save': - return handle_save_to_kb(detected) - - # 处理"订阅到知识库"触发 - if trigger == 'subscribe': - return handle_subscribe_to_kb(detected) - - # 以下是原有的仅链接检测逻辑 - # First, check for short links and provide help - has_short, short_msg = detect_and_format(text) - - # If we have short links, show help message first - if has_short: - return short_msg - - # Check if this is a confirmation (y/yes) - text_lower = text.lower().strip() - if text_lower in ('y', 'yes', '是', '好的', 'ok'): - # This would need state management - for now return a message - # about using explicit command - return "请使用 /kb-add 命令添加内容，或重新发送链接" - - # Show preview - preview = format_preview(detected) - prompt = get_action_prompt(detected) - - return f"{preview}\n{prompt}" - - -# CLI interface for testing -def main(): - """CLI for testing the handler.""" - if len(sys.argv) < 2: - print("Usage: python3 handler.py [options]") - print("\nCommands:") - print(" detect 'text' - 检测文本中的内容链接") - print(" process - 使用主流程处理视频") - print(" process --git - 处理视频并同步到Git") - print("\nExamples:") - print(' python3 handler.py detect "https://www.bilibili.com/video/BV1xx411c7mD"') - print(' python3 handler.py process BV1xx411c7mD') - print(' python3 handler.py process BV1xx411c7mD --git') - sys.exit(1) - - command = sys.argv[1].lower() - - if command == 'detect': - # 检测链接 - if len(sys.argv) < 3: - print("Error: Please provide text to detect") - sys.exit(1) - text = ' '.join(sys.argv[2:]) - response = handle_message(text) - if response: - print(response) - else: - print("No content links detected") - - elif command == 'process': - # 使用主流程处理视频 - if len(sys.argv) < 3: - print("Error: Please provide BV number or URL") - sys.exit(1) - - source_id = sys.argv[2] - enable_git = '--git' in sys.argv - - print(f"Processing video: {source_id}") - if enable_git: - print("Git sync: enabled") - else: - print("Git sync: disabled") - print() - - success, message = process_video_full(source_id, enable_git_sync=enable_git) - - if success: - print(message) - sys.exit(0) - else: - print(f"\n❌ Failed: {message}") - sys.exit(1) - - else: - # 兼容旧格式：直接传入文本进行检测 - text = sys.argv[1] - response = handle_message(text) - if response: - print(response) - else: - print("No content links detected") - - -if __name__ == '__main__': - main() diff --git a/skills/kb-workflow/kb_pipeline.py b/skills/kb-workflow/kb_pipeline.py deleted file mode 100755 index c9d9e5a..0000000 --- a/skills/kb-workflow/kb_pipeline.py +++ /dev/null @@ -1,489 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -""" -Obsidian KB - 主流程集成 - -整合提取层、处理层、输出层，实现完整的数据流： -提取(BilibiliExtractor) → 处理(DeepAnalyzer/ContentChunker) → 输出(ObsidianWriter) → 同步(GitSync) -""" - -import os -import sys -import logging -from typing import Dict, Any, Optional -from datetime import datetime -from pathlib import Path - -# 添加项目路径 -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from src.extractors.bilibili_extractor import BilibiliExtractor, ExtractionError -from src.deep_analyzer import DeepAnalyzer -from src.content_chunker import ContentChunker, Chunk -from src.outputs.obsidian_writer import ObsidianWriter -from src.outputs.git_sync import GitSync -from src.models import Content, ContentType, ContentStatus, TimelineEntry -from src.config import get_config - - -# 配置日志 -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', - handlers=[ - logging.StreamHandler(sys.stdout), - logging.FileHandler('obsidian_kb.log', encoding='utf-8') - ] -) -logger = logging.getLogger('ObsidianKB') - - -class MainPipeline: - """主流程集成器""" - - def __init__(self, config=None, enable_git_sync: bool = False): - """ - 初始化主流程 - - Args: - config: 配置对象 - enable_git_sync: 是否启用Git同步 - """ - self.config = config or get_config() - self.enable_git_sync = enable_git_sync - - logger.info("初始化主流程...") - logger.info(f"Obsidian vault 路径: {self.config.vault_path}") - logger.info(f"Git 同步: {'启用' if enable_git_sync else '禁用'}") - - # 初始化各层组件 - self.extractor = BilibiliExtractor(config=self.config) - self.analyzer = DeepAnalyzer(ai_client=None) # 暂时不传入ai_client - self.chunker = ContentChunker() - self.writer = ObsidianWriter(config=self.config) - - # 初始化Git同步器（仅在需要时） - self.git_sync = None - if enable_git_sync and self._is_git_enabled(): - self.git_sync = GitSync(config=self.config) - logger.info("Git 同步器已初始化") - - def _is_git_enabled(self) -> bool: - """检查Git同步是否在配置中启用""" - return self.config.get('git.enabled', False) - - def process_video(self, source_id: str) -> Dict[str, Any]: - """ - 处理单个视频的完整流程 - - Args: - source_id: 视频ID（支持BV号、URL） - - Returns: - 处理结果字典，包含： - - success: bool - 是否成功 - - content: Content对象 - 处理的内容 - - message: str - 处理消息 - - error: str - 错误信息（如果失败） - """ - result = { - 'success': False, - 'content': None, - 'message': '', - 'error': '' - } - - try: - logger.info("=" * 80) - logger.info(f"开始处理视频: {source_id}") - logger.info("=" * 80) - - # Step 1: 提取层 - 提取视频信息 - logger.info("\n[Step 1/5] 提取视频信息...") - extractor_data = self._extract_video(source_id) - - # Step 2: 构建Content对象 - logger.info("\n[Step 2/5] 构建内容对象...") - content = self._build_content(extractor_data) - - # Step 3: 处理层 - 深度分析 - logger.info("\n[Step 3/5] 深度分析...") - analysis_result = self._analyze_content(extractor_data) - - # Step 4: 处理层 - 内容分片 - logger.info("\n[Step 4/5] 内容分片...") - chunks = self._chunk_content(extractor_data) - - # 合并分析结果和分片信息到content - self._merge_analysis_to_content(content, analysis_result, chunks) - - # Step 5: 输出层 - 写入Obsidian - logger.info("\n[Step 5/5] 写入Obsidian...") - output_path = self._write_obsidian_note(content) - - # Step 6: 同步层 - Git同步（可选） - if self.enable_git_sync and self.git_sync: - logger.info("\n[Step 6/6] Git同步...") - self._sync_to_git(content) - - # 更新状态 - content.update_status(ContentStatus.GENERATED) - content.output_path = output_path - - # 构建成功结果 - result['success'] = True - result['content'] = content - result['message'] = self._format_success_message(content, chunks) - - logger.info("\n" + "=" * 80) - logger.info(f"✅ 处理完成: {content.title}") - logger.info(f" 📄 笔记路径: {output_path}") - logger.info(f" 📊 字幕条数: {len(chunks) if chunks else 0}") - logger.info("=" * 80 + "\n") - - except ExtractionError as e: - logger.error(f"❌ 提取失败: {e}") - result['error'] = f"提取失败: {e}" - result['message'] = f"提取视频信息失败: {e}" - - except Exception as e: - logger.error(f"❌ 处理失败: {e}", exc_info=True) - result['error'] = str(e) - result['message'] = f"处理过程中发生错误: {e}" - - return result - - def _extract_video(self, source_id: str) -> Dict[str, Any]: - """ - 提取视频信息 - - Args: - source_id: 视频ID或URL - - Returns: - 提取器返回的数据结构 - - Raises: - ExtractionError: 提取失败 - """ - logger.info(f" 正在提取: {source_id}") - - try: - extractor_data = self.extractor.extract(source_id) - logger.info(f" ✅ 提取成功: {extractor_data.get('metadata', {}).get('title', '未知标题')}") - logger.info(f" 作者: {extractor_data.get('metadata', {}).get('author', '未知')}") - logger.info(f" 时长: {extractor_data.get('metadata', {}).get('duration', '0')}") - - # 检查字幕 - subtitle = extractor_data.get('subtitle', '') - if subtitle: - subtitle_lines = subtitle.split('\n') - logger.info(f" 字幕: {len(subtitle_lines)} 条") - else: - logger.warning(f" ⚠️ 无字幕") - - return extractor_data - - except ExtractionError as e: - logger.error(f" ❌ 提取失败: {e}") - raise - except Exception as e: - logger.error(f" ❌ 提取异常: {e}", exc_info=True) - raise ExtractionError(f"提取过程异常: {e}", source_id=source_id) - - def _build_content(self, extractor_data: Dict[str, Any]) -> Content: - """ - 从提取器数据构建Content对象 - - Args: - extractor_data: 提取器返回的数据 - - Returns: - Content对象 - """ - metadata = extractor_data.get('metadata', {}) - - content = Content( - source='bilibili', - source_id=metadata.get('bvid', ''), - url=metadata.get('url', ''), - title=metadata.get('title', ''), - author=metadata.get('author', ''), - description=extractor_data.get('description', ''), - duration=self._parse_duration(metadata.get('duration', '0')), - content_type=ContentType.VIDEO, - transcript=extractor_data.get('subtitle', ''), - status=ContentStatus.SCANNED - ) - - logger.info(f" ✅ Content对象已创建: {content.id}") - return content - - def _parse_duration(self, duration_str: str) -> Optional[int]: - """ - 解析时长字符串为秒数 - - Args: - duration_str: 时长字符串（如 "10:30"） - - Returns: - 秒数 - """ - try: - if ':' in duration_str: - parts = duration_str.split(':') - if len(parts) == 2: - # MM:SS - return int(parts[0]) * 60 + int(parts[1]) - elif len(parts) == 3: - # HH:MM:SS - return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2]) - except: - pass - return None - - def _analyze_content(self, extractor_data: Dict[str, Any]) -> Dict[str, Any]: - """ - 深度分析内容 - - Args: - extractor_data: 提取器返回的数据 - - Returns: - 分析结果字典 - """ - try: - logger.info(" 正在进行深度分析...") - - # 使用DeepAnalyzer进行分析 - # DeepAnalyzer接收extractor_data作为输入 - analysis_result = self.analyzer.analyze(extractor_data) - - # 将AnalysisResult转换为字典以便合并 - result_dict = { - 'main_topics': analysis_result.main_topics, - 'key_insights': analysis_result.key_insights, - 'action_items': analysis_result.action_items, - 'summary': analysis_result.summary, - 'timeline': analysis_result.timeline, - 'recommended_tags': analysis_result.recommended_tags, - 'full_analysis': analysis_result.full_analysis - } - - logger.info(f" ✅ 分析完成: {len(result_dict['main_topics'])} 个主题") - - return result_dict - - except Exception as e: - logger.error(f" ❌ 分析失败: {e}", exc_info=True) - # 返回默认结果，不中断流程 - return { - 'main_topics': ['未分类'], - 'key_insights': [], - 'action_items': [], - 'summary': '分析失败，无法生成摘要', - 'timeline': [], - 'recommended_tags': ['未分类'], - 'full_analysis': '' - } - - def _chunk_content(self, extractor_data: Dict[str, Any]) -> list: - """ - 内容分片 - - Args: - extractor_data: 提取器返回的数据 - - Returns: - Chunk对象列表 - """ - try: - logger.info(" 正在进行内容分片...") - - # 使用ContentChunker进行分片 - chunks = self.chunker.chunk_from_extractor(extractor_data) - - logger.info(f" ✅ 分片完成: {len(chunks)} 段") - - return chunks - - except Exception as e: - logger.error(f" ❌ 分片失败: {e}", exc_info=True) - # 返回空列表，不中断流程 - return [] - - def _merge_analysis_to_content(self, content: Content, analysis_result: Dict[str, Any], chunks: list): - """ - 将分析结果和分片信息合并到Content对象 - - Args: - content: Content对象 - analysis_result: 分析结果字典 - chunks: Chunk对象列表 - """ - # 合并基本信息 - content.summary = analysis_result.get('summary', '') - content.key_points = analysis_result.get('main_topics', []) - content.tags = analysis_result.get('recommended_tags', []) - - # 合并时间轴 - timeline_data = analysis_result.get('timeline', []) - if timeline_data: - content.timeline = [ - TimelineEntry( - timestamp=item.get('start_time', ''), - title=item.get('title', ''), - description=item.get('summary', '') - ) - for item in timeline_data - ] - else: - # 从chunks生成时间轴 - if chunks: - content.timeline = [ - TimelineEntry( - timestamp=chunk.time_range, - title=f"片段 {chunk.index + 1}", - description=f"时长 {int(chunk.duration)}秒" - ) - for chunk in chunks - ] - - logger.info(f" ✅ 数据合并完成: 时间轴 {len(content.timeline)} 条") - - def _write_obsidian_note(self, content: Content) -> Optional[str]: - """ - 写入Obsidian笔记 - - Args: - content: Content对象 - - Returns: - 文件路径，失败返回None - """ - try: - logger.info(f" 正在写入笔记...") - - # 确保输出目录存在 - self.config.vault_path.mkdir(parents=True, exist_ok=True) - - # 调用ObsidianWriter写入 - output_path = self.writer.write(content, self.config) - - if output_path: - logger.info(f" ✅ 笔记已写入: {output_path}") - else: - logger.warning(f" ⚠️ 笔记写入失败") - - return output_path - - except Exception as e: - logger.error(f" ❌ 写入失败: {e}", exc_info=True) - return None - - def _sync_to_git(self, content: Content): - """ - 同步到Git仓库 - - Args: - content: Content对象 - """ - if not self.git_sync: - logger.warning(" ⚠️ Git同步器未初始化，跳过同步") - return - - try: - logger.info(" 正在同步到Git...") - - success = self.git_sync.sync(self.config) - - if success: - logger.info(" ✅ Git同步成功") - else: - logger.warning(" ⚠️ Git同步失败") - - except Exception as e: - logger.error(f" ❌ Git同步异常: {e}", exc_info=True) - # Git同步失败不视为整体失败 - - def _format_success_message(self, content: Content, chunks: list) -> str: - """ - 格式化成功消息 - - Args: - content: Content对象 - chunks: Chunk列表 - - Returns: - 格式化的消息字符串 - """ - lines = [ - "✅ 处理完成！", - "", - f"📺 标题: {content.title}", - f"👤 作者: {content.author}", - f"📄 笔记: {content.output_path or 'N/A'}", - ] - - if content.duration: - lines.append(f"⏱️ 时长: {self._format_duration(content.duration)}") - - if chunks: - lines.append(f"📊 分片: {len(chunks)} 段") - - if content.tags: - lines.append(f"🏷️ 标签: {', '.join(content.tags[:5])}") - - lines.append("") - - return "\n".join(lines) - - def _format_duration(self, seconds: int) -> str: - """格式化时长""" - if not seconds: - return "N/A" - - hours = seconds // 3600 - minutes = (seconds % 3600) // 60 - secs = seconds % 60 - - if hours > 0: - return f"{hours}小时{minutes}分钟" - elif minutes > 0: - return f"{minutes}分钟{secs}秒" - else: - return f"{secs}秒" - - -def main(): - """命令行入口""" - import argparse - - parser = argparse.ArgumentParser(description='Obsidian KB - 主流程集成') - parser.add_argument('source_id', help='视频ID（BV号或URL）') - parser.add_argument('--git', action='store_true', help='启用Git同步') - parser.add_argument('--config', help='配置文件路径') - - args = parser.parse_args() - - # 加载配置 - config = get_config(args.config) if args.config else get_config() - - # 创建主流程 - pipeline = MainPipeline(config=config, enable_git_sync=args.git) - - # 处理视频 - result = pipeline.process_video(args.source_id) - - # 输出结果 - print(result['message']) - - if not result['success']: - print(f"\n❌ 失败: {result['error']}") - sys.exit(1) - else: - sys.exit(0) - - -if __name__ == '__main__': - main() diff --git a/skills/kb-workflow/src/extractors/bilibili_extractor.py b/skills/kb-workflow/src/extractors/bilibili_extractor.py index 1c822e7..c7c55a9 100644 --- a/skills/kb-workflow/src/extractors/bilibili_extractor.py +++ b/skills/kb-workflow/src/extractors/bilibili_extractor.py @@ -22,11 +22,13 @@ from bs4 import BeautifulSoup from .base import Extractor, ExtractionError -# 导入 shortlink_resolver +# 导入 kb-link-parser 用于短链解析 import sys import os -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) -from shortlink_resolver import ShortLinkResolver +# 添加 kb-link-parser 到导入路径 +KB_LINK_PARSER_PATH = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) +sys.path.insert(0, os.path.join(KB_LINK_PARSER_PATH, 'kb-link-parser')) +from link_parser_shortlink import ShortLinkResolver from .hard_subtitle_extractor import HardSubtitleExtractor # 尝试导入 stealth 支持