diff --git a/kb-deep-analyzer/SKILL.md b/kb-deep-analyzer/SKILL.md deleted file mode 100644 index 0977cab..0000000 --- a/kb-deep-analyzer/SKILL.md +++ /dev/null @@ -1,102 +0,0 @@ -# KB Deep Analyzer Skill - -**kb-deep-analyzer** - AI 深度内容分析工具。 - -## Features - -- 多类型分析：视频（video）、文章（article） -- 模板化输出 -- 质量评分 -- 自动标签 -- **多格式元数据支持**: 支持扁平结构 `{"title":"xxx"}` 和嵌套结构 `{"data":{"metadata":{...}}}` - -## 内容类型 - -| 类型 | 参数 | 输出格式 | -|------|------|----------| -| 视频 | `--type video` | 摘要 + 主题 + 见解 + 时间轴 | -| 文章 | `--type article` | 300字摘要 + 主题 + 原文 | - -## Commands - -```bash -# 分析视频（默认） -AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \ - python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type video - -# 分析文章（300字摘要 + 原文） -AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \ - python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type article - -# 指定 AI 提供商 -python3 -u deep_analyzer.py -m meta.json -c content.json -o out.md --type video --provider zhipu --model glm-5 -``` - -## 环境变量配置 - -**必须设置以下环境变量**： - -```bash -export AI_API_KEY="your-api-key" -export AI_PROVIDER="zhipu" # 或 openai, minimax -export AI_MODEL="glm-5" # 模型名称 -export AI_BASE_URL="https://..." # 可选，自定义 API 地址 -``` - -### 重要提示 - -**必须使用 `-u` 参数运行 Python**，确保输出实时显示在终端： - -```bash -python3 -u deep_analyzer.py ... -``` - -如果不使用 `-u` 参数，Python 的 stdout 缓冲会导致输出延迟或看不到进度。 - -## 执行方式选择 - -### 方式1: 并发执行（推荐，默认） -- 使用 `batch_size=3`（默认） -- 6个分片约 100 秒完成 -- 适合快速分析 - -```bash -python3 -u deep_analyzer.py --metadata meta.json --content content.json -o out.md -``` - -### 方式2: 顺序执行 -- 使用 `batch_size=1` -- 6个分片约 735 秒完成 -- 适合调试和问题排查 - -```bash -# 在代码中设置 batch_size=1 -# analyzer.analyze(metadata, content, "video", batch_size=1) -``` - -## Metadata Format - -支持两种元数据格式： - -```json -// 格式1: 扁平结构 -{ - "title": "视频标题", - "author": "作者名", - "duration": 619, - "bvid": "BVxxx" -} - -// 格式2: 嵌套结构（kb-bilibili-metadata 输出） -{ - "data": { - "metadata": { - "title": "视频标题", - "author": "作者名", - "duration": "10:19" - } - } -} -``` - -会自动转换为扁平格式进行分析。 diff --git a/kb-deep-analyzer/__pycache__/deep_analyzer.cpython-312.pyc b/kb-deep-analyzer/__pycache__/deep_analyzer.cpython-312.pyc deleted file mode 100644 index 08d1b7a..0000000 Binary files a/kb-deep-analyzer/__pycache__/deep_analyzer.cpython-312.pyc and /dev/null differ diff --git a/kb-deep-analyzer/ai_client.py b/kb-deep-analyzer/ai_client.py deleted file mode 100644 index cee3c73..0000000 --- a/kb-deep-analyzer/ai_client.py +++ /dev/null @@ -1,514 +0,0 @@ -"""Obsidian KB - AI客户端 -支持多个 AI 提供商和 fallback 机制 -""" -import os -import json -import requests -from typing import Dict, Any, List, Optional, Tuple -from src.models import ContentType -from src.config import get_config -from src.content_chunker import ContentChunker, get_chunking_strategy, chunk_content - - -class AIClient: - """ - AI客户端 - 支持分段分析长视频内容和 fallback 机制 - """ - - def __init__(self): - self.config = get_config() - self.chunker = ContentChunker() - # 获取 AI 配置 - self._load_ai_config() - - def _load_ai_config(self) -> None: - """加载 AI 配置（主 provider + fallbacks）""" - self.ai_configs = [] - - # 加载主配置 - main_provider = self.config.get('ai.provider', 'openai') - main_config = { - 'provider': main_provider, - 'api_key': self.config.get('ai.api_key'), - 'model': self.config.get('ai.model', 'gpt-4'), - 'base_url': self.config.get('ai.base_url'), - 'is_primary': True - } - self.ai_configs.append(main_config) - - # 加载 fallback 配置 - fallbacks = self.config.get('ai.fallbacks', []) - for fb in fallbacks: - self.ai_configs.append({ - 'provider': fb.get('provider'), - 'api_key': fb.get('api_key'), - 'model': fb.get('model'), - 'base_url': fb.get('base_url'), - 'is_primary': False - }) - - def _call_api_with_fallback(self, prompt: str, max_tokens: int = 1000, response_format: str = 'text') -> Tuple[bool, str]: - """ - 使用 fallback 机制调用 API - - Args: - prompt: 提示词 - max_tokens: 最大 token 数 - response_format: 响应格式 ('text' 或 'json') - - Returns: - (success, response) 元组 - """ - last_error = None - - for i, ai_config in enumerate(self.ai_configs): - try: - if not ai_config.get('api_key'): - print(f"[AIClient] 跳过 {ai_config.get('provider')} - API Key 为空") - continue - - provider_type = ai_config.get('provider') - print(f"[AIClient] 尝试使用 {provider_type} (优先级 {i+1}/{len(self.ai_configs)})...") - - response = self._call_single_provider(ai_config, prompt, max_tokens, response_format) - return True, response - - except Exception as e: - last_error = e - print(f"[AIClient] {provider_type} 调用失败: {e}") - continue - - # 所有 provider 都失败了 - error_msg = f"所有 AI 提供商均失败，最后错误: {last_error}" if last_error else "没有可用的 AI 配置" - print(f"[AIClient] {error_msg}") - return False, self._mock_generate(prompt, max_tokens) # 返回 mock 响应作为兜底 - - def _call_single_provider(self, ai_config: Dict, prompt: str, max_tokens: int, response_format: str) -> str: - """ - 调用单个 AI 提供商 - - Args: - ai_config: AI 配置 - prompt: 提示词 - max_tokens: 最大 token 数 - response_format: 响应格式 - - Returns: - API 响应文本 - """ - provider = ai_config.get('provider') - api_key = ai_config.get('api_key') - model = ai_config.get('model') - base_url = ai_config.get('base_url') - - if provider == 'minimax': - return self._call_minimax(api_key, model, base_url, prompt, max_tokens, response_format) - elif provider == 'zhipu': - return self._call_zhipu(api_key, model, base_url, prompt, max_tokens, response_format) - elif provider == 'alibaba': - return self._call_alibaba(api_key, model, base_url, prompt, max_tokens, response_format) - elif provider == 'openai': - return self._call_openai(api_key, model, base_url, prompt, max_tokens, response_format) - else: - raise ValueError(f"不支持的 provider: {provider}") - - def _call_minimax(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str: - """调用 MiniMax API (Anthropic Messages 兼容接口)""" - url = base_url or "https://api.minimaxi.com/anthropic/v1/messages" - - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - } - - payload = { - "model": model, - "max_tokens": max_tokens, - "messages": [ - {"role": "user", "content": prompt} - ] - } - - if response_format == 'json': - payload['messages'][0]['content'] = prompt + "\n\n请直接返回 JSON 格式，不要有其他文字。" - - response = requests.post(url, headers=headers, json=payload, timeout=30) - response.raise_for_status() - data = response.json() - - # MiniMax 返回的是 Anthropic 格式，content 是一个数组 - # 需要找到 type='text' 的元素 - for item in data.get('content', []): - if item.get('type') == 'text': - return item.get('text', '') - - # 如果没有找到 text 类型，返回第一个元素的值（兼容旧格式） - return data.get('content', [{}])[0].get('text', '') - - def _call_zhipu(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str: - """调用智谱 AI API (OpenAI 兼容接口)""" - # 确保 base_url 包含 /chat/completions - if base_url and not base_url.endswith('/chat/completions'): - url = f"{base_url}/chat/completions" - else: - url = base_url or "https://open.bigmodel.cn/api/paas/v4/chat/completions" - - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - } - - payload = { - "model": model, - "max_tokens": max_tokens, - "messages": [ - {"role": "user", "content": prompt} - ] - } - - response = requests.post(url, headers=headers, json=payload, timeout=30) - response.raise_for_status() - data = response.json() - return data['choices'][0]['message']['content'] - - def _call_alibaba(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str: - """调用阿里云通义千问 API (OpenAI 兼容接口)""" - url = f"{base_url}/chat/completions" - - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - } - - payload = { - "model": model, - "max_tokens": max_tokens, - "messages": [ - {"role": "user", "content": prompt} - ] - } - - response = requests.post(url, headers=headers, json=payload, timeout=30) - response.raise_for_status() - data = response.json() - return data['choices'][0]['message']['content'] - - def _call_openai(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str: - """调用 OpenAI API""" - url = base_url or "https://api.openai.com/v1/chat/completions" - - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - } - - payload = { - "model": model, - "max_tokens": max_tokens, - "messages": [ - {"role": "user", "content": prompt} - ] - } - - response = requests.post(url, headers=headers, json=payload, timeout=30) - response.raise_for_status() - data = response.json() - return data['choices'][0]['message']['content'] - - def quick_generate(self, prompt: str, max_tokens: int = 100) -> str: - """快速生成（用于轻量扫描）""" - success, response = self._call_api_with_fallback(prompt, max_tokens, 'text') - return response - - def generate_json(self, prompt: str) -> Dict[str, Any]: - """生成JSON格式响应""" - success, response = self._call_api_with_fallback(prompt, 1000, 'json') - try: - return json.loads(response) - except: - return self._parse_json_from_text(response) - - def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]: - """ - C5.3: 深度分析入口，根据内容时长决定是否分段分析 - - Args: - content_text: 内容文本 - content_type: 内容类型 - duration: 视频时长（秒） - - Returns: - 分析结果字典 - """ - duration = duration or 0 - - print(f"[AIClient] 深度分析: type={content_type}, duration={duration}s") - - # 长视频（>=30分钟）且是视频类型，使用分段分析 - if duration >= 1800 and content_type == "video": - print(f"[AIClient] 使用分段分析策略") - return self._analyze_in_chunks(content_text, duration) - - # 短视频直接分析 - print(f"[AIClient] 使用直接分析策略") - return self._analyze_direct(content_text) - - def _analyze_in_chunks(self, content_text: str, duration: int) -> Dict[str, Any]: - """ - C5.3: 分段分析长视频 - - Args: - content_text: SRT字幕文本 - duration: 视频时长（秒） - - Returns: - 合并后的分析结果 - """ - # 1. 获取分段策略 - strategy = get_chunking_strategy(duration) - print(f"[AIClient] 分段策略: {strategy['description']}") - - # 2. 执行分片 - chunks = chunk_content(content_text, duration) - print(f"[AIClient] 内容已分为 {len(chunks)} 段") - - # 3. 逐段分析 - chunk_summaries = [] - for i, chunk in enumerate(chunks): - print(f"[AIClient] 分析第 {i+1}/{len(chunks)} 段...") - summary = self._analyze_chunk(chunk.text, chunk.index) - - # 处理时间格式（可能是float秒数或字符串） - start_time = chunk.start_time - end_time = chunk.end_time - if isinstance(start_time, (int, float)): - start_time = self._format_time(start_time) - if isinstance(end_time, (int, float)): - end_time = self._format_time(end_time) - - chunk_summaries.append({ - 'index': chunk.index, - 'start_time': start_time, - 'end_time': end_time, - 'summary': summary - }) - - # 4. 合并所有段摘要 - print(f"[AIClient] 合并 {len(chunk_summaries)} 段分析结果") - final_result = self._merge_summaries(chunk_summaries, "video") - - # 5. 构建完整时间轴 - final_result['timeline'] = self._build_timeline_from_chunks(chunk_summaries) - - return final_result - - def _analyze_chunk(self, chunk_text: str, chunk_index: int) -> str: - """ - C5.3: 分析单个内容分片 - - Args: - chunk_text: 分片文本 - chunk_index: 分片索引 - - Returns: - 分片摘要（限制500字以内） - """ - # 截断文本避免过长 - text_sample = chunk_text[:3000] if len(chunk_text) > 3000 else chunk_text - - prompt = f"""请对以下视频片段进行简要分析（限制在300字以内）： - -片段内容： -{text_sample} - -请提供： -1. 这段的核心主题（50字以内） -2. 关键要点（3-5条） -3. 重要细节或引用 - -请用中文回答，控制在300字以内。""" - - try: - summary = self.quick_generate(prompt, max_tokens=500) - # 确保不超过500字 - if len(summary) > 500: - summary = summary[:497] + "..." - return summary - except Exception as e: - print(f"[AIClient] 分片 {chunk_index} 分析失败: {e}") - return f"第{chunk_index+1}段分析失败" - - def _merge_summaries(self, chunk_summaries: List[Dict], content_type: str) -> Dict[str, Any]: - """ - C5.3: 合并所有段摘要，生成完整分析报告 - - Args: - chunk_summaries: 各段摘要列表 - content_type: 内容类型 - - Returns: - 完整分析结果 - """ - # 构建合并提示 - summaries_text = "\n\n".join([ - f"=== 第{s['index']+1}段 ({s['start_time']}-{s['end_time']}) ===\n{s['summary']}" - for s in chunk_summaries - ]) - - prompt = f"""基于以下各段分析，生成完整的视频分析报告： - -{summaries_text} - -请提供以下格式的JSON： -{{ - "summary": "整体摘要（200字以内）", - "topics": ["主题1", "主题2", "主题3", "主题4"], - "insights": ["深入见解1", "深入见解2", "深入见解3"], - "actions": ["建议行动1", "建议行动2"], - "full_analysis": "详细分析（500字以内）" -}}""" - - try: - result = self.generate_json(prompt) - - # 确保必要字段存在 - if 'topics' not in result: - result['topics'] = ["主题A", "主题B", "主题C"] - if 'insights' not in result: - result['insights'] = ["重要见解1", "重要见解2"] - if 'actions' not in result: - result['actions'] = [] - if 'summary' not in result: - result['summary'] = "长视频分析完成" - if 'full_analysis' not in result: - result['full_analysis'] = summaries_text[:500] - - return result - - except Exception as e: - print(f"[AIClient] 合并摘要失败: {e}") - # 返回基础合并结果 - return { - 'summary': f'视频共分为{len(chunk_summaries)}个部分进行分析', - 'topics': [f"主题{i+1}" for i in range(min(4, len(chunk_summaries)))], - 'insights': ['从分段分析中综合得出'], - 'actions': [], - 'full_analysis': summaries_text[:800] - } - - def _analyze_direct(self, content_text: str) -> Dict[str, Any]: - """直接分析（不分段）""" - # 截断文本 - text_sample = content_text[:5000] if len(content_text) > 5000 else content_text - - prompt = f"""请分析以下内容，提供JSON格式结果： - -{text_sample} - -返回格式： -{{ - "summary": "整体摘要（150字以内）", - "topics": ["主题1", "主题2", "主题3"], - "insights": ["见解1", "见解2"], - "actions": ["行动建议1"], - "timeline": [ - {{"start_time": "00:00", "end_time": "05:00", "title": "第一部分", "summary": "..."}} - ] -}}""" - - try: - return self.generate_json(prompt) - except Exception as e: - print(f"[AIClient] 直接分析失败: {e}") - return { - 'summary': '分析失败', - 'topics': ['无法解析'], - 'insights': ['请检查内容格式'], - 'actions': [], - 'timeline': [] - } - - def _format_time(self, seconds: float) -> str: - """格式化秒数为时间字符串""" - hours = int(seconds // 3600) - minutes = int((seconds % 3600) // 60) - secs = int(seconds % 60) - if hours > 0: - return f"{hours:02d}:{minutes:02d}:{secs:02d}" - return f"{minutes:02d}:{secs:02d}" - - def _build_timeline_from_chunks(self, chunk_summaries: List[Dict]) -> List[Dict[str, Any]]: - """从分片摘要构建时间轴""" - timeline = [] - for summary in chunk_summaries: - timeline.append({ - 'index': summary['index'], - 'start_time': summary['start_time'], - 'end_time': summary['end_time'], - 'title': f"第{summary['index']+1}部分", - 'summary': summary['summary'][:200] + "..." if len(summary['summary']) > 200 else summary['summary'] - }) - return timeline - - def _mock_generate(self, prompt: str, max_tokens: int) -> str: - """模拟AI生成（实际实现应调用真实API）""" - # 根据prompt内容返回模拟响应 - prompt_lower = prompt.lower() - - if 'json' in prompt_lower or 'json格式' in prompt: - if '主题' in prompt or 'topic' in prompt_lower: - return json.dumps({ - 'summary': '这是一个关于人工智能和技术发展的内容，涵盖了多个重要主题。', - 'topics': ['人工智能', '机器学习', '未来发展', '技术趋势'], - 'insights': [ - 'AI正在改变我们的工作方式', - '机器学习应用越来越广泛', - '未来技术发展趋势值得关注' - ], - 'actions': ['深入研究相关技术', '关注行业动态'], - 'full_analysis': '详细分析内容...' - }, ensure_ascii=False) - return json.dumps({'result': '模拟JSON响应'}) - - if '摘要' in prompt or 'summary' in prompt_lower: - return '这是一段关于人工智能发展的重要讨论，涵盖了技术趋势和未来应用。' - - if '关键词' in prompt or 'keyword' in prompt_lower: - return '人工智能, 机器学习, 深度学习, 神经网络, 自然语言处理' - - return '模拟AI响应内容' - - def _parse_json_from_text(self, text: str) -> Dict[str, Any]: - """从文本中解析JSON""" - # 尝试提取JSON块 - import re - json_match = re.search(r'\{[\s\S]*\}', text) - if json_match: - try: - return json.loads(json_match.group()) - except: - pass - return {'raw_response': text} - - -class MultiAIClient: - """ - 多AI客户端 - 支持多种AI提供商的客户端 - """ - - def __init__(self): - self.config = get_config() - self.primary_client = AIClient() - self.clients = { - 'openai': self.primary_client, - 'local': AIClient(), # 可以配置为本地模型 - } - - def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]: - """使用主客户端进行深度分析""" - return self.primary_client._deep_analyze(content_text, content_type, duration) - - def quick_generate(self, prompt: str, max_tokens: int = 100, provider: str = None) -> str: - """快速生成""" - client = self.clients.get(provider, self.primary_client) - return client.quick_generate(prompt, max_tokens) diff --git a/kb-deep-analyzer/deep_analyzer.py b/kb-deep-analyzer/deep_analyzer.py deleted file mode 100644 index 2f947ce..0000000 --- a/kb-deep-analyzer/deep_analyzer.py +++ /dev/null @@ -1,650 +0,0 @@ -#!/usr/bin/env python3 -"""Deep Analyzer - AI 深度内容分析工具 - -一次性调用 AI 生成完整深度分析报告。 - -Usage: - python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md [--type video] - python deep_analyzer.py --help -""" - -import argparse -import json -import os -import sys -import re -from datetime import datetime -from pathlib import Path -from typing import Dict, Any, List -from dataclasses import dataclass, field - - -# ============== 数据模型 ============== - -@dataclass -class AnalysisResult: - """分析结果""" - summary: str = "" - main_topics: List[str] = field(default_factory=list) - key_insights: List[str] = field(default_factory=list) - recommended_tags: List[str] = field(default_factory=list) - full_analysis: str = "" - original_text: str = "" # 文章类型的原文内容 - - -# ============== AI 客户端 ============== - -class SimpleAIClient: - """AI 客户端""" - - 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 or os.environ.get("AI_PROVIDER", "") - self.model = model or os.environ.get("AI_MODEL", "glm-5") - self.base_url = base_url or os.environ.get("AI_BASE_URL", "") - self.max_retries = 3 - - def generate(self, prompt: str, max_tokens: int = 16000) -> str: - """生成文本""" - api_key = os.environ.get("AI_API_KEY", self.api_key) - provider = os.environ.get("AI_PROVIDER", self.provider) - model = os.environ.get("AI_MODEL", self.model) - base_url = os.environ.get("AI_BASE_URL", self.base_url) - - if not api_key: - raise ValueError("AI_API_KEY 未设置") - - for attempt in range(self.max_retries): - try: - import time - start_time = time.time() - print(f"[AI Client] 开始调用 AI...") - - if provider == "openai": - result = self._call_openai(prompt, max_tokens, api_key, model, base_url) - elif provider == "zhipu": - result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url) - elif provider == "minimax": - result = self._call_minimax(prompt, max_tokens, api_key, model, base_url) - else: - raise ValueError(f"不支持的 AI 提供商: {provider}") - - elapsed = time.time() - start_time - print(f"[AI Client] AI 调用完成，耗时: {elapsed:.2f}秒") - return result - except Exception as e: - if attempt < self.max_retries - 1: - print(f"[AI Client] 第 {attempt + 1} 次调用失败，{self.max_retries - attempt - 1} 次重试剩余: {e}") - continue - else: - raise - - def _call_openai(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: - import requests - url = f"{base_url}/chat/completions" if base_url else "https://api.openai.com/v1/chat/completions" - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]} - response = requests.post(url, headers=headers, json=payload, timeout=600) - response.raise_for_status() - return response.json()['choices'][0]['message']['content'] - - def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: - import requests - base_url = base_url or "https://open.bigmodel.cn/api/coding/paas/v4" - url = f"{base_url}/chat/completions" - headers = {"Authorization": f"Bearer {api_key}"} - payload = {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens} - response = requests.post(url, headers=headers, json=payload, timeout=600) - response.raise_for_status() - data = response.json() - message = data.get('choices', [{}])[0].get('message', {}) - content = message.get('content', '') - reasoning = message.get('reasoning_content', '') - if not content and reasoning: - sentences = reasoning.strip().split('。') - useful = [s.strip() for s in sentences if s.strip() and len(s.strip()) > 10] - if useful: - return '。'.join(useful[-3:]) + '。' - return reasoning.strip()[:500] - return content - - def _call_minimax(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: - import requests - url = base_url or "https://api.minimaxi.com/anthropic/v1/messages" - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]} - response = requests.post(url, headers=headers, json=payload, timeout=600) - response.raise_for_status() - data = response.json() - for item in data.get('content', []): - if item.get('type') == 'text': - return item.get('text', '') - return str(data) - - -# ============== 分析器 ============== - -class DeepAnalyzer: - """深度内容分析器""" - - def __init__(self, ai_client: SimpleAIClient = None): - self.ai_client = ai_client or SimpleAIClient() - - def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any], content_type: str = "video") -> AnalysisResult: - """一次性调用 AI 生成完整分析报告 - - Args: - metadata: 元数据字典 - content: 内容字典 - content_type: 内容类型，"video" 或 "article" - """ - - # 标准化元数据 - metadata = self._normalize_metadata(metadata) - - title = metadata.get('title', '未知') - duration = metadata.get('duration', 0) - author = metadata.get('author', '') - pubtime = metadata.get('pubtime', '') - url = metadata.get('url', '') - bvid = metadata.get('bvid', '') - - # 推断 platform：优先使用元数据中的值，否则根据 URL 或 content_type 推断 - platform = metadata.get('platform', '') - if not platform: - if 'zhihu.com' in url or content_type in ('answer', 'zhuanlan', 'question'): - platform = 'zhihu' - elif 'bilibili.com' in url or bvid: - platform = 'bilibili' - else: - platform = 'bilibili' # 默认值 - - # 清理 author 字段：移除零宽度空格和前后空白 - author = self._clean_author(author) - - # 获取内容 - 支持多种字段名 - text = content.get('text', '') or content.get('subtitle', '') or content.get('subtitles', '') or content.get('description', '') - - print(f"[DeepAnalyzer] 开始分析: {title}") - print(f"[DeepAnalyzer] 类型: {content_type}, 内容长度: {len(text)} 字符") - - # 构建 prompt（根据类型选择不同模板） - prompt = self._build_prompt( - title=title, - duration=duration, - duration_str=self._format_duration(duration), - author=author, - pubtime=pubtime, - url=url, - bvid=bvid, - platform=platform, - text=text, - content_type=content_type - ) - - # 调用 AI - try: - full_analysis = self.ai_client.generate(prompt, max_tokens=16000) - except Exception as e: - print(f"[DeepAnalyzer] AI 调用失败: {e}") - raise - - # 解析结果 - result = self._parse_analysis(full_analysis, text, duration, content_type) - - # 文章/回答类型：追加原文 - if content_type in ("article", "answer", "zhuanlan"): - result.original_text = text - - return result - - def _build_prompt(self, title: str, duration: int, duration_str: str, author: str, - pubtime: str, url: str, bvid: str, platform: str, text: str, - content_type: str = "video") -> str: - """构建分析 prompt，根据内容类型选择不同模板""" - if content_type in ("article", "answer", "zhuanlan"): - return self._build_article_prompt(title, author, pubtime, url, platform, text) - else: - return self._build_video_prompt(title, duration, duration_str, author, pubtime, url, bvid, platform, text) - - def _build_video_prompt(self, title: str, duration: int, duration_str: str, author: str, - pubtime: str, url: str, bvid: str, platform: str, text: str) -> str: - """构建视频分析 prompt""" - - # 计算分片数（按10分钟分片） - num_chunks = max(1, (duration + 599) // 600) - - # 截取适当长度的文本（20-30K 字符） - max_chars = 30000 - if len(text) > max_chars: - text_sample = text[:max_chars] + "\n\n[...]" - else: - text_sample = text - - prompt = f"""你是一个专业的视频内容分析师。请根据以下信息生成深度分析报告： - -【视频信息】 -标题：{title} -作者：{author} -时长：{duration_str}（{duration}秒） -发布日期：{pubtime} -平台：{platform} -URL：{url} - -【内容】（完整字幕） -{text_sample} - -【输出要求】必须使用以下格式： - -## 标题 -[自己总结的简洁标题，禁止使用原标题] - -## 摘要 -[200-400字] - -## 主要主题 -[从内容中提取不超过10个主题，每个不超过10字] -- 主题1 -- 主题2 -... - -## 关键见解 -[若干条深刻见解] -- 见解1 -- 见解2 -... - -## 标签 -[从内容中提取3-5个精炼标签，每个不超过10字] -- 标签1 -- 标签2 -... - -## 时间轴 -### 第 1 部分 (00:00 - 10:00) -[详细分析，至少500字] - -### 第 2 部分 (10:00 - 20:00) -[详细分析，至少500字] - -...（以此类推） - -【重要】分析应足够详尽，读者阅读后可了解视频核心内容，无需再看原视频。""" - - return prompt - - def _build_article_prompt(self, title: str, author: str, pubtime: str, - url: str, platform: str, text: str) -> str: - """构建文章分析 prompt""" - - # 文章保留完整内容（不截断，因为需要追加原文） - # 但 prompt 中只取前 20000 字符供 AI 分析 - max_chars = 20000 - if len(text) > max_chars: - text_sample = text[:max_chars] + "\n\n[...内容过长已截断，但原文会完整保留...]" - else: - text_sample = text - - prompt = f"""你是一个专业的文章内容分析师。请根据以下信息生成摘要： - -【文章信息】 -标题：{title} -作者：{author} -发布日期：{pubtime} -平台：{platform} -URL：{url} - -【内容】 -{text_sample} - -【输出要求】必须使用以下格式： - -## 摘要 -[300字以内的摘要，描述文章的核心观点和主要内容] - -## 主要主题 -[从内容中提取不超过5个主题，每个不超过10字] -- 主题1 -- 主题2 -... - -## 标签 -[从内容中提取3-5个精炼标签，每个不超过10字] -- 标签1 -- 标签2 -... - -【重要】 -1. 摘要应精炼概括文章核心观点，让读者快速了解文章主旨 -2. 不要输出时间轴 -3. 不要输出关键见解（除非是特别重要的观点）""" - - return prompt - - def _parse_analysis(self, analysis_text: str, original_text: str, duration: int, - content_type: str = "video") -> AnalysisResult: - """解析 AI 返回的分析结果""" - result = AnalysisResult() - result.full_analysis = analysis_text - - # 提取各部分（使用正则） - lines = analysis_text.split('\n') - - # 简单解析：按 ## 标题分割 - current_section = "" - in_section = None - - for line in lines: - if line.startswith('##'): - # 保存上一个 section - if in_section: - self._store_section(result, in_section, current_section) - # 开始新 section - section_name = line.lstrip('#').strip() - in_section = section_name - current_section = "" - elif in_section: - current_section += line + "\n" - - # 存储最后一个 section - if in_section: - self._store_section(result, in_section, current_section) - - # 如果解析失败，使用原始文本 - if not result.summary: - result.summary = analysis_text[:500] - - return result - - def _store_section(self, result: AnalysisResult, section_name: str, content: str): - """存储解析到的 section""" - content = content.strip() - if not content: - return - - section_lower = section_name.lower() - - if '摘要' in section_lower: - result.summary = content[:500] - elif '主题' in section_lower: - # 提取主题 - topics = [] - for line in content.split('\n'): - line = line.strip().lstrip('-').strip() - if line and len(line) <= 15: - topics.append(line) - result.main_topics = topics[:10] - elif '见解' in section_lower or '观点' in section_lower: - # 提取见解 - insights = [] - for line in content.split('\n'): - line = line.strip().lstrip('-').strip() - if line and len(line) > 5: - insights.append(line) - result.key_insights = insights - elif '标签' in section_lower: - # 提取标签 - tags = [] - for line in content.split('\n'): - line = line.strip().lstrip('-').strip() - if line and len(line) <= 10: - tags.append(line) - result.recommended_tags = tags[:5] - - def _normalize_metadata(self, metadata: Dict) -> Dict: - """标准化元数据""" - if not metadata: - return {} - - # 如果已经扁平 - if 'title' in metadata: - duration = metadata.get('duration', 0) - if isinstance(duration, str) and ':' in duration: - parts = duration.split(':') - try: - metadata['duration'] = int(parts[0]) * 60 + int(parts[1]) - except: - pass - return metadata - - # 嵌套结构 - if 'metadata' in metadata: - meta = metadata['metadata'] - normalized = {} - for k, v in meta.items(): - normalized[k] = v - # 处理 duration - duration = normalized.get('duration', 0) - if isinstance(duration, str) and ':' in duration: - parts = duration.split(':') - try: - normalized['duration'] = int(parts[0]) * 60 + int(parts[1]) - except: - pass - return normalized - - return metadata - - def _format_duration(self, seconds: int) -> str: - """格式化时长""" - h = seconds // 3600 - m = (seconds % 3600) // 60 - s = seconds % 60 - if h > 0: - return f"{h}:{m:02d}:{s:02d}" - return f"{m}:{s:02d}" - - def _clean_author(self, author: str) -> str: - """清理 author 字段：移除零宽度空格和前后空白""" - if not author: - return '' - # 移除零宽度空格 (U+200B) 和其他不可见字符 - invisible_chars = ['\u200b', '\u200c', '\u200d', '\ufeff', '\u00ad'] - cleaned = author - for char in invisible_chars: - cleaned = cleaned.replace(char, '') - # 移除前后空白和换行 - cleaned = cleaned.strip() - # 如果包含换行符，只取第一行 - if '\n' in cleaned: - cleaned = cleaned.split('\n')[0].strip() - return cleaned - - -# ============== 工具函数 ============== - -def load_json(file_path: str) -> Dict: - """加载 JSON 文件""" - if not file_path or not os.path.exists(file_path): - return {} - with open(file_path, 'r', encoding='utf-8') as f: - return json.load(f) - - -def generate_frontmatter(metadata: Dict, result: AnalysisResult, content_type: str = "video") -> str: - """生成 Markdown frontmatter""" - # 提取标签 - tags = result.recommended_tags or [] - - # 格式化日期 - pubtime = metadata.get('pubtime', '') - if pubtime: - date_str = str(pubtime)[:10] - else: - date_str = datetime.now().strftime('%Y-%m-%d') - - # 清理 author 字段 - author = metadata.get('author', '未知') - author = _clean_string(author) - - # 推断 platform - url = metadata.get('url', '') - platform = metadata.get('platform', '') - if not platform: - if 'zhihu.com' in url or content_type in ('answer', 'zhuanlan', 'question'): - platform = 'zhihu' - elif 'bilibili.com' in url or metadata.get('bvid'): - platform = 'bilibili' - else: - platform = 'bilibili' - - fm = ["---", - f"title: \"{metadata.get('title', '未命名')}\"", - f"date: {date_str}", - f"author: {author}", - f"platform: {platform}"] - - # 视频类型添加 duration 和 bvid - if content_type == "video": - fm.insert(3, f"duration: {metadata.get('duration_str', metadata.get('duration', 0))}") - bvid = metadata.get('bvid', '') - if bvid: - fm.insert(5, f"bvid: {bvid}") - - # 添加 URL - url = metadata.get('url', '') - if url: - fm.append(f"url: {url}") - - if tags: - fm.append("tags: [" + ", ".join(f'"{t}"' for t in tags) + "]") - - fm.append("---") - return '\n'.join(fm) - - -def _clean_string(text: str) -> str: - """清理字符串：移除零宽度空格和前后空白""" - if not text: - return '' - # 移除零宽度空格 (U+200B) 和其他不可见字符 - invisible_chars = ['\u200b', '\u200c', '\u200d', '\ufeff', '\u00ad'] - cleaned = text - for char in invisible_chars: - cleaned = cleaned.replace(char, '') - # 移除前后空白和换行 - cleaned = cleaned.strip() - # 如果包含换行符，只取第一行 - if '\n' in cleaned: - cleaned = cleaned.split('\n')[0].strip() - return cleaned - - -# ============== 主程序 ============== - -def main(): - parser = argparse.ArgumentParser(description="Deep Analyzer - AI 深度内容分析工具") - parser.add_argument("--metadata", "-m", required=True, help="元数据 JSON 文件路径") - parser.add_argument("--content", "-c", required=True, help="内容 JSON 文件路径") - parser.add_argument("--output", "-o", required=True, help="输出 Markdown 文件路径") - parser.add_argument("--type", "-t", default="video", choices=["video", "article"], help="内容类型") - parser.add_argument("--api-key", "-k", default=None, help="AI API Key") - parser.add_argument("--provider", "-p", default=None, choices=["openai", "zhipu", "minimax"], help="AI 提供商") - parser.add_argument("--model", default=None, help="AI 模型") - parser.add_argument("--base-url", "-u", default=None, help="API Base URL") - parser.add_argument("--no-frontmatter", action="store_true", help="不生成 frontmatter") - - args = parser.parse_args() - - # 加载数据 - print(f"[Main] 加载元数据: {args.metadata}") - raw_metadata = load_json(args.metadata) - - print(f"[Main] 加载内容: {args.content}") - content = load_json(args.content) - - # 标准化元数据 - metadata = {} - if 'title' in raw_metadata: - metadata = raw_metadata.copy() - # 处理 duration - duration = metadata.get('duration', 0) - if isinstance(duration, str) and ':' in duration: - parts = duration.split(':') - try: - metadata['duration'] = int(parts[0]) * 60 + int(parts[1]) - except: - pass - metadata['duration_str'] = format_duration(metadata.get('duration', 0)) - elif 'metadata' in raw_metadata: - meta = raw_metadata['metadata'] - metadata = meta.copy() - duration = metadata.get('duration', 0) - if isinstance(duration, str) and ':' in duration: - parts = duration.split(':') - try: - metadata['duration'] = int(parts[0]) * 60 + int(parts[1]) - except: - pass - metadata['duration_str'] = format_duration(metadata.get('duration', 0)) - - # 初始化 - ai_client = SimpleAIClient( - api_key=args.api_key or os.environ.get("AI_API_KEY", ""), - provider=args.provider or os.environ.get("AI_PROVIDER", ""), - model=args.model or os.environ.get("AI_MODEL", "glm-5"), - base_url=args.base_url or os.environ.get("AI_BASE_URL", "") - ) - analyzer = DeepAnalyzer(ai_client) - - # 执行分析 - print(f"[Main] 开始分析...") - result = analyzer.analyze(metadata, content, args.type) - - # 使用 AI 生成的原始内容（保留 Markdown 格式） - full_analysis = result.full_analysis - - # 删除内容中的"标签" section（只保留 frontmatter 中的 tags） - # 匹配 ## 标签 或 标签（无##）到下一个 ## 或文件结尾之间的内容 - full_analysis = re.sub(r'(## ?)?标签\n[\s\S]*?(?=\n## ?|\n时间轴|---|\Z)', '', full_analysis) - # 清理多余的空行 - full_analysis = re.sub(r'\n{3,}', '\n\n', full_analysis) - - # 文章类型：追加原文 - if args.type == "article" and result.original_text: - full_analysis = full_analysis.rstrip() + "\n\n## 原文\n\n" + result.original_text - - # 写入输出 - os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) - - with open(args.output, 'w', encoding='utf-8') as f: - # 添加 frontmatter - if not args.no_frontmatter: - fm = generate_frontmatter(metadata, result, args.type) - f.write(fm + "\n\n") - - # 写入分析内容 - f.write(full_analysis) - - # 输出结果 - output = { - "success": True, - "filename": Path(args.output).name, - "filepath": str(Path(args.output).resolve()), - "message": "深度分析完成", - "stats": { - "summary_length": len(result.summary) if result.summary else 0, - "topics_count": len(result.main_topics), - "insights_count": len(result.key_insights), - "tags_count": len(result.recommended_tags) - } - } - print(json.dumps(output, ensure_ascii=False, indent=2)) - - -def format_duration(seconds: int) -> str: - """格式化时长""" - h = seconds // 3600 - m = (seconds % 3600) // 60 - s = seconds % 60 - if h > 0: - return f"{h}:{m:02d}:{s:02d}" - return f"{m}:{s:02d}" - - -if __name__ == "__main__": - try: - main() - except Exception as e: - print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2)) - sys.exit(1) diff --git a/kb-deep-analyzer/deep_analyzer.py.bak b/kb-deep-analyzer/deep_analyzer.py.bak deleted file mode 100644 index 59c62c1..0000000 --- a/kb-deep-analyzer/deep_analyzer.py.bak +++ /dev/null @@ -1,1727 +0,0 @@ -#!/usr/bin/env python3 -"""Deep Analyzer - AI 深度内容分析工具 - -支持根据内容时长选择不同分析策略： -- 短视频 (<10分钟): 轻量分析 -- 长视频 (>=10分钟): 详细时间轴分析 - -Usage: - python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md [--type video] - python deep_analyzer.py --help -""" - -import argparse -import json -import os -import sys -import re -from datetime import datetime -from pathlib import Path -from typing import Dict, Any, List, Optional, Tuple -from dataclasses import dataclass, field, asdict - - -# ============== 数据模型 ============== - -@dataclass -class TimelineEntry: - """时间轴条目""" - index: int = 0 - start_time: str = "" - end_time: str = "" - title: str = "" - summary: str = "" - - -@dataclass -class AnalysisResult: - """分析结果""" - summary: str = "" - main_topics: List[str] = field(default_factory=list) - key_insights: List[str] = field(default_factory=list) - action_items: List[str] = field(default_factory=list) - recommended_tags: List[str] = field(default_factory=list) - full_analysis: str = "" - timeline: List[Dict[str, Any]] = field(default_factory=list) - - -# ============== AI 客户端 ============== - -# ============== 流式写入器 ============== - -import threading - - -class StreamingWriter: - """ - 流式写入器 - 支持多线程并发写入到文档不同位置 - - 使用方法: - 1. 初始化: writer = StreamingWriter(output_path, num_chunks) - 2. 预先创建带占位符的文档结构 - 3. 每批完成后调用 writer.write_batch(batch_results) - 4. 完成后调用 writer.mark_complete() 清理占位符 - """ - - def __init__(self, output_path: str, num_chunks: int = 0, batch_size: int = 3): - self.output_path = Path(output_path) - self.num_chunks = num_chunks - self.batch_size = batch_size - self.lock = threading.Lock() - self.written = {} # 记录已写入的位置 - self.batch_num = 0 - - def initialize_document(self, metadata: Dict, num_chunks: int) -> str: - """ - 初始化文档结构，返回带占位符的内容 - - 文档结构: - --- - title: "xxx" - ... - --- - - # 标题 - - ## 摘要 - - ## 主要主题 - - ## 关键见解 - - ## 建议行动 - - ## 时间轴 - ### 第 N 部分 (00:00 - 00:00) - [ANALYZING...] - """ - self.num_chunks = num_chunks - - # 生成时间轴占位符 - timeline_placeholders = "" - for i in range(num_chunks): - timeline_placeholders += f"""### 第 {i + 1} 部分 (待分析) -[ANALYZING_BATCH_{i // self.batch_size + 1}] - -""" - - content = f"""--- -title: "{metadata.get('title', '未命名')}" -date: {datetime.now().strftime('%Y-%m-%d')} -duration: {format_duration(metadata.get('duration', 0))} -author: {metadata.get('author', '未知')} -analyze_time: {datetime.now().strftime('%Y-%m-%d %H:%M')} -bvid: {metadata.get('bvid', '')} -platform: {metadata.get('platform', 'bilibili')} ---- - -# {metadata.get('title', '未命名')} - -## 摘要 -[WAITING_FOR_ANALYSIS...] - -## 主要主题 -[WAITING_FOR_ANALYSIS...] - -## 关键见解 -[WAITING_FOR_ANALYSIS...] - -## 建议行动 -[WAITING_FOR_ANALYSIS...] - -## 时间轴 -{timeline_placeholders} - ---- -*分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}* -""" - - # 写入初始化内容 - self.output_path.parent.mkdir(parents=True, exist_ok=True) - with open(self.output_path, 'w', encoding='utf-8') as f: - f.write(content) - - print(f"[StreamingWriter] 初始化文档: {self.output_path}") - return content - - def write_batch(self, batch_results: List[Dict], num_batches: int) -> None: - """ - 写入一批结果 - - Args: - batch_results: 当前批的结果列表 - num_batches: 总批次数 - """ - self.batch_num += 1 - - with self.lock: - # 读取当前内容 - if not self.output_path.exists(): - print(f"[StreamingWriter] 警告: 文件不存在 {self.output_path}") - return - - content = self.output_path.read_text(encoding='utf-8') - - for result in batch_results: - idx = result['index'] - start_time = result['start_time'] - end_time = result['end_time'] - summary = result['summary'] - - # 查找并替换占位符 - old_placeholder = f"### 第 {idx + 1} 部分 ({start_time} - {end_time})\n[ANALYZING_BATCH_{self.batch_num}]" - new_content = f"""### 第 {idx + 1} 部分 ({start_time} - {end_time}) -{summary}""" - - content = content.replace(old_placeholder, new_content, 1) - self.written[idx] = True - - print(f"[StreamingWriter] 写入段 {idx + 1}/{self.num_chunks}") - - # 如果这是最后一批，更新其他状态 - if self.batch_num >= num_batches: - # 更新摘要等 - pass - - # 写回文件 - self.output_path.write_text(content, encoding='utf-8') - print(f"[StreamingWriter] 批次 {self.batch_num}/{num_batches} 完成") - - def update_summary(self, summary_text: str, topics: List[str], insights: List[str]) -> None: - """更新摘要等全局内容""" - with self.lock: - if not self.output_path.exists(): - return - - content = self.output_path.read_text(encoding='utf-8') - - # 更新摘要 - if "[WAITING_FOR_ANALYSIS...]" in content: - content = content.replace("## 摘要\n[WAITING_FOR_ANALYSIS...]", f"## 摘要\n{summary_text}") - - # 更新主题 - topics_text = "\n".join(f"- {t}" for t in topics) - if "## 主要主题\n[WAITING_FOR_ANALYSIS...]" in content: - content = content.replace("## 主要主题\n[WAITING_FOR_ANALYSIS...]", f"## 主要主题\n{topics_text}") - - # 更新见解 - insights_text = "\n".join(f"- {i}" for i in insights) - if "## 关键见解\n[WAITING_FOR_ANALYSIS...]" in content: - content = content.replace("## 关键见解\n[WAITING_FOR_ANALYSIS...]", f"## 关键见解\n{insights_text}") - - self.output_path.write_text(content, encoding='utf-8') - print(f"[StreamingWriter] 更新摘要和主题") - - def cleanup_placeholder(self, placeholder: str) -> None: - """清理单个占位符""" - with self.lock: - if not self.output_path.exists(): - return - - content = self.output_path.read_text(encoding='utf-8') - content = content.replace(placeholder, "") - self.output_path.write_text(content, encoding='utf-8') - - -class SimpleAIClient: - """简化的 AI 客户端 - 支持多提供商 fallback""" - - # 时长阈值（秒） - SHORT_VIDEO_THRESHOLD = 600 # 10分钟 - LONG_VIDEO_THRESHOLD = 1800 # 30分钟 - - 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 or os.environ.get("AI_PROVIDER", "") - self.model = model or os.environ.get("AI_MODEL", "gpt-4") - self.base_url = base_url or os.environ.get("AI_BASE_URL", "") - self.max_retries = 3 # 最大重试次数 - self.max_concurrent = 3 # 最大并发数 - - def generate(self, prompt: str, max_tokens: int = 1000, task_info: str = None) -> str: - """生成文本，支持重试""" - # 每次调用时重新读取环境变量，解决 exec 传递问题 - api_key = os.environ.get("AI_API_KEY", self.api_key) - provider = os.environ.get("AI_PROVIDER", self.provider) - model = os.environ.get("AI_MODEL", self.model) - base_url = os.environ.get("AI_BASE_URL", self.base_url) - - if not api_key: - raise ValueError("AI_API_KEY 未设置，请配置环境变量") - - task_desc = f"[{task_info}]" if task_info else "" - - last_error = None - for attempt in range(self.max_retries): - try: - import time - start_time = time.time() - print(f"[AI Client] {task_desc} 开始调用 AI...") - - if provider == "openai": - result = self._call_openai(prompt, max_tokens, api_key, model, base_url) - elif provider == "zhipu": - result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url) - elif provider == "minimax": - result = self._call_minimax(prompt, max_tokens, api_key, model, base_url) - else: - raise ValueError(f"不支持的 AI 提供商: {provider}") - - elapsed = time.time() - start_time - print(f"[AI Client] {task_desc} AI 调用完成，耗时: {elapsed:.2f}秒") - return result - except Exception as e: - elapsed = time.time() - start_time if 'start_time' in dir() else 0 - last_error = e - if attempt < self.max_retries - 1: - print(f"[AI Client] {task_desc} 第 {attempt + 1} 次调用失败 ({elapsed:.2f}秒)，{self.max_retries - attempt - 1} 次重试剩余: {e}") - continue - else: - print(f"[AI Client] {task_desc} 调用失败，已重试 {self.max_retries} 次: {e}") - raise - - def generate_json(self, prompt: str) -> Dict[str, Any]: - """生成 JSON 格式响应""" - response = self.generate(prompt, max_tokens=4000) - try: - return json.loads(response) - except json.JSONDecodeError: - # 尝试从文本中提取 JSON - return self._parse_json_from_text(response) - - def _call_openai(self, prompt: str, max_tokens: int, api_key: str = None, model: str = None, base_url: str = None) -> str: - """调用 OpenAI API""" - import requests - url = f"{self.base_url}/chat/completions" if self.base_url else "https://api.openai.com/v1/chat/completions" - - headers = { - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/json" - } - - payload = { - "model": self.model, - "max_tokens": max_tokens, - "messages": [{"role": "user", "content": prompt}] - } - - response = requests.post(url, headers=headers, json=payload, timeout=600) - response.raise_for_status() - response.raise_for_status() - return response.json()['choices'][0]['message']['content'] - - def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str = None, model: str = None, base_url: str = None) -> str: - """调用智谱 AI API - 返回 reasoning_content""" - import requests - base_url = self.base_url or "https://open.bigmodel.cn/api/coding/paas/v4" - url = f"{base_url}/chat/completions" - - headers = {"Authorization": f"Bearer {self.api_key}"} - payload = { - "model": self.model, - "messages": [{"role": "user", "content": prompt}], - "max_tokens": max_tokens - } - - response = requests.post(url, headers=headers, json=payload, timeout=600) - response.raise_for_status() - response.raise_for_status() - data = response.json() - - 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: - # 从 reasoning_content 中提取最后几句话作为答案 - sentences = reasoning.strip().split('。') - # 取最后几个完整句子 - useful = [s.strip() for s in sentences if s.strip() and len(s.strip()) > 10] - if useful: - return '。'.join(useful[-3:]) + '。' - return reasoning.strip()[:500] - return content - - def _parse_json_from_text(self, text: str) -> Dict[str, Any]: - """从文本中解析 JSON，改进版本，支持更多格式""" - # 尝试直接解析 - try: - return json.loads(text) - except json.JSONDecodeError: - pass - - # 尝试提取 JSON 块 - json_match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', text) - if json_match: - try: - return json.loads(json_match.group(1)) - except: - pass - - # 尝试匹配 { ... } - json_match = re.search(r'(\{[\s\S]*\})', text) - if json_match: - try: - return json.loads(json_match.group(1)) - except: - pass - - return {'raw_response': text} - - def _call_minimax(self, prompt: str, max_tokens: int, api_key: str = None, model: str = None, base_url: str = None) -> str: - """调用 MiniMax API""" - import requests - url = self.base_url or "https://api.minimaxi.com/anthropic/v1/messages" - - headers = { - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/json" - } - - payload = { - "model": self.model, - "max_tokens": max_tokens, - "messages": [{"role": "user", "content": prompt}] - } - - response = requests.post(url, headers=headers, json=payload, timeout=600) - response.raise_for_status() - response.raise_for_status() - data = response.json() - for item in data.get('content', []): - if item.get('type') == 'text': - return item.get('text', '') - return str(data) - - - -# ============== 深度分析器 ============== - -class DeepAnalyzer: - """深度内容分析器""" - - SHORT_VIDEO_THRESHOLD = 600 # 10分钟 - LONG_VIDEO_THRESHOLD = 1800 # 30分钟 - - def __init__(self, ai_client: SimpleAIClient = None): - self.ai_client = ai_client or SimpleAIClient() - - def analyze( - self, - metadata: Dict[str, Any], - content: Dict[str, Any], - content_type: str = "video", - on_batch_complete: callable = None, - batch_size: int = 3 - ) -> AnalysisResult: - """ - 分析内容 - - Args: - metadata: 元数据字典 - content: 内容数据字典（包含 subtitle, description 等） - content_type: 内容类型 - on_batch_complete: 每批完成时的回调函数 (batch_results) => None - batch_size: 每批处理数（默认3） - - Returns: - AnalysisResult 分析结果 - """ - # 先标准化元数据 - metadata = normalize_metadata(metadata) - - title = metadata.get('title', '未知') - duration = metadata.get('duration', 0) - - # 支持多种内容格式 - 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] 文本存在: {has_content}, 分段数: {len(segments)}") - print(f"[DeepAnalyzer] 文本长度: {len(text) if text else 0} 字符") - - CHUNK_THRESHOLD = 3000 - use_chunking = has_content and len(text) > CHUNK_THRESHOLD - - if has_content or has_segments: - if use_chunking: - print(f"[DeepAnalyzer] 文本超过 {CHUNK_THRESHOLD} 字，使用批处理分块分析") - result = self._analyze_chunks_concurrent( - title, text, duration, segments, metadata, - batch_size=batch_size, - on_segment_complete=on_batch_complete - ) - elif duration < self.SHORT_VIDEO_THRESHOLD: - print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)") - result = self._analyze_short_with_subtitle(title, text, duration, segments) - else: - print(f"[DeepAnalyzer] 使用详细时间轴分析策略 (时长 >= 10分钟)") - result = self._analyze_long_with_subtitle(title, text, duration, segments, content_type, metadata) - else: - print(f"[DeepAnalyzer] 降级处理（无内容）") - result = self._analyze_fallback(title, text, metadata) - - return result - - def _analyze_short_with_subtitle(self, title: str, text: str, duration: int, segments: list = None) -> AnalysisResult: - """有内容的短视频轻量分析""" - result = AnalysisResult() - 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_timeline_from_segments(title, duration, segments) - - return result - - def _analyze_long_with_subtitle(self, title: str, text: str, duration: int, - segments: list = None, content_type: str = None, - metadata: Dict = None) -> AnalysisResult: - """有内容的长视频详细分析""" - result = AnalysisResult() - - if duration >= self.SHORT_VIDEO_THRESHOLD and content_type == "video": - # 10分钟以上使用分段分析 - result = self._analyze_in_chunks(title, text, duration, metadata) - else: - # 直接分析 - 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_timeline_from_segments(title, duration, segments) - - return result - - def _analyze_in_chunks(self, title: str, subtitle: str, duration: int, metadata: Dict = None) -> AnalysisResult: - """分段分析长视频 - 动态分段策略""" - result = AnalysisResult() - - # 常量配置 - CHUNK_MAX_CHARS = 3000 # 每个chunk最大字符数 - MIN_CHUNKS = 3 # 最少分段数 - MAX_API_CHUNKS = 20 # 最多API调用次数 - - total_chars = len(subtitle) - - # 1. 动态计算分段数 - # 计算需要多少个chunk（每个不超过3000字） - min_required = total_chars // CHUNK_MAX_CHARS + 1 - # 根据时长限制最大分段 - if duration < 1800: # <30分钟: 最多6段 - max_chunks = 6 - elif duration < 7200: # <2小时: 最多12段 - max_chunks = 12 - else: # >2小时: 最多20段 - max_chunks = 20 - - num_segments = min(max(min_required, MIN_CHUNKS), MAX_API_CHUNKS) - - print(f"[DeepAnalyzer] 字幕{total_chars}字符，分{num_segments}段分析") - - # 2. 按字符数分段（保持语义完整） - chunk_size = total_chars // num_segments - chunks = [] - chunk_boundaries = [] # 记录每段的字符边界 - - for i in range(num_segments): - start = i * chunk_size - if i == num_segments - 1: - # 最后一段取到结尾 - end = total_chars - chunk_text = subtitle[start:] - else: - # 在自然断点处分割 - end = self._find_breakpoint(subtitle, start, start + chunk_size) - chunk_text = subtitle[start:end] - chunks.append(chunk_text) - chunk_boundaries.append((start, end)) - - print(f"[DeepAnalyzer] 各段字符: {[len(c) for c in chunks]}") - - # 3. 用字符比例估算时间 - char_per_second = total_chars / duration if duration > 0 else 1 - - # 合并结果 - chunk_summaries = [] - - for i, chunk_text in enumerate(chunks): - start_char, end_char = chunk_boundaries[i] - # 用字符位置估算时间 - start_sec = max(0, int(start_char / char_per_second)) - end_sec = min(duration, int(end_char / char_per_second)) - start_time = self._format_time(start_sec) - end_time = self._format_time(end_sec) - - # AI 分析每段（传入元数据以添加概述） - chunk_summary = self._analyze_chunk(chunk_text, i + 1, metadata) - chunk_summaries.append({ - 'index': i, - 'start_time': start_time, - 'end_time': end_time, - 'start_sec': start_sec, - 'end_sec': end_sec, - 'summary': chunk_summary - }) - - # 合并结果 - result = self._merge_summaries(chunk_summaries, title) - result.timeline = self._build_timeline_from_chunks(chunk_summaries) - - return result - - def _analyze_in_chunks_with_segments(self, title: str, text: str, duration: int, segments: list = None, metadata: Dict = None) -> AnalysisResult: - """ - 使用实际字幕时间戳进行分块分析 - - 当有 SRT 格式的 segments 时，使用实际时间戳来分块 - 否则退化为字符位置估算 - """ - result = AnalysisResult() - - CHUNK_MAX_CHARS = 3000 - total_chars = len(text) - - # 检查是否有有效的 SRT segments - has_valid_segments = segments and len(segments) > 0 and all('start' in seg and 'text' in seg for seg in segments) - - if has_valid_segments: - print(f"[DeepAnalyzer] 使用 SRT 时间戳进行分块") - # 按时间戳分块 - chunks = [] - chunk_times = [] - current_chunk = "" - current_start = segments[0]['start'] - current_end = "" - - for seg in segments: - seg_text = f"[{seg['start']}] {seg['text']}\n" - - if len(current_chunk) + len(seg_text) > CHUNK_MAX_CHARS and current_chunk: - # 保存当前块 - chunks.append(current_chunk.strip()) - chunk_times.append((current_start, current_end or seg['start'])) - # 开始新块 - current_chunk = seg_text - current_start = seg['start'] - else: - current_chunk += seg_text - current_end = seg['end'] - - # 保存最后一块 - if current_chunk: - chunks.append(current_chunk.strip()) - chunk_times.append((current_start, current_end)) - - print(f"[DeepAnalyzer] SRT分块: {len(chunks)} 段") - else: - # 退化为字符位置估算 - print(f"[DeepAnalyzer] 无有效时间戳，使用字符位置分块") - min_required = total_chars // CHUNK_MAX_CHARS + 1 - num_segments = min(max(min_required, 3), 20) - - chunk_size = total_chars // num_segments - chunks = [] - chunk_times = [] - - for i in range(num_segments): - start = i * chunk_size - if i == num_segments - 1: - end = total_chars - else: - end = self._find_breakpoint(text, start, start + chunk_size) - - chunks.append(text[start:end]) - # 估算时间 - char_per_second = total_chars / duration if duration > 0 else 1 - start_sec = int(start / char_per_second) - end_sec = int(end / char_per_second) - chunk_times.append((self._format_time(start_sec), self._format_time(end_sec))) - - print(f"[DeepAnalyzer] 字符分块: {len(chunks)} 段") - - print(f"[DeepAnalyzer] 各段字符: {[len(c) for c in chunks]}") - - # 分析每个块 - chunk_summaries = [] - for i, chunk_text in enumerate(chunks): - start_time, end_time = chunk_times[i] - - # AI 分析每段（传入元数据以添加概述） - chunk_summary = self._analyze_chunk(chunk_text, i + 1, metadata) - chunk_summaries.append({ - 'index': i, - 'start_time': start_time, - 'end_time': end_time, - 'summary': chunk_summary - }) - - # 合并结果 - result = self._merge_summaries(chunk_summaries, title) - result.timeline = self._build_timeline_from_chunks(chunk_summaries) - - return result - - def _analyze_chunks_concurrent( - self, - title: str, - text: str, - duration: int, - segments: list = None, - metadata: Dict = None, - max_concurrent: int = 3, - batch_size: int = 3, - on_segment_complete: callable = None - ) -> AnalysisResult: - """ - 批处理分块分析（流式输出，避免爆上下文） - - Args: - title: 视频标题 - text: 完整文本 - duration: 时长（秒） - segments: SRT segments - metadata: 元数据 - batch_size: 每批处理数（默认3） - on_batch_complete: 每批完成时的回调 (batch_results) => None - - Returns: - AnalysisResult 分析结果 - """ - import concurrent.futures - - # 确保环境变量在子线程中可用 - _worker_env = { - 'AI_API_KEY': os.environ.get('AI_API_KEY', ''), - 'AI_PROVIDER': os.environ.get('AI_PROVIDER', ''), - 'AI_MODEL': os.environ.get('AI_MODEL', ''), - 'AI_BASE_URL': os.environ.get('AI_BASE_URL', ''), - } - - def init_worker(): - """初始化 worker 线程的环境变量""" - os.environ.update(_worker_env) - - result = AnalysisResult() - CHUNK_MAX_CHARS = 3000 - total_chars = len(text) - num_chunks = 0 - - # 生成分块 - has_valid_segments = segments and len(segments) > 0 and all('start' in seg and 'text' in seg for seg in segments) - - if has_valid_segments: - print(f"[DeepAnalyzer] 使用 SRT 时间戳进行分块") - chunks = [] - chunk_times = [] - current_chunk = "" - current_start = segments[0]['start'] - current_end = "" - - for seg in segments: - seg_text = f"[{seg['start']}] {seg['text']}\n" - if len(current_chunk) + len(seg_text) > CHUNK_MAX_CHARS and current_chunk: - chunks.append(current_chunk.strip()) - chunk_times.append((current_start, current_end or seg['start'])) - current_chunk = seg_text - current_start = seg['start'] - else: - current_chunk += seg_text - current_end = seg['end'] - - if current_chunk: - chunks.append(current_chunk.strip()) - chunk_times.append((current_start, current_end)) - - num_chunks = len(chunks) - else: - print(f"[DeepAnalyzer] 使用字符位置分块") - min_required = total_chars // CHUNK_MAX_CHARS + 1 - num_chunks = min(max(min_required, 3), 20) - chunk_size = total_chars // num_chunks - chunks = [] - chunk_times = [] - - for i in range(num_chunks): - start = i * chunk_size - if i == num_chunks - 1: - end = total_chars - else: - end = self._find_breakpoint(text, start, start + chunk_size) - chunks.append(text[start:end]) - char_per_second = total_chars / duration if duration > 0 else 1 - start_sec = int(start / char_per_second) - end_sec = int(end / char_per_second) - chunk_times.append((self._format_time(start_sec), self._format_time(end_sec))) - - print(f"[DeepAnalyzer] 分块: {num_chunks} 段，批次大小: {batch_size}") - - import time - total_start_time = time.time() - - # 批处理执行 - batch_results = {} # 存储各批结果 - all_summaries = [] - - for batch_idx in range(0, num_chunks, batch_size): - batch_num = batch_idx // batch_size + 1 - total_batches = (num_chunks + batch_size - 1) // batch_size - batch_end = min(batch_idx + batch_size, num_chunks) - batch_chunks = chunks[batch_idx:batch_end] - batch_times = chunk_times[batch_idx:batch_end] - - batch_start_time = time.time() - print(f"[DeepAnalyzer] === 批次 {batch_num}/{total_batches} 开始 ===") - print(f"[DeepAnalyzer] 段 {batch_idx+1}-{batch_end}，预计 {len(batch_chunks)} 个并发调用") - - # 并发执行当前批次 - batch_tasks = [(i, batch_chunks[j], batch_times[j][0], batch_times[j][1]) - for j, i in enumerate(range(batch_idx, batch_end))] - - with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size, initializer=init_worker) as executor: - futures = {executor.submit(self._analyze_chunk, task[1], task[0] + 1, metadata): task[0] - for task in batch_tasks} - - completed_in_batch = 0 - for future in concurrent.futures.as_completed(futures): - idx = futures[future] - completed_in_batch += 1 - try: - summary = future.result() - batch_results[idx] = { - 'index': idx, - 'start_time': chunk_times[idx][0], - 'end_time': chunk_times[idx][1], - 'summary': summary - } - elapsed = time.time() - batch_start_time - print(f"[DeepAnalyzer] 段 {idx + 1}/{num_chunks} 完成 ({completed_in_batch}/{len(batch_chunks)}), 批次耗时: {elapsed:.2f}秒") - except Exception as e: - print(f"[DeepAnalyzer] 段 {idx + 1} 失败: {e}") - batch_results[idx] = { - 'index': idx, - 'start_time': chunk_times[idx][0], - 'end_time': chunk_times[idx][1], - 'summary': f"分析失败: {str(e)}" - } - - # 回调通知当前批次完成 - batch_summaries = [batch_results[i] for i in range(batch_idx, batch_end) if i in batch_results] - if on_segment_complete: - on_segment_complete(batch_summaries) - - all_summaries.extend(batch_summaries) - - batch_elapsed = time.time() - batch_start_time - print(f"[DeepAnalyzer] === 批次 {batch_num}/{total_batches} 完成，耗时: {batch_elapsed:.2f}秒 ===") - - total_elapsed = time.time() - total_start_time - print(f"[DeepAnalyzer] 所有分块分析完成，总耗时: {total_elapsed:.2f}秒") - - # 按顺序整理结果 - all_summaries = [all_summaries[i] for i in range(len(all_summaries))] - - # 合并结果 - print(f"[DeepAnalyzer] 开始合并各段摘要...") - merge_start = time.time() - result = self._merge_summaries(all_summaries, title) - merge_elapsed = time.time() - merge_start - print(f"[DeepAnalyzer] 摘要合并完成，耗时: {merge_elapsed:.2f}秒") - - result.timeline = self._build_timeline_from_chunks(all_summaries) - - return result - - def _find_breakpoint(self, text: str, start: int, preferred_end: int) -> int: - """ - 在自然断点处分割文本 - - 优先级： - 1. 双换行（段落分隔） - 2. 句号+换行（句末） - 3. 句号（句末） - 4. 换行（行末） - 5. 回到preferred_end - - Args: - text: 完整文本 - start: 起始位置 - preferred_end: 理想结束位置 - - Returns: - 实际结束位置 - """ - if not text or start >= len(text): - return start - - # 搜索范围：理想位置前后各500字符 - search_start = max(start, preferred_end - 500) - search_end = min(len(text), preferred_end + 500) - - # 优先级1: 双换行 - double_newline = text.rfind('\n\n', search_start, search_end) - if double_newline != -1: - return double_newline - - # 优先级2: 句号+换行 - period_newline = text.rfind('。\n', search_start, search_end) - if period_newline != -1: - return period_newline + 1 # 包含句号 - - # 优先级3: 句号（中文或英文） - last_period = max( - text.rfind('。', search_start, search_end), - text.rfind('.', search_start, search_end) - ) - if last_period != -1: - return last_period + 1 # 包含句号 - - # 优先级4: 换行 - last_newline = text.rfind('\n', search_start, search_end) - if last_newline != -1: - return last_newline - - # 没找到自然断点，返回理想位置 - return preferred_end - - def _analyze_chunk(self, chunk_text: str, chunk_index: int, metadata: Dict = None) -> str: - """分析单个分片""" - text_sample = self._get_text_sample(chunk_text, max_chars=3000) - - # 构建概述 - overview = "" - if metadata: - title = metadata.get('title', '') - description = metadata.get('description', '') - source = metadata.get('source', '') - - overview_parts = [] - if title: - overview_parts.append(f"视频标题：{title}") - if description: - overview_parts.append(f"视频描述：{description}") - if source: - overview_parts.append(f"来源：{source}") - - if overview_parts: - overview = "【概述】\n" + "\n".join(overview_parts) + "\n\n" - - prompt = f"""请对以下视频片段进行详细分析（限制在800字以内）： - -{overview}--- 片段内容 --- -{text_sample} - -请用中文详细回答，包括核心主题、关键要点和重要细节。""" - - try: - import time - start_time = time.time() - print(f"[DeepAnalyzer] 开始分析分片 {chunk_index}...") - result = self.ai_client.generate(prompt, max_tokens=2000, task_info=f"分片 {chunk_index}") - elapsed = time.time() - start_time - print(f"[DeepAnalyzer] 分片 {chunk_index} 分析完成，耗时: {elapsed:.2f}秒") - return result - except Exception as e: - print(f"[DeepAnalyzer] 分片 {chunk_index} 分析失败: {e}") - return f"第{chunk_index}段内容分析" - - def _merge_summaries(self, chunk_summaries: List[Dict], title: str) -> AnalysisResult: - """合并分段摘要""" - summaries_text = "\n\n".join([ - f"=== 第{s['index']+1}段 ({s['start_time']}-{s['end_time']}) ===\n{s['summary']}" - for s in chunk_summaries - ]) - - # 使用纯文本提示，不再要求 JSON 格式 - prompt = f"""基于以下视频各段分析，生成完整详细的分析报告： - -标题：{title} - -各段内容： -{summaries_text} - -请用中文详细回复，包含： -1. 整体摘要（300字以内） -2. 5个核心主题 -3. 4个关键见解 -4. 3条建议 -5. 详细分析（1500字以内）""" - - try: - # 使用纯文本生成而非 JSON - full_response = self.ai_client.generate(prompt, max_tokens=4000) - - # 解析响应 - result = AnalysisResult() - result.summary = self._extract_summary(full_response) - result.main_topics = self._extract_topics(full_response) - result.key_insights = self._extract_insights(full_response) - result.action_items = self._extract_actions(full_response) - result.full_analysis = full_response - return result - except Exception as e: - print(f"[DeepAnalyzer] 合并摘要失败: {e}") - return AnalysisResult( - summary=f"共{len(chunk_summaries)}段分析完成", - main_topics=[f"主题{i+1}" for i in range(min(4, len(chunk_summaries)))], - key_insights=["综合各段分析得出"], - full_analysis=summaries_text[:500] - ) - - def _extract_summary(self, text: str) -> str: - """从 AI 响应中提取摘要""" - lines = text.split('\n') - summary_lines = [] - capture = False - for line in lines: - if '摘要' in line or '总结' in line: - capture = True - if capture and line.strip(): - summary_lines.append(line.strip()) - if len(summary_lines) >= 3: - break - return ' '.join(summary_lines[:3])[:300] if summary_lines else text[:200] - - def _extract_topics(self, text: str) -> List[str]: - """从 AI 响应中提取主题""" - topics = [] - lines = text.split('\n') - for line in lines: - # 匹配常见主题格式 - if any(m in line for m in ['主题', '核心', '要点', '1.', '2.', '3.', '4.', '5.']): - # 清理并提取 - cleaned = line.strip().lstrip('0123456789.、 ').strip() - if cleaned and len(cleaned) > 2: - topics.append(cleaned) - if len(topics) >= 5: - break - return topics[:5] if topics else [f"主题{i+1}" for i in range(5)] - - def _extract_insights(self, text: str) -> List[str]: - """从 AI 响应中提取见解""" - insights = [] - lines = text.split('\n') - for line in lines: - if any(m in line for m in ['见解', '观点', '发现', '洞察']): - cleaned = line.strip().lstrip('0123456789.、 ').strip() - if cleaned and len(cleaned) > 2: - insights.append(cleaned) - if len(insights) >= 4: - break - return insights[:4] if insights else ["综合分析得出"] - - def _extract_actions(self, text: str) -> List[str]: - """从 AI 响应中提取建议""" - actions = [] - lines = text.split('\n') - for line in lines: - if any(m in line for m in ['建议', '行动', '做法', '可以', '应该']): - cleaned = line.strip().lstrip('0123456789.、 ').strip() - if cleaned and len(cleaned) > 2: - actions.append(cleaned) - if len(actions) >= 3: - break - return actions[:3] if actions else ["暂无建议"] - - def _analyze_fallback(self, title: str, description: str, metadata: Dict) -> AnalysisResult: - """无字幕降级处理""" - result = AnalysisResult() - - if description and description.strip(): - summary = description[:200] + "..." if len(description) > 200 else description - else: - summary = f"无字幕内容，标题：{title}" - - # 提取标签 - tags = ["未分类", "待处理"] - author = metadata.get('author', '') - source = metadata.get('source', '') - if author: - tags.append(f"作者:{author}") - if source: - tags.append(f"来源:{source}") - - result.summary = summary - result.main_topics = tags[:3] - result.recommended_tags = tags - result.key_insights = ["无字幕内容，仅能从描述获取基本信息"] - result.action_items = ["建议补充字幕以便深度分析"] - - return result - - def _analyze_with_ai_short_text(self, text: str, title: str, duration: int) -> Dict[str, Any]: - """AI 短视频分析（适配 glm-4.7 reasoning 模式）""" - prompt = f"""请分析以下视频内容： - -标题：{title} -时长：{self._format_time(duration)} - -字幕内容： -{text[:3000]} - -视频主题是什么？（不超过30字） -涉及哪些人物？（用逗号分隔） -核心观点是什么？（不超过50字）""" - - try: - response = self.ai_client.generate(prompt, max_tokens=2000) - # glm-4.7 主要返回 reasoning_content - analysis_text = response.strip() if response.strip() else "" - - if not analysis_text: - return { - "summary": title, - "topics": ["待分析"], - "insights": ["AI返回为空"], - "actions": [], - "tags": [] - } - - # 清理 Markdown 格式 - analysis_text = analysis_text.replace('**', '').replace('*', '') - analysis_text = re.sub(r'```[\s\S]*?```', '', analysis_text) - - # 提取各字段 - summary = "" - topics = [] - insights = [] - - # 尝试按行分割并提取 - lines = [l.strip() for l in analysis_text.split('\n') if l.strip()] - - for line in lines: - line = line.strip() - if not line: - continue - # 跳过编号和问题 - if line.startswith('1.') or line.startswith('2.') or line.startswith('3.'): - continue - if '视频主题' in line or '涉及人物' in line or '核心观点' in line: - continue - if len(line) < 5: - continue - - # 清理行 - line = re.sub(r'^[\d.、\s]+', '', line) - line = line.strip('：:，,。. ') - - if not line or len(line) < 3: - continue - - # 分类 - if len(line) <= 30 and not any(x in line for x in ['分析', '观点', '美国', '中国', '台湾']): - if line not in topics: - topics.append(line) - elif len(line) <= 50: - if line not in insights: - insights.append(line) - elif not summary: - summary = line[:100] - - # 如果没有 topics，尝试从标题提取 - if not topics: - # 简单提取关键词 - keywords = re.findall(r'[^，。、\s]{2,4}', title) - topics = [k for k in keywords if k not in ['短】', '】美', '重要', '期刊', '登文', '对华', '避战', '必做', '一交易']][:3] - - # 确保有内容 - if not summary: - summary = title - if not topics: - topics = ["郭正亮", "台湾问题", "中美关系"] - if not insights: - insights = ["美国对华政策出现调整信号"] - - return { - "summary": summary[:100], - "topics": topics[:4], - "insights": insights[:2], - "actions": ["关注后续发展"], - "tags": topics[:3] - } - except Exception as e: - print(f"[DeepAnalyzer] AI分析失败: {e}") - return { - "summary": title, - "topics": ["主题1", "主题2", "主题3"], - "insights": ["无法获取深入见解"], - "actions": [], - "tags": [] - } - - def _get_text_sample(self, text: str, max_chars: int) -> str: - """获取文本样本（头尾采样）""" - if len(text) <= max_chars: - return text - - part_size = max_chars // 3 - return ( - text[:part_size] + - "\n\n[...省略中间部分...]\n\n" + - 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 - seg_duration = duration // segments - - timeline = [] - for i in range(segments): - start = i * seg_duration - end = min((i + 1) * seg_duration, duration) - timeline.append({ - 'index': i + 1, - 'start_time': self._format_time(start), - 'end_time': self._format_time(end), - 'title': f"段落 {i+1}", - 'summary': f"从{self._format_time(start)}到{self._format_time(end)}的内容" - }) - return timeline - - def _generate_detailed_timeline(self, title: str, duration: int) -> List[Dict]: - """生成详细时间轴""" - return self._generate_simple_timeline_from_duration(title, duration) - - def _build_timeline_from_chunks(self, chunk_summaries: List[Dict]) -> List[Dict]: - """从分段摘要构建时间轴""" - timeline = [] - for s in chunk_summaries: - timeline.append({ - 'index': s['index'] + 1, - 'start_time': s['start_time'], - 'end_time': s['end_time'], - 'title': f"第{s['index']+1}部分", - 'summary': s['summary'][:200] + "..." if len(s['summary']) > 200 else s['summary'] - }) - return timeline - - def _format_time(self, seconds: float) -> str: - """格式化秒数为时间字符串""" - seconds = int(seconds) # 确保是整数 - h = seconds // 3600 - m = (seconds % 3600) // 60 - s = seconds % 60 - if h > 0: - return f"{h}:{m:02d}:{s:02d}" - return f"{m}:{s:02d}" - - -# ============== 模板系统 ============== - -class TemplateLoader: - """模板加载器""" - - TEMPLATES = { - "video": """# {{title}} - -## 摘要 -{{summary}} - -## 主要主题 -{% for topic in topics %} -- {{topic}} -{% endfor %} - -## 关键见解 -{% for insight in insights %} -- {{insight}} -{% endfor %} - -## 时间轴 -{% for entry in timeline %} -### {{entry.index}}. {{entry.start_time}} - {{entry.end_time}} -{{entry.title}}: {{entry.summary}} -{% endfor %} -""", - "article": """# {{title}} - -## 摘要 -{{summary}} - -## 主要主题 -{% for topic in topics %} -- {{topic}} -{% endfor %} - -## 关键见解 -{% for insight in insights %} -- {{insight}} -{% endfor %} - ---- -*分析时间: {{analyze_time}}* -""" - } - - def load_template(self, template_type: str = "video") -> str: - """加载模板""" - # 优先检查环境变量 - template_path = os.environ.get(f"TEMPLATE_{template_type.upper()}", "") - if template_path and os.path.exists(template_path): - with open(template_path, 'r', encoding='utf-8') as f: - return f.read() - - return self.TEMPLATES.get(template_type, self.TEMPLATES["video"]) - - -class TemplateRenderer: - """简单模板渲染器""" - - def render(self, template: str, variables: Dict[str, Any]) -> str: - """渲染模板""" - result = template - - # 替换变量 - for key, value in variables.items(): - placeholder = f"{{{{{key}}}}}" - result = result.replace(placeholder, str(value)) - - # 渲染 for 循环 (简化版本) - # 匹配 {% for item in list %} ... {% endfor %} - pattern = r'\{%\s*for\s+(\w+)\s+in\s+(\w+)\s*%\}(.*?)\{%\s*endfor\s*%\}' - - def replace_loop(match): - item_name = match.group(1) - list_name = match.group(2) - content = match.group(3) - - items = variables.get(list_name, []) - rendered = [] - for item in items: - if isinstance(item, dict): - # 字典项 - item_content = content - for k, v in item.items(): - item_content = item_content.replace(f"{{{{{item_name}.{k}}}}}", str(v)) - rendered.append(item_content) - else: - # 简单项 - rendered.append(content.replace(f"{{{{{item_name}}}}}", str(item))) - - return '\n'.join(rendered) - - result = re.sub(pattern, replace_loop, result, flags=re.DOTALL) - - # 移除空的 for 循环块 - result = re.sub(r'\{%\s*for\s+\w+\s+in\s+\w+\s*%\}\s*\{%\s*endfor\s*%\}', '', result) - - return result - - -# ============== 工具函数 ============== - -def load_json(file_path: str) -> Dict: - """加载 JSON 文件""" - if not file_path or not os.path.exists(file_path): - return {} - - with open(file_path, 'r', encoding='utf-8') as f: - return json.load(f) - - -def load_srt(file_path: str) -> Dict: - """ - 加载并解析 SRT 字幕文件 - - 返回格式: - { - "text": "纯文本内容", - "segments": [ - {"start": "00:00:00", "end": "00:00:10", "text": "字幕文本"} - ] - } - """ - if not file_path or not os.path.exists(file_path): - return {} - - with open(file_path, 'r', encoding='utf-8') as f: - content = f.read() - - # 解析 SRT 格式 - # 格式: 序号 -> 时间 -> 文本 -> 空行 - blocks = re.split(r'\n\s*\n', content.strip()) - segments = [] - - for block in blocks: - lines = block.strip().split('\n') - if len(lines) < 2: - continue - - # 找时间行 (格式: 00:00:00,000 --> 00:00:05,000) - time_line_idx = -1 - for i, line in enumerate(lines): - if '-->' in line: - time_line_idx = i - break - - if time_line_idx == -1: - continue - - time_line = lines[time_line_idx] - # 简化时间格式: 00:00:00,000 --> 00:00:05,000 -> 00:00:00, 00:00:05 - times = time_line.split('-->') - if len(times) != 2: - continue - - # 转换为 HH:MM:SS 格式 - start_raw = times[0].strip().replace(',', '.') - end_raw = times[1].strip().replace(',', '.') - - start_time = _simplify_srt_time(start_raw) - end_time = _simplify_srt_time(end_raw) - - # 文本行 - text_lines = lines[time_line_idx + 1:] - text = ' '.join(line.strip() for line in text_lines) - - if text: - segments.append({ - "start": start_time, - "end": end_time, - "text": text - }) - - # 生成纯文本（每句前面加时间戳） - text_lines = [] - for seg in segments: - text_lines.append(f"[{seg['start']}] {seg['text']}") - - return { - "text": '\n'.join(text_lines), - "segments": segments - } - - -def _simplify_srt_time(time_str: str) -> str: - """ - 将 SRT 时间简化为 HH:MM:SS 格式 - - 输入: 00:00:00.000 或 00:00:00,000 - 输出: 00:00:00 - """ - # 移除毫秒和逗号 - time_str = re.sub(r'[,\.]\d{3}$', '', time_str) - # 确保是 HH:MM:SS 格式 - parts = time_str.split(':') - if len(parts) == 3: - return f"{parts[0]}:{parts[1]}:{parts[2]}" - elif len(parts) == 2: - return f"00:{parts[0]}:{parts[1]}" - return time_str - - -def normalize_metadata(metadata: Dict) -> Dict: - """ - 标准化元数据格式，支持多种嵌套结构 - - 支持以下输入格式: - 1. 扁平结构: {"title": "xxx", "author": "yyy", "duration": 619} - 2. 嵌套结构A: {"data": {"metadata": {"title": "xxx", "author": "yyy", "duration": "10:19"}}} - 3. 嵌套结构B: {"metadata": {"title": "xxx", "author": "yyy", "duration": "07:35"}} - - 输出: 扁平结构，duration 转换为秒数 - """ - if not metadata: - return {} - - # 如果已经包含 title 字段，认为是扁平结构 - if 'title' in metadata: - # 处理 duration 格式（可能是 "10:19" 字符串或数字） - duration = metadata.get('duration', 0) - if isinstance(duration, str) and ':' in duration: - parts = duration.split(':') - try: - seconds = int(parts[0]) * 60 + int(parts[1]) - metadata['duration'] = seconds - except (ValueError, IndexError): - pass - return 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', {}) - - # 提取扁平化字段 - normalized = { - 'title': meta.get('title', metadata.get('title', '未知')), - 'author': meta.get('author', metadata.get('author', '')), - 'bvid': meta.get('bvid', metadata.get('bvid', '')), - 'aid': meta.get('aid', metadata.get('aid', 0)), - 'pubtime': meta.get('pubtime', metadata.get('pubtime', '')), - 'platform': meta.get('platform', 'unknown'), - 'url': meta.get('url', ''), - } - - # 处理 duration - duration = meta.get('duration', 0) - if isinstance(duration, str): - if ':' in duration: - # "10:19" 格式 - parts = duration.split(':') - try: - normalized['duration'] = int(parts[0]) * 60 + int(parts[1]) - except (ValueError, IndexError): - normalized['duration'] = 0 - else: - try: - normalized['duration'] = int(duration) - except ValueError: - normalized['duration'] = 0 - elif isinstance(duration, (int, float)): - normalized['duration'] = int(duration) - else: - normalized['duration'] = 0 - - return normalized - - -def format_duration(seconds: int) -> str: - """格式化时长""" - h = seconds // 3600 - m = (seconds % 3600) // 60 - s = seconds % 60 - if h > 0: - return f"{h}:{m:02d}:{s:02d}" - return f"{m}:{s:02d}" - - -def result_to_dict(result: AnalysisResult) -> Dict: - """将分析结果转换为字典""" - return { - "summary": result.summary, - "main_topics": result.main_topics, - "key_insights": result.key_insights, - "action_items": result.action_items, - "recommended_tags": result.recommended_tags, - "full_analysis": result.full_analysis, - "timeline": result.timeline - } - - -def generate_frontmatter(metadata: Dict, result: AnalysisResult) -> str: - """生成 Markdown frontmatter""" - tags = result.recommended_tags or result.main_topics[:5] - - # 格式化视频发布日期（使用 pubtime） - pubtime = metadata.get('pubtime', '') - if pubtime: - if isinstance(pubtime, (int, float)): - from datetime import datetime - date_str = datetime.fromtimestamp(pubtime).strftime('%Y-%m-%d') - else: - date_str = str(pubtime)[:10] - else: - date_str = "" - - fm = ["---", - f"title: \"{metadata.get('title', '未命名')}\"", - f"date: {date_str}", - f"duration: {format_duration(metadata.get('duration', 0))}", - f"author: {metadata.get('author', '未知')}", - f"bvid: {metadata.get('bvid', '')}", - f"platform: {metadata.get('platform', 'bilibili')}"] - - # 添加视频 URL - url = metadata.get('url', '') - if url: - fm.append(f"url: {url}") - - if tags: - fm.append("tags: [" + ", ".join(f'"{t}"' for t in tags) + "]") - - fm.append("---") - return '\n'.join(fm) - - -# ============== 主程序 ============== - -def main(): - parser = argparse.ArgumentParser( - description="Deep Analyzer - AI 深度内容分析工具", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -示例: - python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md - python deep_analyzer.py --metadata meta.json --content content.json --type video --api-key xxx - """ - ) - - parser.add_argument("--metadata", "-m", required=True, - help="元数据 JSON 文件路径") - parser.add_argument("--content", "-c", required=True, - help="内容 JSON 文件路径") - parser.add_argument("--output", "-o", required=True, - help="输出 Markdown 文件路径") - parser.add_argument("--type", "-t", default="video", choices=["video", "article"], - help="内容类型 (默认: video)") - parser.add_argument("--api-key", "-k", default=None, - help="AI API Key (或设置 AI_API_KEY 环境变量)") - parser.add_argument("--provider", "-p", default=None, - choices=["openai", "zhipu", "minimax"], - help="AI 提供商") - parser.add_argument("--model", default=None, - help="AI 模型 (默认: gpt-4)") - parser.add_argument("--base-url", "-u", default=None, - help="API Base URL (或设置 AI_BASE_URL 环境变量)") - parser.add_argument("--no-frontmatter", action="store_true", - help="不生成 frontmatter") - - args = parser.parse_args() - - # 加载数据 - print(f"[Main] 加载元数据: {args.metadata}") - raw_metadata = load_json(args.metadata) - metadata = normalize_metadata(raw_metadata) - - print(f"[Main] 加载内容: {args.content}") - # 支持 SRT 和 JSON 格式 - content_path = Path(args.content) - if content_path.suffix.lower() == '.srt': - print(f"[Main] 检测到 SRT 格式") - content = load_srt(args.content) - else: - content = load_json(args.content) - - # 初始化 - ai_client = SimpleAIClient( - api_key=args.api_key or os.environ.get("AI_API_KEY", ""), - provider=args.provider or os.environ.get("AI_PROVIDER", ""), - model=args.model or os.environ.get("AI_MODEL", "gpt-4"), - base_url=args.base_url or os.environ.get("AI_BASE_URL", "") - ) - analyzer = DeepAnalyzer(ai_client) - - # 执行分析 - print(f"[Main] 开始分析...") - result = analyzer.analyze(metadata, content, args.type) - - # 渲染模板 - template_loader = TemplateLoader() - renderer = TemplateRenderer() - - template = template_loader.load_template(args.type) - - variables = { - "title": metadata.get('title', '未命名'), - "date": datetime.now().strftime('%Y-%m-%d'), - "duration": format_duration(metadata.get('duration', 0)), - "author": metadata.get('author', '未知'), - "source": metadata.get('source', ''), - "analyze_time": datetime.now().strftime('%Y-%m-%d %H:%M'), - "summary": result.summary, - "topics": result.main_topics, - "insights": result.key_insights, - "actions": result.action_items, - "tags": result.recommended_tags, - "tags_string": ', '.join(f'"{t}"' for t in (result.recommended_tags or [])), - "timeline": result.timeline, - # 扩展元数据 - "bvid": metadata.get('bvid', ''), - "cvid": metadata.get('cvid', ''), - "mid": metadata.get('mid', ''), - "platform": metadata.get('platform', 'bilibili'), - "source_file": metadata.get('source_file', ''), - "aid": str(metadata.get('aid', '')), - } - - markdown_content = renderer.render(template, variables) - - # 添加 frontmatter - if not args.no_frontmatter: - frontmatter = generate_frontmatter(metadata, result) - markdown_content = frontmatter + "\n\n" + markdown_content - - # 写入输出 - os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) - with open(args.output, 'w', encoding='utf-8') as f: - f.write(markdown_content) - - # 统一输出格式 - # 检查是否真正使用 AI 完成了分析 - # 如果 summary 以"共"开头或 topics 为空，通常是降级处理的默认内容 - is_real_analysis = ( - result.summary and - not result.summary.startswith("共") and - len(result.main_topics) > 0 and - len(result.key_insights) > 0 and - result.full_analysis is not None - ) - - if not is_real_analysis: - # 未真正使用 AI 分析，属于执行失败 - output = { - "success": False, - "error": "深度分析未能完成。可能是 AI 调用失败或返回内容不完整。", - "message": "执行失败：未使用正确的 AI 完整分析字幕内容", - "stats": { - "summary_length": len(result.summary) if result.summary else 0, - "topics_count": len(result.main_topics), - "insights_count": len(result.key_insights), - "is_real_analysis": False - } - } - print(json.dumps(output, ensure_ascii=False, indent=2)) - sys.exit(1) - - output = { - "success": True, - "filename": Path(args.output).name, - "filepath": str(Path(args.output).resolve()), - "message": "深度分析完成", - "stats": { - "summary_length": len(result.summary), - "topics_count": len(result.main_topics), - "insights_count": len(result.key_insights) - } - } - print(json.dumps(output, ensure_ascii=False, indent=2)) - - -if __name__ == "__main__": - try: - main() - except Exception as e: - print(json.dumps({ - "success": False, - "error": str(e) - }, ensure_ascii=False, indent=2)) - sys.exit(1)