OpenClaw Agent
f26d4d8f7f
- 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
229 lines
8.1 KiB
Python
229 lines
8.1 KiB
Python
"""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)
|