starmen1265
9fc52ffbdd
- Move kb-deep-analyzer/deep_analyzer.py to kb_workflow/deep_analyzer/__init__.py - Update analyze.py to import from kb_workflow.deep_analyzer directly - Remove sys.path manipulation for external package - DeepAnalyzer, SimpleAIClient, generate_frontmatter now available as: from kb_workflow.deep_analyzer import DeepAnalyzer, SimpleAIClient, generate_frontmatter Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
651 lines
23 KiB
Python
651 lines
23 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 = ""
|
||
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)
|