starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: remove old kb-deep-analyzer directory
PR Test / lint (pull_request) Successful in 1s
Details

Browse Source 
Code has been integrated into kb_workflow/deep_analyzer package.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
starmen1265
2026-03-17 15:31:18 +08:00
parent 9fc52ffbdd
commit c77a45ab0b
5 changed files with 0 additions and 2993 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kb-deep-analyzer/SKILL.md
-102
View File
@@ -1,102 +0,0 @@
# KB Deep Analyzer Skill
**kb-deep-analyzer** - AI 深度内容分析工具。
## Features
- 多类型分析：视频（video）、文章（article）
- 模板化输出
- 质量评分
- 自动标签
- **多格式元数据支持**: 支持扁平结构 `{"title":"xxx"}` 和嵌套结构 `{"data":{"metadata":{...}}}`
## 内容类型
| 类型 | 参数 | 输出格式 |
|------|------|----------|
| 视频 | `--type video` | 摘要 + 主题 + 见解 + 时间轴 |
| 文章 | `--type article` | 300字摘要 + 主题 + 原文 |
## Commands
```bash
# 分析视频（默认）
AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \
python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type video
# 分析文章（300字摘要 + 原文）
AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \
python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type article
# 指定 AI 提供商
python3 -u deep_analyzer.py -m meta.json -c content.json -o out.md --type video --provider zhipu --model glm-5
```
## 环境变量配置
**必须设置以下环境变量**：
```bash
export AI_API_KEY="your-api-key"
export AI_PROVIDER="zhipu" # 或 openai, minimax
export AI_MODEL="glm-5" # 模型名称
export AI_BASE_URL="https://..." # 可选，自定义 API 地址
```
### 重要提示
**必须使用 `-u` 参数运行 Python**，确保输出实时显示在终端：
```bash
python3 -u deep_analyzer.py ...
```
如果不使用 `-u` 参数，Python 的 stdout 缓冲会导致输出延迟或看不到进度。
## 执行方式选择
### 方式1: 并发执行（推荐，默认）
- 使用 `batch_size=3`（默认）
- 6个分片约 100 秒完成
- 适合快速分析
```bash
python3 -u deep_analyzer.py --metadata meta.json --content content.json -o out.md
```
### 方式2: 顺序执行
- 使用 `batch_size=1`
- 6个分片约 735 秒完成
- 适合调试和问题排查
```bash
# 在代码中设置 batch_size=1
# analyzer.analyze(metadata, content, "video", batch_size=1)
```
## Metadata Format
支持两种元数据格式：
```json
// 格式1: 扁平结构
{
"title": "视频标题",
"author": "作者名",
"duration": 619,
"bvid": "BVxxx"
}
// 格式2: 嵌套结构（kb-bilibili-metadata 输出）
{
"data": {
"metadata": {
"title": "视频标题",
"author": "作者名",
"duration": "10:19"
}
}
}
```
会自动转换为扁平格式进行分析。
kb-deep-analyzer/__pycache__/deep_analyzer.cpython-312.pyc
BIN
View File
Binary file not shown.
kb-deep-analyzer/ai_client.py
-514
View File
@@ -1,514 +0,0 @@
"""Obsidian KB - AI客户端
支持多个 AI 提供商和 fallback 机制
"""
import os
import json
import requests
from typing import Dict, Any, List, Optional, Tuple
from src.models import ContentType
from src.config import get_config
from src.content_chunker import ContentChunker, get_chunking_strategy, chunk_content
class AIClient:
"""
AI客户端 - 支持分段分析长视频内容和 fallback 机制
"""
def __init__(self):
self.config = get_config()
self.chunker = ContentChunker()
# 获取 AI 配置
self._load_ai_config()
def _load_ai_config(self) -> None:
"""加载 AI 配置（主 provider + fallbacks）"""
self.ai_configs = []
# 加载主配置
main_provider = self.config.get('ai.provider', 'openai')
main_config = {
'provider': main_provider,
'api_key': self.config.get('ai.api_key'),
'model': self.config.get('ai.model', 'gpt-4'),
'base_url': self.config.get('ai.base_url'),
'is_primary': True
}
self.ai_configs.append(main_config)
# 加载 fallback 配置
fallbacks = self.config.get('ai.fallbacks', [])
for fb in fallbacks:
self.ai_configs.append({
'provider': fb.get('provider'),
'api_key': fb.get('api_key'),
'model': fb.get('model'),
'base_url': fb.get('base_url'),
'is_primary': False
})
def _call_api_with_fallback(self, prompt: str, max_tokens: int = 1000, response_format: str = 'text') -> Tuple[bool, str]:
"""
使用 fallback 机制调用 API
Args:
prompt: 提示词
max_tokens: 最大 token 数
response_format: 响应格式 ('text' 或 'json')
Returns:
(success, response) 元组
"""
last_error = None
for i, ai_config in enumerate(self.ai_configs):
try:
if not ai_config.get('api_key'):
print(f"[AIClient] 跳过 {ai_config.get('provider')} - API Key 为空")
continue
provider_type = ai_config.get('provider')
print(f"[AIClient] 尝试使用 {provider_type} (优先级 {i+1}/{len(self.ai_configs)})...")
response = self._call_single_provider(ai_config, prompt, max_tokens, response_format)
return True, response
except Exception as e:
last_error = e
print(f"[AIClient] {provider_type} 调用失败: {e}")
continue
# 所有 provider 都失败了
error_msg = f"所有 AI 提供商均失败，最后错误: {last_error}" if last_error else "没有可用的 AI 配置"
print(f"[AIClient] {error_msg}")
return False, self._mock_generate(prompt, max_tokens) # 返回 mock 响应作为兜底
def _call_single_provider(self, ai_config: Dict, prompt: str, max_tokens: int, response_format: str) -> str:
"""
调用单个 AI 提供商
Args:
ai_config: AI 配置
prompt: 提示词
max_tokens: 最大 token 数
response_format: 响应格式
Returns:
API 响应文本
"""
provider = ai_config.get('provider')
api_key = ai_config.get('api_key')
model = ai_config.get('model')
base_url = ai_config.get('base_url')
if provider == 'minimax':
return self._call_minimax(api_key, model, base_url, prompt, max_tokens, response_format)
elif provider == 'zhipu':
return self._call_zhipu(api_key, model, base_url, prompt, max_tokens, response_format)
elif provider == 'alibaba':
return self._call_alibaba(api_key, model, base_url, prompt, max_tokens, response_format)
elif provider == 'openai':
return self._call_openai(api_key, model, base_url, prompt, max_tokens, response_format)
else:
raise ValueError(f"不支持的 provider: {provider}")
def _call_minimax(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用 MiniMax API (Anthropic Messages 兼容接口)"""
url = base_url or "https://api.minimaxi.com/anthropic/v1/messages"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
if response_format == 'json':
payload['messages'][0]['content'] = prompt + "\n\n请直接返回 JSON 格式，不要有其他文字。"
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
# MiniMax 返回的是 Anthropic 格式，content 是一个数组
# 需要找到 type='text' 的元素
for item in data.get('content', []):
if item.get('type') == 'text':
return item.get('text', '')
# 如果没有找到 text 类型，返回第一个元素的值（兼容旧格式）
return data.get('content', [{}])[0].get('text', '')
def _call_zhipu(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用智谱 AI API (OpenAI 兼容接口)"""
# 确保 base_url 包含 /chat/completions
if base_url and not base_url.endswith('/chat/completions'):
url = f"{base_url}/chat/completions"
else:
url = base_url or "https://open.bigmodel.cn/api/paas/v4/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def _call_alibaba(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用阿里云通义千问 API (OpenAI 兼容接口)"""
url = f"{base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def _call_openai(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
"""调用 OpenAI API"""
url = base_url or "https://api.openai.com/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"max_tokens": max_tokens,
"messages": [
{"role": "user", "content": prompt}
]
}
response = requests.post(url, headers=headers, json=payload, timeout=30)
response.raise_for_status()
data = response.json()
return data['choices'][0]['message']['content']
def quick_generate(self, prompt: str, max_tokens: int = 100) -> str:
"""快速生成（用于轻量扫描）"""
success, response = self._call_api_with_fallback(prompt, max_tokens, 'text')
return response
def generate_json(self, prompt: str) -> Dict[str, Any]:
"""生成JSON格式响应"""
success, response = self._call_api_with_fallback(prompt, 1000, 'json')
try:
return json.loads(response)
except:
return self._parse_json_from_text(response)
def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]:
"""
C5.3: 深度分析入口，根据内容时长决定是否分段分析
Args:
content_text: 内容文本
content_type: 内容类型
duration: 视频时长（秒）
Returns:
分析结果字典
"""
duration = duration or 0
print(f"[AIClient] 深度分析: type={content_type}, duration={duration}s")
# 长视频（>=30分钟）且是视频类型，使用分段分析
if duration >= 1800 and content_type == "video":
print(f"[AIClient] 使用分段分析策略")
return self._analyze_in_chunks(content_text, duration)
# 短视频直接分析
print(f"[AIClient] 使用直接分析策略")
return self._analyze_direct(content_text)
def _analyze_in_chunks(self, content_text: str, duration: int) -> Dict[str, Any]:
"""
C5.3: 分段分析长视频
Args:
content_text: SRT字幕文本
duration: 视频时长（秒）
Returns:
合并后的分析结果
"""
# 1. 获取分段策略
strategy = get_chunking_strategy(duration)
print(f"[AIClient] 分段策略: {strategy['description']}")
# 2. 执行分片
chunks = chunk_content(content_text, duration)
print(f"[AIClient] 内容已分为 {len(chunks)} 段")
# 3. 逐段分析
chunk_summaries = []
for i, chunk in enumerate(chunks):
print(f"[AIClient] 分析第 {i+1}/{len(chunks)} 段...")
summary = self._analyze_chunk(chunk.text, chunk.index)
# 处理时间格式（可能是float秒数或字符串）
start_time = chunk.start_time
end_time = chunk.end_time
if isinstance(start_time, (int, float)):
start_time = self._format_time(start_time)
if isinstance(end_time, (int, float)):
end_time = self._format_time(end_time)
chunk_summaries.append({
'index': chunk.index,
'start_time': start_time,
'end_time': end_time,
'summary': summary
})
# 4. 合并所有段摘要
print(f"[AIClient] 合并 {len(chunk_summaries)} 段分析结果")
final_result = self._merge_summaries(chunk_summaries, "video")
# 5. 构建完整时间轴
final_result['timeline'] = self._build_timeline_from_chunks(chunk_summaries)
return final_result
def _analyze_chunk(self, chunk_text: str, chunk_index: int) -> str:
"""
C5.3: 分析单个内容分片
Args:
chunk_text: 分片文本
chunk_index: 分片索引
Returns:
分片摘要（限制500字以内）
"""
# 截断文本避免过长
text_sample = chunk_text[:3000] if len(chunk_text) > 3000 else chunk_text
prompt = f"""请对以下视频片段进行简要分析（限制在300字以内）：
片段内容：
{text_sample}
请提供：
1. 这段的核心主题（50字以内）
2. 关键要点（3-5条）
3. 重要细节或引用
请用中文回答，控制在300字以内。"""
try:
summary = self.quick_generate(prompt, max_tokens=500)
# 确保不超过500字
if len(summary) > 500:
summary = summary[:497] + "..."
return summary
except Exception as e:
print(f"[AIClient] 分片 {chunk_index} 分析失败: {e}")
return f"第{chunk_index+1}段分析失败"
def _merge_summaries(self, chunk_summaries: List[Dict], content_type: str) -> Dict[str, Any]:
"""
C5.3: 合并所有段摘要，生成完整分析报告
Args:
chunk_summaries: 各段摘要列表
content_type: 内容类型
Returns:
完整分析结果
"""
# 构建合并提示
summaries_text = "\n\n".join([
f"=== 第{s['index']+1}段 ({s['start_time']}-{s['end_time']}) ===\n{s['summary']}"
for s in chunk_summaries
])
prompt = f"""基于以下各段分析，生成完整的视频分析报告：
{summaries_text}
请提供以下格式的JSON：
{{
"summary": "整体摘要（200字以内）",
"topics": ["主题1", "主题2", "主题3", "主题4"],
"insights": ["深入见解1", "深入见解2", "深入见解3"],
"actions": ["建议行动1", "建议行动2"],
"full_analysis": "详细分析（500字以内）"
}}"""
try:
result = self.generate_json(prompt)
# 确保必要字段存在
if 'topics' not in result:
result['topics'] = ["主题A", "主题B", "主题C"]
if 'insights' not in result:
result['insights'] = ["重要见解1", "重要见解2"]
if 'actions' not in result:
result['actions'] = []
if 'summary' not in result:
result['summary'] = "长视频分析完成"
if 'full_analysis' not in result:
result['full_analysis'] = summaries_text[:500]
return result
except Exception as e:
print(f"[AIClient] 合并摘要失败: {e}")
# 返回基础合并结果
return {
'summary': f'视频共分为{len(chunk_summaries)}个部分进行分析',
'topics': [f"主题{i+1}" for i in range(min(4, len(chunk_summaries)))],
'insights': ['从分段分析中综合得出'],
'actions': [],
'full_analysis': summaries_text[:800]
}
def _analyze_direct(self, content_text: str) -> Dict[str, Any]:
"""直接分析（不分段）"""
# 截断文本
text_sample = content_text[:5000] if len(content_text) > 5000 else content_text
prompt = f"""请分析以下内容，提供JSON格式结果：
{text_sample}
返回格式：
{{
"summary": "整体摘要（150字以内）",
"topics": ["主题1", "主题2", "主题3"],
"insights": ["见解1", "见解2"],
"actions": ["行动建议1"],
"timeline": [
{{"start_time": "00:00", "end_time": "05:00", "title": "第一部分", "summary": "..."}}
]
}}"""
try:
return self.generate_json(prompt)
except Exception as e:
print(f"[AIClient] 直接分析失败: {e}")
return {
'summary': '分析失败',
'topics': ['无法解析'],
'insights': ['请检查内容格式'],
'actions': [],
'timeline': []
}
def _format_time(self, seconds: float) -> str:
"""格式化秒数为时间字符串"""
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
if hours > 0:
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
return f"{minutes:02d}:{secs:02d}"
def _build_timeline_from_chunks(self, chunk_summaries: List[Dict]) -> List[Dict[str, Any]]:
"""从分片摘要构建时间轴"""
timeline = []
for summary in chunk_summaries:
timeline.append({
'index': summary['index'],
'start_time': summary['start_time'],
'end_time': summary['end_time'],
'title': f"第{summary['index']+1}部分",
'summary': summary['summary'][:200] + "..." if len(summary['summary']) > 200 else summary['summary']
})
return timeline
def _mock_generate(self, prompt: str, max_tokens: int) -> str:
"""模拟AI生成（实际实现应调用真实API）"""
# 根据prompt内容返回模拟响应
prompt_lower = prompt.lower()
if 'json' in prompt_lower or 'json格式' in prompt:
if '主题' in prompt or 'topic' in prompt_lower:
return json.dumps({
'summary': '这是一个关于人工智能和技术发展的内容，涵盖了多个重要主题。',
'topics': ['人工智能', '机器学习', '未来发展', '技术趋势'],
'insights': [
'AI正在改变我们的工作方式',
'机器学习应用越来越广泛',
'未来技术发展趋势值得关注'
],
'actions': ['深入研究相关技术', '关注行业动态'],
'full_analysis': '详细分析内容...'
}, ensure_ascii=False)
return json.dumps({'result': '模拟JSON响应'})
if '摘要' in prompt or 'summary' in prompt_lower:
return '这是一段关于人工智能发展的重要讨论，涵盖了技术趋势和未来应用。'
if '关键词' in prompt or 'keyword' in prompt_lower:
return '人工智能, 机器学习, 深度学习, 神经网络, 自然语言处理'
return '模拟AI响应内容'
def _parse_json_from_text(self, text: str) -> Dict[str, Any]:
"""从文本中解析JSON"""
# 尝试提取JSON块
import re
json_match = re.search(r'\{[\s\S]*\}', text)
if json_match:
try:
return json.loads(json_match.group())
except:
pass
return {'raw_response': text}
class MultiAIClient:
"""
多AI客户端 - 支持多种AI提供商的客户端
"""
def __init__(self):
self.config = get_config()
self.primary_client = AIClient()
self.clients = {
'openai': self.primary_client,
'local': AIClient(), # 可以配置为本地模型
}
def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]:
"""使用主客户端进行深度分析"""
return self.primary_client._deep_analyze(content_text, content_type, duration)
def quick_generate(self, prompt: str, max_tokens: int = 100, provider: str = None) -> str:
"""快速生成"""
client = self.clients.get(provider, self.primary_client)
return client.quick_generate(prompt, max_tokens)
kb-deep-analyzer/deep_analyzer.py
-650
View File
@@ -1,650 +0,0 @@
#!/usr/bin/env python3
"""Deep Analyzer - AI 深度内容分析工具
一次性调用 AI 生成完整深度分析报告。
Usage:
python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md [--type video]
python deep_analyzer.py --help
"""
import argparse
import json
import os
import sys
import re
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, List
from dataclasses import dataclass, field
# ============== 数据模型 ==============
@dataclass
class AnalysisResult:
"""分析结果"""
summary: str = ""
main_topics: List[str] = field(default_factory=list)
key_insights: List[str] = field(default_factory=list)
recommended_tags: List[str] = field(default_factory=list)
full_analysis: str = ""
original_text: str = "" # 文章类型的原文内容
# ============== AI 客户端 ==============
class SimpleAIClient:
"""AI 客户端"""
def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None):
self.api_key = api_key or os.environ.get("AI_API_KEY", "")
self.provider = provider or os.environ.get("AI_PROVIDER", "")
self.model = model or os.environ.get("AI_MODEL", "glm-5")
self.base_url = base_url or os.environ.get("AI_BASE_URL", "")
self.max_retries = 3
def generate(self, prompt: str, max_tokens: int = 16000) -> str:
"""生成文本"""
api_key = os.environ.get("AI_API_KEY", self.api_key)
provider = os.environ.get("AI_PROVIDER", self.provider)
model = os.environ.get("AI_MODEL", self.model)
base_url = os.environ.get("AI_BASE_URL", self.base_url)
if not api_key:
raise ValueError("AI_API_KEY 未设置")
for attempt in range(self.max_retries):
try:
import time
start_time = time.time()
print(f"[AI Client] 开始调用 AI...")
if provider == "openai":
result = self._call_openai(prompt, max_tokens, api_key, model, base_url)
elif provider == "zhipu":
result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url)
elif provider == "minimax":
result = self._call_minimax(prompt, max_tokens, api_key, model, base_url)
else:
raise ValueError(f"不支持的 AI 提供商: {provider}")
elapsed = time.time() - start_time
print(f"[AI Client] AI 调用完成，耗时: {elapsed:.2f}秒")
return result
except Exception as e:
if attempt < self.max_retries - 1:
print(f"[AI Client] 第 {attempt + 1} 次调用失败，{self.max_retries - attempt - 1} 次重试剩余: {e}")
continue
else:
raise
def _call_openai(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
import requests
url = f"{base_url}/chat/completions" if base_url else "https://api.openai.com/v1/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]}
response = requests.post(url, headers=headers, json=payload, timeout=600)
response.raise_for_status()
return response.json()['choices'][0]['message']['content']
def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
import requests
base_url = base_url or "https://open.bigmodel.cn/api/coding/paas/v4"
url = f"{base_url}/chat/completions"
headers = {"Authorization": f"Bearer {api_key}"}
payload = {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens}
response = requests.post(url, headers=headers, json=payload, timeout=600)
response.raise_for_status()
data = response.json()
message = data.get('choices', [{}])[0].get('message', {})
content = message.get('content', '')
reasoning = message.get('reasoning_content', '')
if not content and reasoning:
sentences = reasoning.strip().split('。')
useful = [s.strip() for s in sentences if s.strip() and len(s.strip()) > 10]
if useful:
return '。'.join(useful[-3:]) + '。'
return reasoning.strip()[:500]
return content
def _call_minimax(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str:
import requests
url = base_url or "https://api.minimaxi.com/anthropic/v1/messages"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {"model": model, "max_tokens": max_tokens, "messages": [{"role": "user", "content": prompt}]}
response = requests.post(url, headers=headers, json=payload, timeout=600)
response.raise_for_status()
data = response.json()
for item in data.get('content', []):
if item.get('type') == 'text':
return item.get('text', '')
return str(data)
# ============== 分析器 ==============
class DeepAnalyzer:
"""深度内容分析器"""
def __init__(self, ai_client: SimpleAIClient = None):
self.ai_client = ai_client or SimpleAIClient()
def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any], content_type: str = "video") -> AnalysisResult:
"""一次性调用 AI 生成完整分析报告
Args:
metadata: 元数据字典
content: 内容字典
content_type: 内容类型，"video" 或 "article"
"""
# 标准化元数据
metadata = self._normalize_metadata(metadata)
title = metadata.get('title', '未知')
duration = metadata.get('duration', 0)
author = metadata.get('author', '')
pubtime = metadata.get('pubtime', '')
url = metadata.get('url', '')
bvid = metadata.get('bvid', '')
# 推断 platform：优先使用元数据中的值，否则根据 URL 或 content_type 推断
platform = metadata.get('platform', '')
if not platform:
if 'zhihu.com' in url or content_type in ('answer', 'zhuanlan', 'question'):
platform = 'zhihu'
elif 'bilibili.com' in url or bvid:
platform = 'bilibili'
else:
platform = 'bilibili' # 默认值
# 清理 author 字段：移除零宽度空格和前后空白
author = self._clean_author(author)
# 获取内容 - 支持多种字段名
text = content.get('text', '') or content.get('subtitle', '') or content.get('subtitles', '') or content.get('description', '')
print(f"[DeepAnalyzer] 开始分析: {title}")
print(f"[DeepAnalyzer] 类型: {content_type}, 内容长度: {len(text)} 字符")
# 构建 prompt（根据类型选择不同模板）
prompt = self._build_prompt(
title=title,
duration=duration,
duration_str=self._format_duration(duration),
author=author,
pubtime=pubtime,
url=url,
bvid=bvid,
platform=platform,
text=text,
content_type=content_type
)
# 调用 AI
try:
full_analysis = self.ai_client.generate(prompt, max_tokens=16000)
except Exception as e:
print(f"[DeepAnalyzer] AI 调用失败: {e}")
raise
# 解析结果
result = self._parse_analysis(full_analysis, text, duration, content_type)
# 文章/回答类型：追加原文
if content_type in ("article", "answer", "zhuanlan"):
result.original_text = text
return result
def _build_prompt(self, title: str, duration: int, duration_str: str, author: str,
pubtime: str, url: str, bvid: str, platform: str, text: str,
content_type: str = "video") -> str:
"""构建分析 prompt，根据内容类型选择不同模板"""
if content_type in ("article", "answer", "zhuanlan"):
return self._build_article_prompt(title, author, pubtime, url, platform, text)
else:
return self._build_video_prompt(title, duration, duration_str, author, pubtime, url, bvid, platform, text)
def _build_video_prompt(self, title: str, duration: int, duration_str: str, author: str,
pubtime: str, url: str, bvid: str, platform: str, text: str) -> str:
"""构建视频分析 prompt"""
# 计算分片数（按10分钟分片）
num_chunks = max(1, (duration + 599) // 600)
# 截取适当长度的文本（20-30K 字符）
max_chars = 30000
if len(text) > max_chars:
text_sample = text[:max_chars] + "\n\n[...]"
else:
text_sample = text
prompt = f"""你是一个专业的视频内容分析师。请根据以下信息生成深度分析报告：
【视频信息】
标题：{title}
作者：{author}
时长：{duration_str}（{duration}秒）
发布日期：{pubtime}
平台：{platform}
URL：{url}
【内容】（完整字幕）
{text_sample}
【输出要求】必须使用以下格式：
## 标题
[自己总结的简洁标题，禁止使用原标题]
## 摘要
[200-400字]
## 主要主题
[从内容中提取不超过10个主题，每个不超过10字]
- 主题1
- 主题2
...
## 关键见解
[若干条深刻见解]
- 见解1
- 见解2
...
## 标签
[从内容中提取3-5个精炼标签，每个不超过10字]
- 标签1
- 标签2
...
## 时间轴
### 第 1 部分 (00:00 - 10:00)
[详细分析，至少500字]
### 第 2 部分 (10:00 - 20:00)
[详细分析，至少500字]
...（以此类推）
【重要】分析应足够详尽，读者阅读后可了解视频核心内容，无需再看原视频。"""
return prompt
def _build_article_prompt(self, title: str, author: str, pubtime: str,
url: str, platform: str, text: str) -> str:
"""构建文章分析 prompt"""
# 文章保留完整内容（不截断，因为需要追加原文）
# 但 prompt 中只取前 20000 字符供 AI 分析
max_chars = 20000
if len(text) > max_chars:
text_sample = text[:max_chars] + "\n\n[...内容过长已截断，但原文会完整保留...]"
else:
text_sample = text
prompt = f"""你是一个专业的文章内容分析师。请根据以下信息生成摘要：
【文章信息】
标题：{title}
作者：{author}
发布日期：{pubtime}
平台：{platform}
URL：{url}
【内容】
{text_sample}
【输出要求】必须使用以下格式：
## 摘要
[300字以内的摘要，描述文章的核心观点和主要内容]
## 主要主题
[从内容中提取不超过5个主题，每个不超过10字]
- 主题1
- 主题2
...
## 标签
[从内容中提取3-5个精炼标签，每个不超过10字]
- 标签1
- 标签2
...
【重要】
1. 摘要应精炼概括文章核心观点，让读者快速了解文章主旨
2. 不要输出时间轴
3. 不要输出关键见解（除非是特别重要的观点）"""
return prompt
def _parse_analysis(self, analysis_text: str, original_text: str, duration: int,
content_type: str = "video") -> AnalysisResult:
"""解析 AI 返回的分析结果"""
result = AnalysisResult()
result.full_analysis = analysis_text
# 提取各部分（使用正则）
lines = analysis_text.split('\n')
# 简单解析：按 ## 标题分割
current_section = ""
in_section = None
for line in lines:
if line.startswith('##'):
# 保存上一个 section
if in_section:
self._store_section(result, in_section, current_section)
# 开始新 section
section_name = line.lstrip('#').strip()
in_section = section_name
current_section = ""
elif in_section:
current_section += line + "\n"
# 存储最后一个 section
if in_section:
self._store_section(result, in_section, current_section)
# 如果解析失败，使用原始文本
if not result.summary:
result.summary = analysis_text[:500]
return result
def _store_section(self, result: AnalysisResult, section_name: str, content: str):
"""存储解析到的 section"""
content = content.strip()
if not content:
return
section_lower = section_name.lower()
if '摘要' in section_lower:
result.summary = content[:500]
elif '主题' in section_lower:
# 提取主题
topics = []
for line in content.split('\n'):
line = line.strip().lstrip('-').strip()
if line and len(line) <= 15:
topics.append(line)
result.main_topics = topics[:10]
elif '见解' in section_lower or '观点' in section_lower:
# 提取见解
insights = []
for line in content.split('\n'):
line = line.strip().lstrip('-').strip()
if line and len(line) > 5:
insights.append(line)
result.key_insights = insights
elif '标签' in section_lower:
# 提取标签
tags = []
for line in content.split('\n'):
line = line.strip().lstrip('-').strip()
if line and len(line) <= 10:
tags.append(line)
result.recommended_tags = tags[:5]
def _normalize_metadata(self, metadata: Dict) -> Dict:
"""标准化元数据"""
if not metadata:
return {}
# 如果已经扁平
if 'title' in metadata:
duration = metadata.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
return metadata
# 嵌套结构
if 'metadata' in metadata:
meta = metadata['metadata']
normalized = {}
for k, v in meta.items():
normalized[k] = v
# 处理 duration
duration = normalized.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
normalized['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
return normalized
return metadata
def _format_duration(self, seconds: int) -> str:
"""格式化时长"""
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
def _clean_author(self, author: str) -> str:
"""清理 author 字段：移除零宽度空格和前后空白"""
if not author:
return ''
# 移除零宽度空格 (U+200B) 和其他不可见字符
invisible_chars = ['\u200b', '\u200c', '\u200d', '\ufeff', '\u00ad']
cleaned = author
for char in invisible_chars:
cleaned = cleaned.replace(char, '')
# 移除前后空白和换行
cleaned = cleaned.strip()
# 如果包含换行符，只取第一行
if '\n' in cleaned:
cleaned = cleaned.split('\n')[0].strip()
return cleaned
# ============== 工具函数 ==============
def load_json(file_path: str) -> Dict:
"""加载 JSON 文件"""
if not file_path or not os.path.exists(file_path):
return {}
with open(file_path, 'r', encoding='utf-8') as f:
return json.load(f)
def generate_frontmatter(metadata: Dict, result: AnalysisResult, content_type: str = "video") -> str:
"""生成 Markdown frontmatter"""
# 提取标签
tags = result.recommended_tags or []
# 格式化日期
pubtime = metadata.get('pubtime', '')
if pubtime:
date_str = str(pubtime)[:10]
else:
date_str = datetime.now().strftime('%Y-%m-%d')
# 清理 author 字段
author = metadata.get('author', '未知')
author = _clean_string(author)
# 推断 platform
url = metadata.get('url', '')
platform = metadata.get('platform', '')
if not platform:
if 'zhihu.com' in url or content_type in ('answer', 'zhuanlan', 'question'):
platform = 'zhihu'
elif 'bilibili.com' in url or metadata.get('bvid'):
platform = 'bilibili'
else:
platform = 'bilibili'
fm = ["---",
f"title: \"{metadata.get('title', '未命名')}\"",
f"date: {date_str}",
f"author: {author}",
f"platform: {platform}"]
# 视频类型添加 duration 和 bvid
if content_type == "video":
fm.insert(3, f"duration: {metadata.get('duration_str', metadata.get('duration', 0))}")
bvid = metadata.get('bvid', '')
if bvid:
fm.insert(5, f"bvid: {bvid}")
# 添加 URL
url = metadata.get('url', '')
if url:
fm.append(f"url: {url}")
if tags:
fm.append("tags: [" + ", ".join(f'"{t}"' for t in tags) + "]")
fm.append("---")
return '\n'.join(fm)
def _clean_string(text: str) -> str:
"""清理字符串：移除零宽度空格和前后空白"""
if not text:
return ''
# 移除零宽度空格 (U+200B) 和其他不可见字符
invisible_chars = ['\u200b', '\u200c', '\u200d', '\ufeff', '\u00ad']
cleaned = text
for char in invisible_chars:
cleaned = cleaned.replace(char, '')
# 移除前后空白和换行
cleaned = cleaned.strip()
# 如果包含换行符，只取第一行
if '\n' in cleaned:
cleaned = cleaned.split('\n')[0].strip()
return cleaned
# ============== 主程序 ==============
def main():
parser = argparse.ArgumentParser(description="Deep Analyzer - AI 深度内容分析工具")
parser.add_argument("--metadata", "-m", required=True, help="元数据 JSON 文件路径")
parser.add_argument("--content", "-c", required=True, help="内容 JSON 文件路径")
parser.add_argument("--output", "-o", required=True, help="输出 Markdown 文件路径")
parser.add_argument("--type", "-t", default="video", choices=["video", "article"], help="内容类型")
parser.add_argument("--api-key", "-k", default=None, help="AI API Key")
parser.add_argument("--provider", "-p", default=None, choices=["openai", "zhipu", "minimax"], help="AI 提供商")
parser.add_argument("--model", default=None, help="AI 模型")
parser.add_argument("--base-url", "-u", default=None, help="API Base URL")
parser.add_argument("--no-frontmatter", action="store_true", help="不生成 frontmatter")
args = parser.parse_args()
# 加载数据
print(f"[Main] 加载元数据: {args.metadata}")
raw_metadata = load_json(args.metadata)
print(f"[Main] 加载内容: {args.content}")
content = load_json(args.content)
# 标准化元数据
metadata = {}
if 'title' in raw_metadata:
metadata = raw_metadata.copy()
# 处理 duration
duration = metadata.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
metadata['duration_str'] = format_duration(metadata.get('duration', 0))
elif 'metadata' in raw_metadata:
meta = raw_metadata['metadata']
metadata = meta.copy()
duration = metadata.get('duration', 0)
if isinstance(duration, str) and ':' in duration:
parts = duration.split(':')
try:
metadata['duration'] = int(parts[0]) * 60 + int(parts[1])
except:
pass
metadata['duration_str'] = format_duration(metadata.get('duration', 0))
# 初始化
ai_client = SimpleAIClient(
api_key=args.api_key or os.environ.get("AI_API_KEY", ""),
provider=args.provider or os.environ.get("AI_PROVIDER", ""),
model=args.model or os.environ.get("AI_MODEL", "glm-5"),
base_url=args.base_url or os.environ.get("AI_BASE_URL", "")
)
analyzer = DeepAnalyzer(ai_client)
# 执行分析
print(f"[Main] 开始分析...")
result = analyzer.analyze(metadata, content, args.type)
# 使用 AI 生成的原始内容（保留 Markdown 格式）
full_analysis = result.full_analysis
# 删除内容中的"标签" section（只保留 frontmatter 中的 tags）
# 匹配 ## 标签 或 标签（无##）到下一个 ## 或文件结尾之间的内容
full_analysis = re.sub(r'(## ?)?标签\n[\s\S]*?(?=\n## ?|\n时间轴|---|\Z)', '', full_analysis)
# 清理多余的空行
full_analysis = re.sub(r'\n{3,}', '\n\n', full_analysis)
# 文章类型：追加原文
if args.type == "article" and result.original_text:
full_analysis = full_analysis.rstrip() + "\n\n## 原文\n\n" + result.original_text
# 写入输出
os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True)
with open(args.output, 'w', encoding='utf-8') as f:
# 添加 frontmatter
if not args.no_frontmatter:
fm = generate_frontmatter(metadata, result, args.type)
f.write(fm + "\n\n")
# 写入分析内容
f.write(full_analysis)
# 输出结果
output = {
"success": True,
"filename": Path(args.output).name,
"filepath": str(Path(args.output).resolve()),
"message": "深度分析完成",
"stats": {
"summary_length": len(result.summary) if result.summary else 0,
"topics_count": len(result.main_topics),
"insights_count": len(result.key_insights),
"tags_count": len(result.recommended_tags)
}
}
print(json.dumps(output, ensure_ascii=False, indent=2))
def format_duration(seconds: int) -> str:
"""格式化时长"""
h = seconds // 3600
m = (seconds % 3600) // 60
s = seconds % 60
if h > 0:
return f"{h}:{m:02d}:{s:02d}"
return f"{m}:{s:02d}"
if __name__ == "__main__":
try:
main()
except Exception as e:
print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2))
sys.exit(1)
kb-deep-analyzer/deep_analyzer.py.bak
-1727
View File
File diff suppressed because it is too large Load Diff