diff --git a/MEMORY.md b/MEMORY.md index 3d663ed..fea0f31 100644 --- a/MEMORY.md +++ b/MEMORY.md @@ -35,6 +35,12 @@ 2. 记录修改内容到 SkillBug.md 3. 考虑使用 monkey patch 或环境变量等非侵入式方案 +## 知识库工作流（2026-02-13） +- **deep_analyzer.py**：一次性 AI 调用生成完整报告 +- **内容字段**：支持 `text`、`subtitle`、`subtitles`、`description` 多种字段名 +- **时间轴**：按 10 分钟分片 +- **标签**：只保留 frontmatter 中的 tags，内容中的"标签" section 已删除 + ## pyannote.audio 兼容性问题（2026-02-13） - **问题**: pyannote 4.0.4 与 torch 2.10+ 存在兼容性问题 - **症状**: diff --git a/memory/2026-02-13.md b/memory/2026-02-13.md index b76c7e6..8887f74 100644 --- a/memory/2026-02-13.md +++ b/memory/2026-02-13.md @@ -1,88 +1,54 @@ -# 2026-02-13 +# 2026-02-13 工作日志 -## kb-deep-analyzer 技能改进 +## 完成事项 -### 批处理模式 -- 改为分批并发处理（3+3+3+1），避免同时启动所有AI请求 -- `analyze()` 方法支持 `on_batch_complete` 回调 -- `batch_size` 参数控制每批数量（默认3） +### deep_analyzer.py 重构 +- 改为一次性 AI 调用生成完整报告 +- 修复内容读取：支持 `text`、`subtitle`、`subtitles`、`description` 多种字段 +- 时间轴分片策略：按 10 分钟分片 +- 删除内容中的"标签" section，只保留 frontmatter 的 tags -### 流式写入功能 -- 新增 `StreamingWriter` 类，支持多线程并发写入文档不同位置 -- `initialize_document()` 预先创建带占位符的文档结构 -- `write_batch()` 每批完成后写入对应位置 +### 报告格式 +``` +--- +title: "xxx" +date: xxx +duration: xxx +author: xxx +bvid: xxx +platform: bilibili +url: xxx +tags: ["tag1", "tag2", ...] +--- -### 文档格式优化 -- `date` 改为视频发布日期（从 pubtime 提取） -- frontmatter 添加 `url` 字段 -- 删除"建议行动"章节 -- 删除 `{% if timeline %}` 和最后的分析时间 -- 新格式更简洁 +## 标题 +[AI 自总结标题] -### 提交记录 -- `94370f2` - feat: 改为批处理模式 -- `c8ddc1e` - feat: 添加流式写入器 -- `518f8e4` - fix: 优化文档格式 -- `c276df2` - feat(kb-workflow): 添加严格失败判断逻辑 +## 摘要 +[200-400字] -## 严格失败判断规则（15:19 添加） +## 主要主题 +- 主题1 +- 主题2 +... -### kb-deep-analyzer 更新 -- 新增判断逻辑：只有 AI 真正完成分析才算成功 -- 判断条件： - - `summary` 不能以"共"开头（降级处理的默认内容） - - `main_topics` 不能为空 - - `key_insights` 不能为空 - - `full_analysis` 必须有值 -- 失败时返回 `success: false` 并退出码 1 +## 关键见解 +- 见解1 +- 见解2 +... -### kb-workflow 更新 -- 新增"严格失败处理原则"章节 -- 明确：深度分析失败 = 整个流程失败 -- 禁止降级处理或使用模板"假装成功" +## 时间轴 +### 第 1 部分 (00:00 - 10:00) +[详细分析] -## 知识库添加：BV1W4f9B6EdC（第二次处理） +### 第 2 部分 (10:00 - 20:00) +[详细分析] +... +``` -### 处理结果 -- **标题**: 爱泼斯坦案惊天动地？美式司法："没事，已经结束了"【逸语道破】 -- **作者**: 沈逸老师 -- **时长**: 32:38 -- **Whisper**: ✅ 完成（8043字符，971段） -- **深度分析**: ❌ 卡住超时 -- **归档**: ❌ 未执行 +## 待办 +- 无 -### 执行记录 -1. 环境变量检查 ✅ -2. 链接解析 ✅ (BV1W4f9B6EdC) -3. 元数据提取 ✅ -4. 视频下载 ✅ -5. Whisper 识别 ✅ (15:29 完成) -6. 深度分析 ❌ (15:29 开始，20分钟后仍卡住，终止) - -### 发现的问题 -1. **参数名不匹配错误** - - 错误: `name 'on_batch_complete' is not defined` - - 修复: 统一使用 `on_segment_complete` 作为内部参数名 - -2. **batch_size 未定义** - - 错误: `name 'batch_size' is not defined` - - 修复: 在 `_analyze_chunks_concurrent()` 签名中添加 `batch_size: int = 3` - -3. **深度分析卡住** - - 症状: 进程运行但无输出，20分钟后仍卡住 - - 原因: 可能是批处理模式逻辑问题或 API 调用卡住 - - 状态: 需要排查 - -## 技能状态 -| 步骤 | 技能 | 状态 | -|------|------|------| -| 链接解析 | kb-link-parser | ✅ | -| 元数据提取 | kb-bilibili-metadata | ✅ | -| 字幕下载 | kb-bilibili-content | ✅ | -| 语音识别 | kb-whisper | ✅ | -| 深度分析 | kb-deep-analyzer | ❌ 卡住 | - -## 待解决问题 -- [ ] 排查深度分析卡住原因 -- [ ] 验证严格失败判断逻辑是否正确工作 -- [ ] 重新测试完整知识库流程 +## 备注 +- AI 模型：glm-4.7 +- 内容长度：约 8000 字符输入，生成详细分析报告 diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 59c62c1..d1524dc 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -1,9 +1,7 @@ #!/usr/bin/env python3 """Deep Analyzer - AI 深度内容分析工具 -支持根据内容时长选择不同分析策略： -- 短视频 (<10分钟): 轻量分析 -- 长视频 (>=10分钟): 详细时间轴分析 +一次性调用 AI 生成完整深度分析报告。 Usage: python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md [--type video] @@ -17,247 +15,50 @@ 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 +from typing import Dict, Any, List +from dataclasses import dataclass, field # ============== 数据模型 ============== -@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分钟 + """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", "gpt-4") self.base_url = base_url or os.environ.get("AI_BASE_URL", "") - self.max_retries = 3 # 最大重试次数 - self.max_concurrent = 3 # 最大并发数 + self.max_retries = 3 - def generate(self, prompt: str, max_tokens: int = 1000, task_info: str = None) -> str: - """生成文本，支持重试""" - # 每次调用时重新读取环境变量，解决 exec 传递问题 + 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 未设置，请配置环境变量") + 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...") - + print(f"[AI Client] 开始调用 AI...") + if provider == "openai": result = self._call_openai(prompt, max_tokens, api_key, model, base_url) elif provider == "zhipu": @@ -266,128 +67,53 @@ class SimpleAIClient: 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}秒") + print(f"[AI Client] 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}") + print(f"[AI Client] 第 {attempt + 1} 次调用失败，{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""" + def _call_openai(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: 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}] - } - + 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() - 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""" + def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: import requests - base_url = self.base_url or "https://open.bigmodel.cn/api/coding/paas/v4" + base_url = 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 - } - + 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() - 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""" + def _call_minimax(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: 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}] - } - + 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() - response.raise_for_status() data = response.json() for item in data.get('content', []): if item.get('type') == 'text': @@ -395,842 +121,241 @@ class SimpleAIClient: 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) + def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any]) -> AnalysisResult: + """一次性调用 AI 生成完整分析报告""" + + # 标准化元数据 + 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', '') - # 支持多种内容格式 - text = content.get('text', '') or content.get('subtitle', '') or content.get('description', '') - segments = content.get('segments', []) + # 获取内容 - 支持多种字段名 + text = content.get('text', '') or content.get('subtitle', '') or content.get('subtitles', '') or content.get('description', '') - 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: - """ - 使用实际字幕时间戳进行分块分析 + print(f"[DeepAnalyzer] 时长: {duration}秒, 内容长度: {len(text)} 字符") - 当有 SRT 格式的 segments 时，使用实际时间戳来分块 - 否则退化为字符位置估算 - """ - result = AnalysisResult() - - CHUNK_MAX_CHARS = 3000 - total_chars = len(text) + # 计算分片数（按10分钟分片） + num_chunks = max(1, (duration + 599) // 600) # 向上取整到10分钟 + print(f"[DeepAnalyzer] 将分 {num_chunks} 段分析") - # 检查是否有有效的 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) + # 构建 prompt + prompt = self._build_prompt( + title=title, + duration=duration, + duration_str=self._format_duration(duration), + author=author, + pubtime=pubtime, + url=url, + bvid=bvid, + text=text, + num_chunks=num_chunks ) - 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" + # 调用 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) + + return result + + def _build_prompt(self, title: str, duration: int, duration_str: str, author: str, + pubtime: str, url: str, bvid: str, text: str, num_chunks: int) -> str: + """构建分析 prompt""" + + # 截取适当长度的文本（20-30K 字符） + max_chars = 30000 + if len(text) > max_chars: + text_sample = text[:max_chars] + "\n\n[...]" + else: + text_sample = text + + prompt = f"""你是一个专业的视频内容分析师。请根据以下信息生成深度分析报告： - prompt = f"""请对以下视频片段进行详细分析（限制在800字以内）： +【视频信息】 +标题：{title} +作者：{author} +时长：{duration_str}（{duration}秒） +发布日期：{pubtime} +平台：bilibili +URL：{url} -{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 - ]) +## 摘要 +[200-400字] - # 使用纯文本提示，不再要求 JSON 格式 - prompt = f"""基于以下视频各段分析，生成完整详细的分析报告： +## 主要主题 +[从内容中提取不超过10个主题，每个不超过10字] +- 主题1 +- 主题2 +... -标题：{title} +## 关键见解 +[若干条深刻见解] +- 见解1 +- 见解2 +... -各段内容： -{summaries_text} +## 标签 +[从内容中提取3-5个精炼标签，每个不超过10字] +- 标签1 +- 标签2 +... -请用中文详细回复，包含： -1. 整体摘要（300字以内） -2. 5个核心主题 -3. 4个关键见解 -4. 3条建议 -5. 详细分析（1500字以内）""" +## 时间轴 +### 第 1 部分 (00:00 - 10:00) +[详细分析，至少500字] - 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] - ) +### 第 2 部分 (10:00 - 20:00) +[详细分析，至少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: - """无字幕降级处理""" + return prompt + + def _parse_analysis(self, analysis_text: str, original_text: str, duration: int) -> AnalysisResult: + """解析 AI 返回的分析结果""" 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 = ["建议补充字幕以便深度分析"] - + result.full_analysis = analysis_text + + # 提取各部分（使用正则） + lines = analysis_text.split('\n') + + # 清理 Markdown 符号的辅助函数 + def clean_markdown(text): + text = re.sub(r'\*\*([^*]+)\*\*', r'\1', text) + text = re.sub(r'\*([^*]+)\*', r'\1', text) + text = re.sub(r'#+\s*', '', text) + return text.strip() + + # 简单解析：按 ## 标题分割 + 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 _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 = "" + + def _store_section(self, result: AnalysisResult, section_name: str, content: str): + """存储解析到的 section""" + content = content.strip() + if not content: + return + + # 清理 content 中的 Markdown + content = re.sub(r'\*\*([^*]+)\*\*', r'\1', content) + content = re.sub(r'\*([^*]+)\*', r'\1', content) + + 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 = [] - - # 尝试按行分割并提取 - 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) + 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 {} - timeline = [] - # 取前10个最重要的分段 - selected_segments = segments[:10] + # 如果已经扁平 + 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 - 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字 - }) + # 嵌套结构 + 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 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) # 确保是整数 + return metadata + + def _format_duration(self, seconds: int) -> str: + """格式化时长""" h = seconds // 3600 m = (seconds % 3600) // 60 s = seconds % 60 @@ -1239,279 +364,147 @@ class DeepAnalyzer: 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 字幕文件 +def generate_frontmatter(metadata: Dict, result: AnalysisResult) -> str: + """生成 Markdown frontmatter""" + # 提取标签 + tags = result.recommended_tags or [] - 返回格式: - { - "text": "纯文本内容", - "segments": [ - {"start": "00:00:00", "end": "00:00:10", "text": "字幕文本"} - ] - } - """ - if not file_path or not os.path.exists(file_path): - return {} + # 格式化日期 + pubtime = metadata.get('pubtime', '') + if pubtime: + date_str = str(pubtime)[:10] + else: + date_str = datetime.now().strftime('%Y-%m-%d') - with open(file_path, 'r', encoding='utf-8') as f: - content = f.read() + fm = ["---", + f"title: \"{metadata.get('title', '未命名')}\"", + f"date: {date_str}", + f"duration: {metadata.get('duration_str', metadata.get('duration', 0))}", + f"author: {metadata.get('author', '未知')}", + f"bvid: {metadata.get('bvid', '')}", + f"platform: {metadata.get('platform', 'bilibili')}"] - # 解析 SRT 格式 - # 格式: 序号 -> 时间 -> 文本 -> 空行 - blocks = re.split(r'\n\s*\n', content.strip()) - segments = [] + # 添加 URL + url = metadata.get('url', '') + if url: + fm.append(f"url: {url}") - 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 - }) + if tags: + fm.append("tags: [" + ", ".join(f'"{t}"' for t in tags) + "]") - # 生成纯文本（每句前面加时间戳） - text_lines = [] - for seg in segments: - text_lines.append(f"[{seg['start']}] {seg['text']}") - - return { - "text": '\n'.join(text_lines), - "segments": segments - } + fm.append("---") + return '\n'.join(fm) -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 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") -def normalize_metadata(metadata: Dict) -> Dict: - """ - 标准化元数据格式，支持多种嵌套结构 + args = parser.parse_args() - 支持以下输入格式: - 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"}} + # 加载数据 + print(f"[Main] 加载元数据: {args.metadata}") + raw_metadata = load_json(args.metadata) - 输出: 扁平结构，duration 转换为秒数 - """ - if not metadata: - return {} + print(f"[Main] 加载内容: {args.content}") + content = load_json(args.content) - # 如果已经包含 title 字段，认为是扁平结构 - if 'title' in metadata: - # 处理 duration 格式（可能是 "10:19" 字符串或数字） + # 标准化元数据 + 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: - seconds = int(parts[0]) * 60 + int(parts[1]) - metadata['duration'] = seconds - except (ValueError, IndexError): + metadata['duration'] = int(parts[0]) * 60 + int(parts[1]) + except: 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" 格式 + 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: - 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 + 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", "gpt-4"), + base_url=args.base_url or os.environ.get("AI_BASE_URL", "") + ) + analyzer = DeepAnalyzer(ai_client) - return normalized + # 执行分析 + print(f"[Main] 开始分析...") + result = analyzer.analyze(metadata, content) + + # 清理 Markdown 符号 + full_analysis = result.full_analysis + full_analysis = re.sub(r'^#+\s*', '', full_analysis, flags=re.MULTILINE) + full_analysis = re.sub(r'\*\*([^*]+)\*\*', r'\1', full_analysis) + full_analysis = re.sub(r'\*([^*]+)\*', r'\1', 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) + + # 写入输出 + 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) + 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: @@ -1524,204 +517,9 @@ def format_duration(seconds: int) -> str: 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)) + print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2)) sys.exit(1) diff --git a/skills/kb-deep-analyzer/deep_analyzer.py.bak b/skills/kb-deep-analyzer/deep_analyzer.py.bak new file mode 100644 index 0000000..59c62c1 --- /dev/null +++ b/skills/kb-deep-analyzer/deep_analyzer.py.bak @@ -0,0 +1,1727 @@ +#!/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)