From faaf79428ba90ec8e7133ffac4c1157d50b68a12 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Feb 2026 11:50:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(deep-analyzer):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E5=85=83=E6=95=B0=E6=8D=AE=E8=A7=A3=E6=9E=90=E5=92=8CAI?= =?UTF-8?q?=E8=B0=83=E7=94=A8=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 支持 {"metadata": {...}} 格式的元数据 - 从环境变量读取 AI_PROVIDER 和 AI_MODEL - 修复 glm-4.7 的 reasoning_content 处理 - 修复 _format_time 支持 float 类型 - 支持 Whisper 的 segments 和 text 字段 --- skills/kb-deep-analyzer/deep_analyzer.py | 100 ++++++++++++++++------- 1 file changed, 72 insertions(+), 28 deletions(-) diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 10fa482..ab22a9e 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -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', {})