OpenClaw Agent
978 lines
33 KiB
Python
978 lines
33 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 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 = "mock", model: str = "gpt-4", base_url: str = ""):
|
||
self.api_key = api_key or os.environ.get("AI_API_KEY", "")
|
||
self.provider = provider
|
||
self.model = model
|
||
self.base_url = base_url or os.environ.get("AI_BASE_URL", "")
|
||
|
||
def generate(self, prompt: str, max_tokens: int = 1000) -> str:
|
||
"""生成文本"""
|
||
if not self.api_key or self.provider == "mock":
|
||
return self._mock_generate(prompt, max_tokens)
|
||
|
||
try:
|
||
if self.provider == "openai":
|
||
return self._call_openai(prompt, max_tokens)
|
||
elif self.provider == "zhipu":
|
||
return self._call_zhipu(prompt, max_tokens)
|
||
elif self.provider == "minimax":
|
||
return self._call_minimax(prompt, max_tokens)
|
||
else:
|
||
return self._mock_generate(prompt, max_tokens)
|
||
except Exception as e:
|
||
print(f"[AI Client] 调用失败: {e}")
|
||
return self._mock_generate(prompt, max_tokens)
|
||
|
||
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) -> 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=30)
|
||
response.raise_for_status()
|
||
return response.json()['choices'][0]['message']['content']
|
||
|
||
def _call_zhipu(self, prompt: str, max_tokens: int) -> str:
|
||
"""调用智谱 AI API"""
|
||
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=30)
|
||
response.raise_for_status()
|
||
return response.json()['choices'][0]['message']['content']
|
||
|
||
def _call_minimax(self, prompt: str, max_tokens: int) -> 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=30)
|
||
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)
|
||
|
||
def _mock_generate(self, prompt: str, max_tokens: int) -> str:
|
||
"""模拟 AI 生成"""
|
||
prompt_lower = prompt.lower()
|
||
|
||
if 'json' in prompt_lower or '请返回' in prompt:
|
||
# 返回模拟 JSON 响应
|
||
if '主题' in prompt or 'topic' in prompt_lower:
|
||
return json.dumps({
|
||
'summary': '这是一个模拟的AI分析结果，包含内容的主要主题和关键见解。',
|
||
'topics': ['主题一', '主题二', '主题三', '主题四'],
|
||
'insights': [
|
||
'AI正在深刻改变内容消费方式',
|
||
'个性化推荐提升了用户体验',
|
||
'技术发展带来新的可能性'
|
||
],
|
||
'actions': ['深入研究相关技术', '关注行业动态'],
|
||
'full_analysis': '详细分析内容基于模拟AI生成。'
|
||
}, ensure_ascii=False)
|
||
if '标签' in prompt or 'tag' in prompt_lower:
|
||
return json.dumps({'tags': ['技术', 'AI', '视频分析', '深度学习']}, ensure_ascii=False)
|
||
return json.dumps({'result': '模拟响应'}, ensure_ascii=False)
|
||
|
||
if '摘要' in prompt or 'summary' in prompt_lower:
|
||
return '这是一段关于内容深度分析的重要讨论，涵盖了多个关键主题和见解。'
|
||
|
||
return '模拟AI响应内容'
|
||
|
||
def _parse_json_from_text(self, text: str) -> Dict[str, Any]:
|
||
"""从文本中解析 JSON"""
|
||
json_match = re.search(r'\{[\s\S]*\}', text)
|
||
if json_match:
|
||
try:
|
||
return json.loads(json_match.group())
|
||
except:
|
||
pass
|
||
return {'raw_response': text}
|
||
|
||
|
||
# ============== 深度分析器 ==============
|
||
|
||
class 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 分析结果
|
||
"""
|
||
title = metadata.get('title', '未知')
|
||
duration = metadata.get('duration', 0)
|
||
subtitle = content.get('subtitle', '')
|
||
description = content.get('description', '')
|
||
|
||
print(f"[DeepAnalyzer] 开始分析: {title}")
|
||
print(f"[DeepAnalyzer] 类型: {content_type}, 时长: {duration}秒")
|
||
print(f"[DeepAnalyzer] 字幕存在: {bool(subtitle and subtitle.strip())}")
|
||
|
||
has_subtitle = bool(subtitle and subtitle.strip())
|
||
|
||
if has_subtitle:
|
||
if duration < self.SHORT_VIDEO_THRESHOLD:
|
||
print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)")
|
||
result = self._analyze_short_with_subtitle(title, subtitle, duration)
|
||
else:
|
||
print(f"[DeepAnalyzer] 使用详细时间轴分析策略 (时长 >= 10分钟)")
|
||
result = self._analyze_long_with_subtitle(title, subtitle, duration, content_type)
|
||
else:
|
||
print(f"[DeepAnalyzer] 降级处理（无字幕）")
|
||
result = self._analyze_fallback(title, description, metadata)
|
||
|
||
return result
|
||
|
||
def _analyze_short_with_subtitle(self, title: str, subtitle: str, duration: int) -> AnalysisResult:
|
||
"""有字幕的短视频轻量分析"""
|
||
result = AnalysisResult()
|
||
text_sample = self._get_text_sample(subtitle, max_chars=5000)
|
||
|
||
analysis = self._analyze_with_ai_short_text(text_sample, title, duration)
|
||
result.summary = analysis.get('summary', title)
|
||
result.main_topics = analysis.get('topics', [])
|
||
result.key_insights = analysis.get('insights', [])
|
||
result.action_items = analysis.get('actions', [])
|
||
result.timeline = self._generate_simple_timeline_from_duration(title, duration)
|
||
|
||
return result
|
||
|
||
def _analyze_long_with_subtitle(self, title: str, subtitle: str, duration: int,
|
||
content_type: str) -> AnalysisResult:
|
||
"""有字幕的长视频详细分析"""
|
||
result = AnalysisResult()
|
||
|
||
if duration >= self.SHORT_VIDEO_THRESHOLD and content_type == "video":
|
||
# 10分钟以上使用分段分析
|
||
result = self._analyze_in_chunks(title, subtitle, duration)
|
||
else:
|
||
# 直接分析
|
||
text_sample = self._get_text_sample(subtitle, max_chars=5000)
|
||
analysis = self._analyze_with_ai_short_text(text_sample, title, duration)
|
||
result.summary = analysis.get('summary', title)
|
||
result.main_topics = analysis.get('topics', [])
|
||
result.key_insights = analysis.get('insights', [])
|
||
result.action_items = analysis.get('actions', [])
|
||
result.full_analysis = analysis.get('full_analysis', '')
|
||
result.timeline = self._generate_detailed_timeline(title, duration)
|
||
|
||
return result
|
||
|
||
def _analyze_in_chunks(self, title: str, subtitle: str, duration: int) -> 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)
|
||
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 _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) -> str:
|
||
"""分析单个分片"""
|
||
text_sample = self._get_text_sample(chunk_text, max_chars=3000)
|
||
|
||
prompt = f"""请对以下视频片段进行详细分析（限制在800字以内）：
|
||
|
||
片段内容：
|
||
{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
|
||
])
|
||
|
||
prompt = f"""基于以下视频各段分析，生成完整详细的分析报告：
|
||
|
||
标题：{title}
|
||
|
||
各段内容：
|
||
{summaries_text}
|
||
|
||
请返回JSON格式：
|
||
{{
|
||
"summary": "整体摘要（300字以内）",
|
||
"topics": ["主题1", "主题2", "主题3", "主题4", "主题5"],
|
||
"insights": ["见解1", "见解2", "见解3", "见解4"],
|
||
"actions": ["建议1", "建议2", "建议3"],
|
||
"full_analysis": "详细分析（1500字以内）"
|
||
}}"""
|
||
|
||
try:
|
||
analysis = self.ai_client.generate_json(prompt)
|
||
result = AnalysisResult()
|
||
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', summaries_text)
|
||
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 _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 短视频分析"""
|
||
prompt = f"""请分析以下视频内容，以JSON格式返回：
|
||
|
||
标题：{title}
|
||
时长：{self._format_time(duration)}
|
||
|
||
字幕内容：
|
||
{text[:3000]}
|
||
|
||
请返回以下格式JSON：
|
||
{{
|
||
"summary": "整体摘要（200字以内）",
|
||
"topics": ["主题1", "主题2", "主题3", "主题4"],
|
||
"insights": ["见解1", "见解2", "见解3"],
|
||
"actions": ["建议1", "建议2"],
|
||
"tags": ["标签1", "标签2", "标签3", "标签4", "标签5"]
|
||
}}"""
|
||
|
||
try:
|
||
return self.ai_client.generate_json(prompt)
|
||
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_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: 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}"
|
||
|
||
|
||
# ============== 模板系统 ==============
|
||
|
||
class TemplateLoader:
|
||
"""模板加载器"""
|
||
|
||
TEMPLATES = {
|
||
"video": """# {{title}}
|
||
|
||
---
|
||
title: "{{title}}"
|
||
date: {{date}}
|
||
duration: {{duration}}
|
||
author: {{author}}
|
||
source: {{source}}
|
||
tags: [{{tags_string}}]
|
||
bvid: {{bvid}}
|
||
platform: {{platform}}
|
||
---
|
||
|
||
## 基本信息
|
||
- **时长**: {{duration}}
|
||
- **作者**: {{author}}
|
||
- **来源**: {{source}}
|
||
- **分析时间**: {{analyze_time}}
|
||
|
||
## 摘要
|
||
{{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 %}
|
||
|
||
{% if tags %}
|
||
## 标签
|
||
{% for tag in tags %}
|
||
- {{tag}}
|
||
{% endfor %}
|
||
{% endif %}
|
||
|
||
---
|
||
|
||
*元数据*
|
||
- **平台**: {{platform}}
|
||
- **BV号**: {{bvid}}
|
||
- **原始文件**: {{source_file}}
|
||
""",
|
||
"article": """# {{title}}
|
||
|
||
---
|
||
title: "{{title}}"
|
||
date: {{date}}
|
||
author: {{author}}
|
||
source: {{source}}
|
||
tags: [{{tags_string}}]
|
||
cvid: {{cvid}}
|
||
platform: {{platform}}
|
||
---
|
||
|
||
## 基本信息
|
||
- **作者**: {{author}}
|
||
- **来源**: {{source}}
|
||
- **分析时间**: {{analyze_time}}
|
||
|
||
## 摘要
|
||
{{summary}}
|
||
|
||
## 主要主题
|
||
{% for topic in topics %}
|
||
- {{topic}}
|
||
{% endfor %}
|
||
|
||
## 关键见解
|
||
{% for insight in insights %}
|
||
- {{insight}}
|
||
{% endfor %}
|
||
|
||
## 建议行动
|
||
{% for action in actions %}
|
||
- [ ] {{action}}
|
||
{% endfor %}
|
||
|
||
{% if tags %}
|
||
## 标签
|
||
{% for tag in tags %}
|
||
- {{tag}}
|
||
{% endfor %}
|
||
{% endif %}
|
||
|
||
---
|
||
|
||
*元数据*
|
||
- **平台**: {{platform}}
|
||
- **CV号**: {{cvid}}
|
||
- **原始文件**: {{source_file}}
|
||
"""
|
||
}
|
||
|
||
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 normalize_metadata(metadata: Dict) -> Dict:
|
||
"""
|
||
标准化元数据格式，支持嵌套和平面两种结构
|
||
|
||
支持以下输入格式:
|
||
1. 扁平结构: {"title": "xxx", "author": "yyy", "duration": 619}
|
||
2. 嵌套结构: {"data": {"metadata": {"title": "xxx", "author": "yyy", "duration": "10:19"}}}
|
||
|
||
输出: 扁平结构，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
|
||
|
||
# 如果是嵌套结构，提取 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]
|
||
|
||
fm = ["---", "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', '未知')}"]
|
||
|
||
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="mock",
|
||
choices=["openai", "zhipu", "minimax", "mock"],
|
||
help="AI 提供商 (默认: mock)")
|
||
parser.add_argument("--model", default="gpt-4",
|
||
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}")
|
||
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", "mock"),
|
||
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)
|