feat: implement obsidian-kb skill with long video chunking

- Multi-source content aggregation (Bilibili/YouTube/Zhihu) - Smart inbox management with AI quality scoring - Long video chunking analysis (30min+ auto-segmentation) - Topic-based organization - Git auto-sync to Obsidian - Template system with Jinja2 and fallback rendering - CLI commands: subscribe, inbox, process, skip, status, search - Comprehensive test suite with unit and integration tests - Documentation: README.md and SKILL.md Checkpoint 7-8 completed by Agent E
2026-02-06 18:15:30 +08:00
commit f26d4d8f7f
37 changed files with 6414 additions and 0 deletions
@@ -0,0 +1,228 @@
+"""Obsidian KB - 轻量扫描器
+C4.1: 实现轻量扫描，生成一句话摘要、关键词和质量评分
+"""
+import re
+from typing import List, Tuple
+from src.models import Content, ContentStatus, QualityMetrics
+from src.config import get_config
+
+
+class LightScanner:
+    """轻量内容扫描器 - 快速生成摘要和关键词"""
+    
+    def __init__(self, ai_client=None):
+        self.config = get_config()
+        self.ai_client = ai_client
+        
+    def scan(self, content: Content, content_text: str) -> Content:
+        """
+        对内容进行轻量扫描
+        
+        Args:
+            content: Content对象
+            content_text: 内容文本（字幕或正文）
+            
+        Returns:
+            更新后的Content对象
+        """
+        # 截断文本避免过长
+        text_sample = self._get_text_sample(content_text)
+        
+        # 生成一句话摘要
+        summary = self._generate_summary(text_sample, content.content_type.value)
+        
+        # 提取关键词
+        keywords = self._extract_keywords(text_sample)
+        
+        # 质量评分
+        quality_score = self._calculate_quality_score(text_sample, content)
+        
+        # 更新Content对象
+        content.summary = summary
+        content.key_points = keywords  # 使用key_points存储关键词
+        content.quality_metrics.overall_score = quality_score / 10  # 转换为0-1范围
+        content.update_status(ContentStatus.SCANNED)
+        
+        return content
+    
+    def _get_text_sample(self, text: str, max_chars: int = 3000) -> str:
+        """获取文本样本，优先取开头、中间和结尾"""
+        if len(text) <= max_chars:
+            return text
+        
+        # 分为三部分：开头40%、中间20%、结尾40%
+        start_len = int(max_chars * 0.4)
+        end_len = int(max_chars * 0.4)
+        middle_len = max_chars - start_len - end_len
+        
+        start = text[:start_len]
+        middle_start = len(text) // 2 - middle_len // 2
+        middle = text[middle_start:middle_start + middle_len]
+        end = text[-end_len:]
+        
+        return f"{start}\n\n[...中间部分...]\n\n{middle}\n\n[...省略...]\n\n{end}"
+    
+    def _generate_summary(self, text_sample: str, content_type: str) -> str:
+        """
+        生成一句话摘要
+        如果有AI客户端则调用AI，否则使用简单启发式方法
+        """
+        if self.ai_client:
+            return self._generate_summary_with_ai(text_sample, content_type)
+        else:
+            return self._generate_summary_heuristic(text_sample)
+    
+    def _generate_summary_with_ai(self, text_sample: str, content_type: str) -> str:
+        """使用AI生成摘要"""
+        prompt = f"""请为以下{content_type}内容生成一句话摘要（不超过50字）：
+
+{text_sample[:2000]}
+
+要求：
+1. 一句话概括核心内容
+2. 不超过50字
+3. 突出主要观点和价值
+
+摘要："""
+        
+        try:
+            response = self.ai_client.quick_generate(prompt, max_tokens=100)
+            # 清理并截断
+            summary = response.strip().replace('\n', ' ')
+            if len(summary) > 60:
+                summary = summary[:57] + "..."
+            return summary
+        except Exception as e:
+            print(f"[LightScanner] AI生成摘要失败: {e}")
+            return self._generate_summary_heuristic(text_sample)
+    
+    def _generate_summary_heuristic(self, text: str) -> str:
+        """启发式生成简单摘要"""
+        # 取前100个字符作为简单摘要
+        sentences = re.split(r'[。！？.!?]', text)
+        for sent in sentences:
+            sent = sent.strip()
+            if len(sent) > 20 and len(sent) < 100:
+                return sent[:50] + "..." if len(sent) > 50 else sent
+        
+        # 回退：取前50字
+        text = text.replace('\n', ' ').strip()
+        return text[:47] + "..." if len(text) > 50 else text
+    
+    def _extract_keywords(self, text: str) -> List[str]:
+        """
+        提取5个关键词
+        如果有AI客户端则调用AI，否则使用简单词频分析
+        """
+        if self.ai_client:
+            return self._extract_keywords_with_ai(text)
+        else:
+            return self._extract_keywords_heuristic(text)
+    
+    def _extract_keywords_with_ai(self, text: str) -> List[str]:
+        """使用AI提取关键词"""
+        prompt = f"""请从以下内容中提取5个最重要的关键词，用逗号分隔：
+
+{text[:1500]}
+
+关键词（5个，逗号分隔）："""
+        
+        try:
+            response = self.ai_client.quick_generate(prompt, max_tokens=100)
+            # 解析逗号分隔的关键词
+            keywords = [k.strip() for k in response.split('，') if k.strip()]
+            if len(keywords) < 5:
+                keywords = [k.strip() for k in response.split(',') if k.strip()]
+            
+            # 确保最多5个，每个不超过10字
+            keywords = [k[:10] for k in keywords[:5]]
+            
+            # 如果不足5个，补充通用关键词
+            while len(keywords) < 5:
+                keywords.append(f"关键词{len(keywords)+1}")
+                
+            return keywords[:5]
+        except Exception as e:
+            print(f"[LightScanner] AI提取关键词失败: {e}")
+            return self._extract_keywords_heuristic(text)
+    
+    def _extract_keywords_heuristic(self, text: str) -> List[str]:
+        """启发式提取关键词（基于词频）"""
+        # 简单的中文/英文分词和词频统计
+        import re
+        from collections import Counter
+        
+        # 提取中文字符串（2-6字）
+        chinese_words = re.findall(r'[\u4e00-\u9fff]{2,6}', text)
+        
+        # 提取英文单词
+        english_words = re.findall(r'\b[a-zA-Z]{4,12}\b', text.lower())
+        
+        # 停用词过滤
+        stopwords = {'这个', '那个', '什么', '一个', '我们', '你们', '他们', '可以', '进行'}
+        chinese_words = [w for w in chinese_words if w not in stopwords]
+        
+        # 统计词频
+        all_words = chinese_words + english_words
+        if not all_words:
+            return ["内容", "分析", "知识", "笔记", "学习"]
+        
+        word_freq = Counter(all_words)
+        top_words = [word for word, _ in word_freq.most_common(5)]
+        
+        # 确保有5个
+        defaults = ["内容", "分析", "知识", "笔记", "学习"]
+        while len(top_words) < 5:
+            top_words.append(defaults[len(top_words)])
+        
+        return top_words[:5]
+    
+    def _calculate_quality_score(self, text: str, content: Content) -> float:
+        """
+        计算内容质量评分（0-10）
+        基于多个维度：文本长度、信息密度、结构完整性
+        """
+        score = 5.0  # 基础分
+        
+        # 1. 文本长度评分 (0-3分)
+        text_len = len(text)
+        if text_len > 10000:
+            score += 3.0
+        elif text_len > 5000:
+            score += 2.5
+        elif text_len > 2000:
+            score += 2.0
+        elif text_len > 1000:
+            score += 1.0
+        elif text_len > 500:
+            score += 0.5
+        
+        # 2. 信息密度评分 (0-2分) - 基于句子平均长度
+        sentences = re.split(r'[。！？.!?]', text)
+        sentences = [s for s in sentences if s.strip()]
+        if sentences:
+            avg_sent_len = text_len / len(sentences)
+            if 15 <= avg_sent_len <= 50:
+                score += 2.0
+            elif 10 <= avg_sent_len <= 80:
+                score += 1.0
+        
+        # 3. 结构完整性 (0-2分)
+        # 检查是否有章节标题模式
+        if re.search(r'[#\d]+[.、\s]', text):
+            score += 1.0
+        # 检查是否有时间戳（视频字幕）
+        if content.duration and content.duration > 0:
+            score += 1.0
+        
+        # 4. 多样性评分 (0-2分) - 基于唯一词比例
+        words = re.findall(r'[\u4e00-\u9fff]', text)
+        if words:
+            unique_ratio = len(set(words)) / len(words)
+            if unique_ratio > 0.5:
+                score += 2.0
+            elif unique_ratio > 0.3:
+                score += 1.0
+        
+        # 确保在0-10范围内
+        return round(max(0.0, min(10.0, score)), 1)