starmen1265/openclaw-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(deep-analyzer): 修复元数据解析和AI调用问题

Browse Source 
- 支持 {"metadata": {...}} 格式的元数据
- 从环境变量读取 AI_PROVIDER 和 AI_MODEL
- 修复 glm-4.7 的 reasoning_content 处理
- 修复 _format_time 支持 float 类型
- 支持 Whisper 的 segments 和 text 字段
This commit is contained in:
starmen1265
2026-02-13 11:50:08 +08:00
parent 1fa90ac21c
commit faaf79428b
1 changed files with 72 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/kb-deep-analyzer/deep_analyzer.py
+72 -28
View File
@@ -54,10 +54,10 @@ class SimpleAIClient:
SHORT_VIDEO_THRESHOLD = 600 # 10分钟
LONG_VIDEO_THRESHOLD = 1800 # 30分钟
def __init__(self, api_key: str = None, provider: str = "mock", model: str = "gpt-4", base_url: str = ""):
def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None):
self.api_key = api_key or os.environ.get("AI_API_KEY", "")
self.provider = provider
self.model = model
self.provider = provider or os.environ.get("AI_PROVIDER", "mock")
self.model = model or os.environ.get("AI_MODEL", "gpt-4")
self.base_url = base_url or os.environ.get("AI_BASE_URL", "")
def generate(self, prompt: str, max_tokens: int = 1000) -> str:
@@ -122,7 +122,17 @@ class SimpleAIClient:
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
return response.json()['choices'][0]['message']['content']
data = response.json()
# 优先使用 content，如果为空则使用 reasoning_content
message = data.get('choices', [{}])[0].get('message', {})
content = message.get('content', '')
reasoning = message.get('reasoning_content', '')
# 如果 content 为空但有 reasoning_content，使用 reasoning
if not content and reasoning:
return reasoning.strip()
return content
def _call_minimax(self, prompt: str, max_tokens: int) -> str:
"""调用 MiniMax API"""
@@ -212,60 +222,66 @@ class DeepAnalyzer:
"""
title = metadata.get('title', '未知')
duration = metadata.get('duration', 0)
subtitle = content.get('subtitle', '')
description = content.get('description', '')
# 支持多种内容格式
# 1. Whisper 格式: {"text": "...", "segments": [...]}
# 2. 字幕格式: {"subtitle": "..."}
# 3. 描述格式: {"description": "..."}
text = content.get('text', '') or content.get('subtitle', '') or content.get('description', '')
segments = content.get('segments', [])
has_content = bool(text and text.strip())
has_segments = bool(segments)
print(f"[DeepAnalyzer] 开始分析: {title}")
print(f"[DeepAnalyzer] 类型: {content_type}, 时长: {duration}秒")
print(f"[DeepAnalyzer] 字幕存在: {bool(subtitle and subtitle.strip())}")
print(f"[DeepAnalyzer] 文本存在: {has_content}, 分段数: {len(segments)}")
has_subtitle = bool(subtitle and subtitle.strip())
if has_subtitle:
if has_content or has_segments:
if duration < self.SHORT_VIDEO_THRESHOLD:
print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)")
result = self._analyze_short_with_subtitle(title, subtitle, duration)
result = self._analyze_short_with_subtitle(title, text, duration, segments)
else:
print(f"[DeepAnalyzer] 使用详细时间轴分析策略 (时长 >= 10分钟)")
result = self._analyze_long_with_subtitle(title, subtitle, duration, content_type)
result = self._analyze_long_with_subtitle(title, text, duration, segments, content_type)
else:
print(f"[DeepAnalyzer] 降级处理（无字幕）")
result = self._analyze_fallback(title, description, metadata)
print(f"[DeepAnalyzer] 降级处理（无内容）")
result = self._analyze_fallback(title, text, metadata)
return result
def _analyze_short_with_subtitle(self, title: str, subtitle: str, duration: int) -> AnalysisResult:
"""有字幕的短视频轻量分析"""
def _analyze_short_with_subtitle(self, title: str, text: str, duration: int, segments: list = None) -> AnalysisResult:
"""有内容的短视频轻量分析"""
result = AnalysisResult()
text_sample = self._get_text_sample(subtitle, max_chars=5000)
text_sample = self._get_text_sample(text, max_chars=5000)
analysis = self._analyze_with_ai_short_text(text_sample, title, duration)
result.summary = analysis.get('summary', title)
result.main_topics = analysis.get('topics', [])
result.key_insights = analysis.get('insights', [])
result.action_items = analysis.get('actions', [])
result.timeline = self._generate_simple_timeline_from_duration(title, duration)
result.timeline = self._generate_timeline_from_segments(title, duration, segments)
return result
def _analyze_long_with_subtitle(self, title: str, subtitle: str, duration: int,
content_type: str) -> AnalysisResult:
"""有字幕的长视频详细分析"""
def _analyze_long_with_subtitle(self, title: str, text: str, duration: int,
segments: list = None, content_type: str = None) -> AnalysisResult:
"""有内容的长视频详细分析"""
result = AnalysisResult()
if duration >= self.SHORT_VIDEO_THRESHOLD and content_type == "video":
# 10分钟以上使用分段分析
result = self._analyze_in_chunks(title, subtitle, duration)
result = self._analyze_in_chunks(title, text, duration)
else:
# 直接分析
text_sample = self._get_text_sample(subtitle, max_chars=5000)
text_sample = self._get_text_sample(text, max_chars=5000)
analysis = self._analyze_with_ai_short_text(text_sample, title, duration)
result.summary = analysis.get('summary', title)
result.main_topics = analysis.get('topics', [])
result.key_insights = analysis.get('insights', [])
result.action_items = analysis.get('actions', [])
result.full_analysis = analysis.get('full_analysis', '')
result.timeline = self._generate_detailed_timeline(title, duration)
result.timeline = self._generate_timeline_from_segments(title, duration, segments)
return result
@@ -525,6 +541,26 @@ class DeepAnalyzer:
text[-part_size:]
)
def _generate_timeline_from_segments(self, title: str, duration: int, segments: list = None) -> List[Dict]:
"""从 Whisper segments 生成时间轴"""
if not segments:
return self._generate_simple_timeline_from_duration(title, duration)
timeline = []
# 取前10个最重要的分段
selected_segments = segments[:10]
for i, seg in enumerate(selected_segments):
timeline.append({
'index': i + 1,
'start_time': self._format_time(seg.get('start', 0)),
'end_time': self._format_time(seg.get('end', 0)),
'title': f"片段 {i+1}",
'summary': seg.get('text', '')[:100] # 截取前100字
})
return timeline
def _generate_simple_timeline_from_duration(self, title: str, duration: int) -> List[Dict]:
"""生成简单时间轴"""
segments = 2 if duration < 300 else 3
@@ -560,8 +596,9 @@ class DeepAnalyzer:
})
return timeline
def _format_time(self, seconds: int) -> str:
def _format_time(self, seconds: float) -> str:
"""格式化秒数为时间字符串"""
seconds = int(seconds) # 确保是整数
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
@@ -754,11 +791,12 @@ def load_json(file_path: str) -> Dict:
def normalize_metadata(metadata: Dict) -> Dict:
"""
标准化元数据格式，支持嵌套和平面两种结构
标准化元数据格式，支持多种嵌套结构
支持以下输入格式:
1. 扁平结构: {"title": "xxx", "author": "yyy", "duration": 619}
2. 嵌套结构: {"data": {"metadata": {"title": "xxx", "author": "yyy", "duration": "10:19"}}}
2. 嵌套结构A: {"data": {"metadata": {"title": "xxx", "author": "yyy", "duration": "10:19"}}}
3. 嵌套结构B: {"metadata": {"title": "xxx", "author": "yyy", "duration": "07:35"}}
输出: 扁平结构，duration 转换为秒数
"""
@@ -778,7 +816,13 @@ def normalize_metadata(metadata: Dict) -> Dict:
pass
return metadata
# 如果是嵌套结构，提取 data.metadata
# 如果有 metadata 字段，提取它（嵌套结构B）
if 'metadata' in metadata:
meta = metadata.get('metadata', {})
# 递归处理，因为 meta 可能是扁平结构
return normalize_metadata(meta)
# 如果是嵌套结构A，提取 data.metadata
data = metadata.get('data', {})
meta = data.get('metadata', {})