starmen1265
bdc84ec75c
- generate() 每次调用时重新读取环境变量 - AI 调用方法接受 api_key, model, base_url 参数 - 解决 exec 工具中环境变量未传递的问题
1331 lines
47 KiB
Python
1331 lines
47 KiB
Python
#!/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 客户端 ==============
|
||
|
||
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 # 最大重试次数
|
||
|
||
def generate(self, prompt: str, max_tokens: int = 1000) -> 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 未设置，请配置环境变量")
|
||
|
||
last_error = None
|
||
for attempt in range(self.max_retries):
|
||
try:
|
||
if provider == "openai":
|
||
return self._call_openai(prompt, max_tokens, api_key, model, base_url)
|
||
elif provider == "zhipu":
|
||
return self._call_zhipu(prompt, max_tokens, api_key, model, base_url)
|
||
elif provider == "minimax":
|
||
return self._call_minimax(prompt, max_tokens, api_key, model, base_url)
|
||
else:
|
||
raise ValueError(f"不支持的 AI 提供商: {provider}")
|
||
except Exception as e:
|
||
last_error = e
|
||
if attempt < self.max_retries - 1:
|
||
print(f"[AI Client] 第 {attempt + 1} 次调用失败，{self.max_retries - attempt - 1} 次重试剩余: {e}")
|
||
continue
|
||
else:
|
||
print(f"[AI Client] 调用失败，已重试 {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=120)
|
||
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=120)
|
||
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=120)
|
||
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") -> AnalysisResult:
|
||
"""
|
||
分析内容
|
||
|
||
Args:
|
||
metadata: 元数据字典
|
||
content: 内容数据字典（包含 subtitle, description 等）
|
||
content_type: 内容类型
|
||
|
||
Returns:
|
||
AnalysisResult 分析结果
|
||
"""
|
||
# 先标准化元数据
|
||
metadata = normalize_metadata(metadata)
|
||
|
||
title = metadata.get('title', '未知')
|
||
duration = metadata.get('duration', 0)
|
||
|
||
# 支持多种内容格式
|
||
# 1. Whisper 格式: {"text": "...", "segments": [...]}
|
||
# 2. 字幕格式: {"subtitle": "..."}
|
||
# 3. 描述格式: {"description": "..."}
|
||
text = content.get('text', '') or content.get('subtitle', '') or content.get('description', '')
|
||
segments = content.get('segments', [])
|
||
|
||
has_content = bool(text and text.strip())
|
||
has_segments = bool(segments)
|
||
|
||
print(f"[DeepAnalyzer] 开始分析: {title}")
|
||
print(f"[DeepAnalyzer] 类型: {content_type}, 时长: {duration}秒")
|
||
print(f"[DeepAnalyzer] 文本存在: {has_content}, 分段数: {len(segments)}")
|
||
print(f"[DeepAnalyzer] 文本长度: {len(text) if text else 0} 字符")
|
||
|
||
# 强制分块阈值：无论视频长短，超过3000字都用分块分析
|
||
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_in_chunks_with_segments(title, text, duration, segments, metadata)
|
||
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 _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:
|
||
return self.ai_client.generate(prompt, max_tokens=2000)
|
||
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 action in actions %}
|
||
- [ ] {{action}}
|
||
{% endfor %}
|
||
|
||
{% if timeline %}
|
||
## 时间轴
|
||
{% for entry in timeline %}
|
||
### {{entry.index}}. {{entry.start_time}} - {{entry.end_time}}
|
||
{{entry.title}}: {{entry.summary}}
|
||
{% endfor %}
|
||
{% endif %}
|
||
|
||
---
|
||
*分析时间: {{analyze_time}}*
|
||
""",
|
||
"article": """# {{title}}
|
||
|
||
## 摘要
|
||
{{summary}}
|
||
|
||
## 主要主题
|
||
{% for topic in topics %}
|
||
- {{topic}}
|
||
{% endfor %}
|
||
|
||
## 关键见解
|
||
{% for insight in insights %}
|
||
- {{insight}}
|
||
{% endfor %}
|
||
|
||
## 建议行动
|
||
{% for action in actions %}
|
||
- [ ] {{action}}
|
||
{% 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]
|
||
analyze_time = datetime.now().strftime('%Y-%m-%d %H:%M')
|
||
|
||
fm = ["---",
|
||
f"title: \"{metadata.get('title', '未命名')}\"",
|
||
f"date: {datetime.now().strftime('%Y-%m-%d')}",
|
||
f"duration: {format_duration(metadata.get('duration', 0))}",
|
||
f"author: {metadata.get('author', '未知')}",
|
||
f"analyze_time: {analyze_time}",
|
||
f"bvid: {metadata.get('bvid', '')}",
|
||
f"platform: {metadata.get('platform', 'bilibili')}"]
|
||
|
||
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)
|
||
|
||
# 统一输出格式
|
||
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)
|