526 lines
19 KiB
Python
526 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""Deep Analyzer - AI 深度内容分析工具
|
||
|
||
一次性调用 AI 生成完整深度分析报告。
|
||
|
||
Usage:
|
||
python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md [--type video]
|
||
python deep_analyzer.py --help
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, Any, List
|
||
from dataclasses import dataclass, field
|
||
|
||
|
||
# ============== 数据模型 ==============
|
||
|
||
@dataclass
|
||
class AnalysisResult:
|
||
"""分析结果"""
|
||
summary: str = ""
|
||
main_topics: List[str] = field(default_factory=list)
|
||
key_insights: List[str] = field(default_factory=list)
|
||
recommended_tags: List[str] = field(default_factory=list)
|
||
full_analysis: str = ""
|
||
|
||
|
||
# ============== AI 客户端 ==============
|
||
|
||
class SimpleAIClient:
|
||
"""AI 客户端"""
|
||
|
||
def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None):
|
||
self.api_key = api_key or os.environ.get("AI_API_KEY", "")
|
||
self.provider = provider or os.environ.get("AI_PROVIDER", "")
|
||
self.model = model or os.environ.get("AI_MODEL", "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 = 16000) -> str:
|
||
"""生成文本"""
|
||
api_key = os.environ.get("AI_API_KEY", self.api_key)
|
||
provider = os.environ.get("AI_PROVIDER", self.provider)
|
||
model = os.environ.get("AI_MODEL", self.model)
|
||
base_url = os.environ.get("AI_BASE_URL", self.base_url)
|
||
|
||
if not api_key:
|
||
raise ValueError("AI_API_KEY 未设置")
|
||
|
||
for attempt in range(self.max_retries):
|
||
try:
|
||
import time
|
||
start_time = time.time()
|
||
print(f"[AI Client] 开始调用 AI...")
|
||
|
||
if provider == "openai":
|
||
result = self._call_openai(prompt, max_tokens, api_key, model, base_url)
|
||
elif provider == "zhipu":
|
||
result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url)
|
||
elif provider == "minimax":
|
||
result = self._call_minimax(prompt, max_tokens, api_key, model, base_url)
|
||
else:
|
||
raise ValueError(f"不支持的 AI 提供商: {provider}")
|
||
|
||
elapsed = time.time() - start_time
|
||
print(f"[AI Client] AI 调用完成，耗时: {elapsed:.2f}秒")
|
||
return result
|
||
except Exception as e:
|
||
if attempt < self.max_retries - 1:
|
||
print(f"[AI Client] 第 {attempt + 1} 次调用失败，{self.max_retries - attempt - 1} 次重试剩余: {e}")
|
||
continue
|
||
else:
|
||
raise
|
||
|
||
def _call_openai(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
|
||
import requests
|
||
url = f"{base_url}/chat/completions" if base_url else "https://api.openai.com/v1/chat/completions"
|
||
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
|
||
payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]}
|
||
response = requests.post(url, headers=headers, json=payload, timeout=600)
|
||
response.raise_for_status()
|
||
return response.json()['choices'][0]['message']['content']
|
||
|
||
def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
|
||
import requests
|
||
base_url = base_url or "https://open.bigmodel.cn/api/coding/paas/v4"
|
||
url = f"{base_url}/chat/completions"
|
||
headers = {"Authorization": f"Bearer {api_key}"}
|
||
payload = {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens}
|
||
response = requests.post(url, headers=headers, json=payload, timeout=600)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
message = data.get('choices', [{}])[0].get('message', {})
|
||
content = message.get('content', '')
|
||
reasoning = message.get('reasoning_content', '')
|
||
if not content and reasoning:
|
||
sentences = reasoning.strip().split('。')
|
||
useful = [s.strip() for s in sentences if s.strip() and len(s.strip()) > 10]
|
||
if useful:
|
||
return '。'.join(useful[-3:]) + '。'
|
||
return reasoning.strip()[:500]
|
||
return content
|
||
|
||
def _call_minimax(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
|
||
import requests
|
||
url = base_url or "https://api.minimaxi.com/anthropic/v1/messages"
|
||
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
|
||
payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]}
|
||
response = requests.post(url, headers=headers, json=payload, timeout=600)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
for item in data.get('content', []):
|
||
if item.get('type') == 'text':
|
||
return item.get('text', '')
|
||
return str(data)
|
||
|
||
|
||
# ============== 分析器 ==============
|
||
|
||
class DeepAnalyzer:
|
||
"""深度内容分析器"""
|
||
|
||
def __init__(self, ai_client: SimpleAIClient = None):
|
||
self.ai_client = ai_client or SimpleAIClient()
|
||
|
||
def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any]) -> 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('subtitles', '') or content.get('description', '')
|
||
|
||
print(f"[DeepAnalyzer] 开始分析: {title}")
|
||
print(f"[DeepAnalyzer] 时长: {duration}秒, 内容长度: {len(text)} 字符")
|
||
|
||
# 计算分片数（按10分钟分片）
|
||
num_chunks = max(1, (duration + 599) // 600) # 向上取整到10分钟
|
||
print(f"[DeepAnalyzer] 将分 {num_chunks} 段分析")
|
||
|
||
# 构建 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
|
||
)
|
||
|
||
# 调用 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"""你是一个专业的视频内容分析师。请根据以下信息生成深度分析报告：
|
||
|
||
【视频信息】
|
||
标题：{title}
|
||
作者：{author}
|
||
时长：{duration_str}（{duration}秒）
|
||
发布日期：{pubtime}
|
||
平台：bilibili
|
||
URL：{url}
|
||
|
||
【内容】（完整字幕）
|
||
{text_sample}
|
||
|
||
【输出要求】必须使用以下格式：
|
||
|
||
## 标题
|
||
[自己总结的简洁标题，禁止使用原标题]
|
||
|
||
## 摘要
|
||
[200-400字]
|
||
|
||
## 主要主题
|
||
[从内容中提取不超过10个主题，每个不超过10字]
|
||
- 主题1
|
||
- 主题2
|
||
...
|
||
|
||
## 关键见解
|
||
[若干条深刻见解]
|
||
- 见解1
|
||
- 见解2
|
||
...
|
||
|
||
## 标签
|
||
[从内容中提取3-5个精炼标签，每个不超过10字]
|
||
- 标签1
|
||
- 标签2
|
||
...
|
||
|
||
## 时间轴
|
||
### 第 1 部分 (00:00 - 10:00)
|
||
[详细分析，至少500字]
|
||
|
||
### 第 2 部分 (10:00 - 20:00)
|
||
[详细分析，至少500字]
|
||
|
||
...（以此类推）
|
||
|
||
【重要】分析应足够详尽，读者阅读后可了解视频核心内容，无需再看原视频。"""
|
||
|
||
return prompt
|
||
|
||
def _parse_analysis(self, analysis_text: str, original_text: str, duration: int) -> AnalysisResult:
|
||
"""解析 AI 返回的分析结果"""
|
||
result = AnalysisResult()
|
||
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 _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 = []
|
||
for line in content.split('\n'):
|
||
line = line.strip().lstrip('-').strip()
|
||
if line and len(line) > 5:
|
||
insights.append(line)
|
||
result.key_insights = insights
|
||
elif '标签' in section_lower:
|
||
# 提取标签
|
||
tags = []
|
||
for line in content.split('\n'):
|
||
line = line.strip().lstrip('-').strip()
|
||
if line and len(line) <= 10:
|
||
tags.append(line)
|
||
result.recommended_tags = tags[:5]
|
||
|
||
def _normalize_metadata(self, metadata: Dict) -> Dict:
|
||
"""标准化元数据"""
|
||
if not metadata:
|
||
return {}
|
||
|
||
# 如果已经扁平
|
||
if 'title' in metadata:
|
||
duration = metadata.get('duration', 0)
|
||
if isinstance(duration, str) and ':' in duration:
|
||
parts = duration.split(':')
|
||
try:
|
||
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
|
||
except:
|
||
pass
|
||
return metadata
|
||
|
||
# 嵌套结构
|
||
if 'metadata' in metadata:
|
||
meta = metadata['metadata']
|
||
normalized = {}
|
||
for k, v in meta.items():
|
||
normalized[k] = v
|
||
# 处理 duration
|
||
duration = normalized.get('duration', 0)
|
||
if isinstance(duration, str) and ':' in duration:
|
||
parts = duration.split(':')
|
||
try:
|
||
normalized['duration'] = int(parts[0]) * 60 + int(parts[1])
|
||
except:
|
||
pass
|
||
return normalized
|
||
|
||
return metadata
|
||
|
||
def _format_duration(self, seconds: int) -> str:
|
||
"""格式化时长"""
|
||
h = seconds // 3600
|
||
m = (seconds % 3600) // 60
|
||
s = seconds % 60
|
||
if h > 0:
|
||
return f"{h}:{m:02d}:{s:02d}"
|
||
return f"{m}:{s:02d}"
|
||
|
||
|
||
# ============== 工具函数 ==============
|
||
|
||
def load_json(file_path: str) -> Dict:
|
||
"""加载 JSON 文件"""
|
||
if not file_path or not os.path.exists(file_path):
|
||
return {}
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
|
||
|
||
def generate_frontmatter(metadata: Dict, result: AnalysisResult) -> str:
|
||
"""生成 Markdown frontmatter"""
|
||
# 提取标签
|
||
tags = result.recommended_tags or []
|
||
|
||
# 格式化日期
|
||
pubtime = metadata.get('pubtime', '')
|
||
if pubtime:
|
||
date_str = str(pubtime)[:10]
|
||
else:
|
||
date_str = datetime.now().strftime('%Y-%m-%d')
|
||
|
||
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')}"]
|
||
|
||
# 添加 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 深度内容分析工具")
|
||
parser.add_argument("--metadata", "-m", required=True, help="元数据 JSON 文件路径")
|
||
parser.add_argument("--content", "-c", required=True, help="内容 JSON 文件路径")
|
||
parser.add_argument("--output", "-o", required=True, help="输出 Markdown 文件路径")
|
||
parser.add_argument("--type", "-t", default="video", choices=["video", "article"], help="内容类型")
|
||
parser.add_argument("--api-key", "-k", default=None, help="AI API Key")
|
||
parser.add_argument("--provider", "-p", default=None, choices=["openai", "zhipu", "minimax"], help="AI 提供商")
|
||
parser.add_argument("--model", default=None, help="AI 模型")
|
||
parser.add_argument("--base-url", "-u", default=None, help="API Base URL")
|
||
parser.add_argument("--no-frontmatter", action="store_true", help="不生成 frontmatter")
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 加载数据
|
||
print(f"[Main] 加载元数据: {args.metadata}")
|
||
raw_metadata = load_json(args.metadata)
|
||
|
||
print(f"[Main] 加载内容: {args.content}")
|
||
content = load_json(args.content)
|
||
|
||
# 标准化元数据
|
||
metadata = {}
|
||
if 'title' in raw_metadata:
|
||
metadata = raw_metadata.copy()
|
||
# 处理 duration
|
||
duration = metadata.get('duration', 0)
|
||
if isinstance(duration, str) and ':' in duration:
|
||
parts = duration.split(':')
|
||
try:
|
||
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
|
||
except:
|
||
pass
|
||
metadata['duration_str'] = format_duration(metadata.get('duration', 0))
|
||
elif 'metadata' in raw_metadata:
|
||
meta = raw_metadata['metadata']
|
||
metadata = meta.copy()
|
||
duration = metadata.get('duration', 0)
|
||
if isinstance(duration, str) and ':' in duration:
|
||
parts = duration.split(':')
|
||
try:
|
||
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
|
||
except:
|
||
pass
|
||
metadata['duration_str'] = format_duration(metadata.get('duration', 0))
|
||
|
||
# 初始化
|
||
ai_client = SimpleAIClient(
|
||
api_key=args.api_key or os.environ.get("AI_API_KEY", ""),
|
||
provider=args.provider or os.environ.get("AI_PROVIDER", ""),
|
||
model=args.model or os.environ.get("AI_MODEL", "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)
|
||
|
||
# 清理 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:
|
||
"""格式化时长"""
|
||
h = seconds // 3600
|
||
m = (seconds % 3600) // 60
|
||
s = seconds % 60
|
||
if h > 0:
|
||
return f"{h}:{m:02d}:{s:02d}"
|
||
return f"{m}:{s:02d}"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception as e:
|
||
print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|