AIによるコンテンツ制作：規制と創造性のバランスを見つける

多くのコンテンツ制作者にとって、生々しく退屈なRSSニュースフィードを、明確に構成され、洞察に富み、一貫したスタイルの論説に変えることは、時間のかかる作業である。情報の洪水は毎日やってくるが、それを魅力的なコンテンツに変える苦労は、しばしば「機械的に動かす」ことと「アイデアが尽きる」ことの狭間で苦闘することになる。

この記事では、このプロセスを効率的かつインテリジェントにする自動化ツールのアイデアを探る。このツールの核となる目的は明確である。RSSフィードから未処理の情報を取り出し、大規模言語モデリング（LLM）処理によって出版に適した高品質の記事に整形することである。

創造性の「温度」をコントロールする

RSSフィードは、情報を正確に伝える忠実なメッセンジャーのようなものだが、解釈や感情が欠けている。一方、優れた自動化ツールは、コラムニストの役割を担い、事実を提示するだけでなく、それらに視点と生命を与えるべきである。

これを実現する鍵は、AIモデルの創造性を正確にコントロールすることだ。そのためには temperature パラメータは、ユーザーがモデルの出力スタイルを操作できるようにします。これは通常0から1の間の値で、モデルの予測値のランダム性を調整するために使用されます。より低い temperature 0.2のような値は、出力をより決定論的で一貫性のあるものにし、厳密で事実に基づいた報告に適している。一方、より高い値（例えば0.8）は、モデルの「想像力」を刺激し、より多様で創造的なプレゼンテーションを生み出す。

ライティング・アシスタントの「三位一体

さまざまなシナリオのニーズに応えるため、このツールには3つの「ライティング・パーソナリティ」が組み込まれており、それぞれが厳密さと創造性の異なるバランスを表している：

• デイリー・ブリーフィング・パーソナリティ技術系ブロガーの親しみやすい口調を模倣し、日々のAIニュースを軽快なブリーフィングにまとめている。生命体のような比喩と適切な絵文字を使い、読者との距離を縮めつつ、核心的なメッセージの正確さを保証する。
• 個性の徹底分析業界アナリストの視点に切り替えると、ニュースの背景にある技術的な性質や市場への影響について掘り下げる。より低い傾向 temperatureその目的は、専門的かつ論理的なアウトプットを確実にすることであり、同時にフレームワークの中で洞察を深めることである。
• 白人翻訳者の性格その中心的な使命は単純化することである。例えば、「ミクスチャー・オブ・エキスパート（MoE）」という言葉については、読者に直接投げかけるのではなく、「専門家がチームを組んで、それぞれが得意な問題の一部分だけを担当することで、全体の効率と精度を向上させる」と例えて、初心者の読者でも理解しやすいようにしている。基礎知識のない読者にも理解しやすいように、「効率性と正確性」を重視している。

これら3つの性格により、文体的には柔軟でありながら、記事構成に関しては一貫したアウトプットが可能となり、読者が安定した読書期待を抱くことができる。

堅牢なオペレーションのための技術的基盤

本当に使えるツールは、単にAPIを呼び出すだけではだめで、スムーズで信頼できるプロセスを保証するために、その背後にある強固なエンジニアリング・セットが必要なのだ。

• 自動リトライ機構ネットワークの変動やAPIの一時的な障害はよくある問題です。エクスポネンシャルバックオフリトライメカニズムを導入することで、失敗後一定時間待ってから自動的にリトライすることができ、タスクの成功率を大幅に向上させることができる。
• キャッシング・システム更新頻度の低いRSSフィードの場合、同じコンテンツを何度もクロールするのはリソースの無駄遣いです。シンプルなキャッシュシステムを使えば、取得したデータを一時的に保存し、一定時間内にローカルから直接読み込むことができるため、ネットワークへのリクエストや処理時間を効果的に削減することができます。
• ストリーミング出力:: ストリーミング出力では、モデルがコンテンツ全体を生成するのを待つ代わりに、ビデオの読み込みのように、生成されたテキストを逐語的または文単位でリアルタイムに表示することができます。これにより、ユーザーはコンテンツの品質を即座にプレビューできるだけでなく、タスクがコースから外れていることが判明した場合、タスクを中止することも容易になります。
• フレキシブルな構成様々な環境に対応するため、コマンドラインパラメータ、環境変数、設定ファイルなどによる設定をサポートしています。ユーザーは、ソースコードを変更することなく、モデル、APIキー、出力ディレクトリなどを調整することができます。
• 包括的な例外処理APIキーの紛失、RSSアドレスへのアクセス不能など、不測の事態に潔く対応する。

実際の効果

想像してみてほしい。 Google リリース名は Gemini 2.5 Pro 組み込み DeepThink 推論機能の新しいモデルが「白の翻訳者人格」によって処理されるとき、それは次のように解釈されるかもしれない：

Gemini 2.5 Pro それはまるで、さまざまな作業を手助けしてくれる非常に賢いAIのパートナーのようだ。さらに DeepThink 一方、モードは深く考える能力を与えてくれる。つまり、もはや「何」に答えるだけでなく、「なぜ」を説明し、問題の核心へと導いてくれるメンターのような存在になるのだ。

このような扱いは、技術的な専門知識を保持したまま、表現の大衆化を実現する。それは、創造性とは野放図な想像力ではなく、明確なルールの枠組みのもとでの秩序ある革新であることを証明している。ジャズの即興演奏は、自由奔放に見えるが、実際にはそれぞれのステップがハーモニーの軌道から逸脱していないのと同じだ。

コンテンツ制作者が「技術的な深み」と「大衆科学」の間で難しい選択をする必要がなくなったとき、テクノロジーの価値は真に発揮される。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import feedparser
import datetime
import requests
import os
import json
import time
import logging
import traceback
import argparse
import configparser
from pathlib import Path
from datetime import timedelta
from functools import wraps
from typing import List, Dict, Any, Optional, Callable, TypeVar, Union
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("ai_news_generator.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger("ai_news_generator")
# 类型声明
T = TypeVar('T')
FeedEntry = Dict[str, Any]
ApiResponse = Dict[str, Any]
# 默认配置
DEFAULT_CONFIG = {
"api": {
"base_url": "https://xxxxxxxxxxxxx/openai",
"model": "xxxxxxxxx",
"api_key": os.environ.get("OPENAI_API_KEY", ""),
"max_tokens": 2000,
"timeout": 60,
"temperature": 0.7
},
"rss": {
"url": "https://news.smol.ai/rss.xml",
"days": 7,
"cache_time": 3600  # 缓存RSS内容的时间（秒）
},
"output": {
"directory": "ai_news_output",
"format": "markdown"
}
}
# 重试装饰器
def retry(max_attempts: int = 3, delay: int = 2, backoff: int = 2,
exceptions: tuple = (Exception,)) -> Callable:
"""
重试装饰器，用于处理可能失败的操作
参数:
max_attempts: 最大尝试次数
delay: 初始延迟时间（秒）
backoff: 延迟的倍数（指数退避）
exceptions: 要捕获的异常元组
"""
def decorator(func):
        @wraps(func)
def wrapper(*args, **kwargs):
attempt = 0
current_delay = delay
while attempt < max_attempts:
try:
return func(*args, **kwargs)
except exceptions as e:
attempt += 1
if attempt == max_attempts:
logger.error(f"最大尝试次数已用完 ({max_attempts})，操作失败: {e}")
raise
logger.warning(f"尝试 {attempt}/{max_attempts} 失败: {e}. "
f"将在 {current_delay} 秒后重试...")
time.sleep(current_delay)
current_delay *= backoff
return wrapper
return decorator
class Config:
"""配置管理类"""
def __init__(self, config_file: Optional[str] = None):
"""
初始化配置
参数:
config_file: 配置文件路径，如果不存在则使用默认配置
"""
self.config = DEFAULT_CONFIG.copy()
if config_file and os.path.exists(config_file):
self._load_from_file(config_file)
else:
logger.info("未找到配置文件，使用默认配置")
# 环境变量优先级高于配置文件
if os.environ.get("OPENAI_API_KEY"):
self.config["api"]["api_key"] = os.environ.get("OPENAI_API_KEY")
def _load_from_file(self, config_file: str) -> None:
"""从文件加载配置"""
try:
parser = configparser.ConfigParser()
parser.read(config_file)
# 将配置文件值更新到默认配置
for section in parser.sections():
if section in self.config:
for key, value in parser.items(section):
# 尝试转换类型以匹配默认配置
if key in self.config[section]:
original_type = type(self.config[section][key])
if original_type is int:
self.config[section][key] = int(value)
elif original_type is float:
self.config[section][key] = float(value)
elif original_type is bool:
self.config[section][key] = value.lower() in ("true", "yes", "1")
else:
self.config[section][key] = value
logger.info(f"从 {config_file} 加载配置")
except Exception as e:
logger.error(f"加载配置文件出错: {e}")
def save_config(self, file_path: str) -> None:
"""将当前配置保存到文件"""
try:
parser = configparser.ConfigParser()
for section, options in self.config.items():
parser.add_section(section)
for key, value in options.items():
parser.set(section, key, str(value))
with open(file_path, 'w') as f:
parser.write(f)
logger.info(f"配置已保存到 {file_path}")
except Exception as e:
logger.error(f"保存配置文件出错: {e}")
def get(self, section: str, key: str, default: Any = None) -> Any:
"""获取配置值，如果不存在则返回默认值"""
try:
return self.config[section][key]
except KeyError:
logger.warning(f"配置 {section}.{key} 不存在，使用默认值: {default}")
return default
class RssReader:
"""RSS订阅内容读取类"""
def __init__(self, config: Config):
"""
初始化RSS读取器
参数:
config: 配置对象
"""
self.config = config
self.cache = {}
self.cache_time = {}
    @retry(max_attempts=3, exceptions=(requests.RequestException,))
def fetch_rss_feed(self, url: Optional[str] = None) -> Optional[bytes]:
"""
获取RSS订阅内容
参数:
url: RSS订阅地址，如果为None则使用配置中的地址
返回:
RSS内容或None（如果获取失败）
"""
url = url or self.config.get("rss", "url")
cache_time = self.config.get("rss", "cache_time")
# 检查缓存
if url in self.cache and url in self.cache_time:
if time.time() - self.cache_time[url] < cache_time:
logger.debug(f"使用缓存的RSS内容: {url}")
return self.cache[url]
try:
logger.info(f"获取RSS订阅: {url}")
response = requests.get(url, timeout=10)
response.raise_for_status()
# 更新缓存
self.cache[url] = response.content
self.cache_time[url] = time.time()
return response.content
except requests.exceptions.RequestException as e:
logger.error(f"获取RSS失败: {e}")
raise
def parse_feed(self, content: Optional[bytes]) -> Optional[feedparser.FeedParserDict]:
"""
解析RSS内容
参数:
content: RSS内容
返回:
解析后的Feed对象或None（如果解析失败）
"""
if not content:
logger.error("没有内容可以解析")
return None
try:
return feedparser.parse(content)
except Exception as e:
logger.error(f"解析RSS内容失败: {e}")
return None
def get_recent_entries(self, feed: Optional[feedparser.FeedParserDict],
days: Optional[int] = None) -> List[FeedEntry]:
"""
获取最近n天的订阅内容
参数:
feed: 解析后的Feed对象
days: 天数，如果为None则使用配置中的值
返回:
最近的条目列表
"""
if not feed:
logger.warning("没有Feed可以获取条目")
return []
days = days or self.config.get("rss", "days")
now = datetime.datetime.now()
cutoff_date = now - timedelta(days=days)
recent_entries = []
logger.info(f"获取最近 {days} 天的订阅内容")
for entry in feed.entries:
# 解析发布日期
pub_date = None
if hasattr(entry, 'published_parsed') and entry.published_parsed:
pub_date = datetime.datetime(*entry.published_parsed[:6])
elif hasattr(entry, 'updated_parsed') and entry.updated_parsed:
pub_date = datetime.datetime(*entry.updated_parsed[:6])
else:
# 如果没有日期信息，跳过该条目
logger.debug(f"跳过没有日期信息的条目: {entry.get('title', 'Unknown')}")
continue
# 只保留最近n天的内容
if pub_date >= cutoff_date:
recent_entries.append({
'title': entry.title,
'link': entry.link,
'published': pub_date.strftime('%Y-%m-%d %H:%M:%S'),
'summary': entry.summary if hasattr(entry, 'summary') else "无摘要",
})
logger.info(f"找到 {len(recent_entries)} 篇最近的文章")
return recent_entries
def display_entries(self, entries: List[FeedEntry]) -> None:
"""
显示条目内容
参数:
entries: 条目列表
"""
if not entries:
logger.info("没有找到最近的文章")
print("没有找到最近的文章")
return
print(f"找到 {len(entries)} 篇最近的文章:")
print("-" * 80)
for i, entry in enumerate(entries, 1):
print(f"{i}. {entry['title']}")
print(f"   发布时间: {entry['published']}")
print(f"   链接: {entry['link']}")
print(f"   摘要: {entry['summary'][:200]}...")  # 只显示部分摘要
print("-" * 80)
class ContentGenerator:
"""内容生成类"""
def __init__(self, config: Config):
"""
初始化内容生成器
参数:
config: 配置对象
"""
self.config = config
self._load_prompt_templates()
def _load_prompt_templates(self) -> None:
"""加载提示词模板"""
self.prompt_templates = {
"daily": """
你是国内顶尖的AI科技公众号编辑，擅长将复杂技术新闻转化为通俗易懂的内容。
请将提供的AI技术新闻整理成一篇微信公众号"每日AI简报"，遵循以下要求：
【内容要求】
1. 使用标题"【每日AI简报】YYYY年MM月DD日"，自动替换为当前日期
2. 开头用2-3句话总结今日AI领域的整体趋势或亮点
3. 为每条新闻设计简短醒目的小标题，形式为"【关键词】+核心内容"
4. 每条新闻包含：
- 事件概述（用最简单的话解释发生了什么）
- 为什么重要（对普通用户或行业的影响）
- 相关背景（如必要，2-3句话解释关键技术概念）
【表达风格】
1. 像"科技博主"而非"新闻记者"的语气，亲切自然
2. 使用生动的类比和比喻解释技术概念
3. 适当使用emoji增强表达（每段1-2个，不要过多）
4. 避免专业术语堆砌，必须使用时提供简明解释
5. 用"你"直接对读者说话，增强亲近感
【格式规范】
1. 通篇采用markdown格式
2. 每条新闻之间用分隔线或明显标题区分
3. 重点信息可用加粗、斜体强调
4. 总篇幅控制在1000-1500字之间
5. 结尾添加"感谢阅读，明天见~"和订阅引导
记住：写作目标是让"对AI感兴趣但没有技术背景的普通用户"轻松理解这些技术进展的价值和意义。
""",
"deep": """
你是一位资深AI领域分析师，擅长深入剖析技术进展和市场影响。
请将提供的AI技术新闻整理成一篇微信公众号"AI技术深度解析"，遵循以下要求：
【内容架构】
1. 开篇：用简明语言概述本期新闻焦点，指出共同趋势或主题
2. 分析框架：将新闻按技术类别或应用领域分组（如LLM进展、多模态、AI应用等）
3. 每则新闻包含：
- 技术本质解析（这项技术/产品的核心机制是什么）
- 进步点评估（与现有技术相比有何突破）
- 行业影响分析（将如何改变相关行业格局）
- 技术路线判断（代表了什么发展方向）
【深度化处理】
1. 剖析核心技术原理，但使用通俗类比让非专业人士理解
2. 关联行业背景和商业模式，解释为何重要
3. 适当引入相关技术发展历史和竞争格局
4. 对技术发展方向做出有见地的推测
【表达规范】
1. 保持客观专业的分析语气，但避免学术化晦涩表达
2. 使用结构化段落和子标题保证清晰度
3. 复杂概念用图示类比或拆解方式解释
4. 适当引用数据或趋势支持分析
最终成文应当让读者不仅了解"发生了什么"，更理解"为什么重要"及"未来走向"，体现你的专业洞察。
""",
"beginner": """
你是一位极擅长技术科普的AI科技博主，你的超能力是把最前沿的AI技术解释得让初中生都能理解。
请将提供的AI技术新闻整理成一篇面向完全零基础读者的微信公众号"AI新手村日报"，遵循以下要求：
【零门槛原则】
1. 假设读者从未接触过AI/ML相关概念，需要从零开始解释
2. 每个技术术语第一次出现时必须立即用括号给出"小白解释"
3. 使用日常生活中的具体例子和类比解释每个概念
4. 把复杂的技术进展转化为"这对你的生活意味着什么"
【内容结构】
1. 开场白：友好问候并用一句话概括"今天AI界发生了什么有趣的事"
2. 新闻主体：每条新闻使用"你知道吗？"或"想象一下"等引导式开头
3. 每则新闻拆解为：
- 这是什么？（用最简单的类比解释）
- 为什么很酷？（用日常场景展示应用）
- 小贴士：提供1-2个延伸知识点，但保持简单
【表达特色】
1. 使用轻松愉快的对话式语气，仿佛朋友间聊天
2. 丰富使用emoji表情和生动比喻
3. 适当加入幽默元素，让技术内容变得有趣
4. 使用"想象一下..."、"就好比..."等引导式表达
5. 问答形式展开解释，预设读者可能的疑问并回答
【视觉辅助】
1. 建议在正文中穿插使用简单示意图的位置标记
2. 关键概念用粗体标记
3. 使用项目符号和短段落提高可读性
记住：如果一个10岁孩子都能听懂你的解释，那你就成功了！
"""
}
def _prepare_input_content(self, entries: List[FeedEntry]) -> str:
"""
准备输入内容
参数:
entries: 条目列表
返回:
格式化的输入内容
"""
input_content = "以下是最近的AI技术新闻动态，请帮我整理成适合微信公众号的每日AI News推送：\n\n"
for entry in entries:
input_content += f"标题: {entry['title']}\n"
input_content += f"时间: {entry['published']}\n"
input_content += f"链接: {entry['link']}\n"
input_content += f"摘要: {entry['summary']}\n\n"
return input_content
    @retry(max_attempts=3, delay=5, exceptions=(requests.RequestException,))
def _call_api(self, messages: List[Dict[str, str]], stream: bool = False) -> Union[str, requests.Response]:
"""
调用API
参数:
messages: 消息列表
stream: 是否流式输出
返回:
生成的内容或流式响应对象
"""
base_url = self.config.get("api", "base_url")
model = self.config.get("api", "model")
api_key = self.config.get("api", "api_key")
timeout = self.config.get("api", "timeout")
max_tokens = self.config.get("api", "max_tokens")
temperature = self.config.get("api", "temperature")
if not api_key:
raise ValueError("API密钥不能为空")
# 构建请求数据
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream
}
# 设置请求头
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
# API 端点
endpoint = f"{base_url}/chat/completions"
logger.info(f"调用 API: {endpoint}, 流式输出: {stream}")
if stream:
response = requests.post(
endpoint,
headers=headers,
json=payload,
stream=True,
timeout=timeout
)
else:
response = requests.post(
endpoint,
headers=headers,
json=payload,
timeout=timeout
)
# 检查响应状态
if response.status_code != 200:
error_msg = f"API 请求失败: 状态码 {response.status_code}, 错误信息: {response.text}"
logger.error(error_msg)
raise requests.RequestException(error_msg)
if stream:
return response
else:
result = response.json()
if "choices" not in result or not result["choices"]:
raise ValueError("API响应格式错误，找不到内容")
return result["choices"][0]["message"]["content"]
def format_with_openai(self, entries: List[FeedEntry], style: str = "daily",
stream: bool = False) -> Optional[str]:
"""
使用OpenAI API对内容进行格式化与润色
参数:
entries: RSS条目列表
style: 输出风格，可选 'daily'(日常简报), 'deep'(深度解析), 'beginner'(小白友好)
stream: 是否使用流式输出
返回:
格式化后的内容或None（如果失败）
"""
if not entries:
logger.warning("没有找到最近的文章可以润色")
return "没有找到最近的文章可以润色"
try:
# 准备输入内容
input_content = self._prepare_input_content(entries)
# 选择对应风格的提示词
system_prompt = self.prompt_templates.get(style, self.prompt_templates["daily"])
# 构建消息
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": input_content}
]
# 调用API
if stream:
# 流式处理
print("正在生成内容，请稍候...")
response = self._call_api(messages, stream=True)
# 处理流式响应
formatted_content = []
client = None
try:
# 先尝试导入sseclient，失败则使用自定义解析
import sseclient
client = sseclient.SSEClient(response)
for event in client.events():
if event.data != "[DONE]":
try:
chunk = json.loads(event.data)
content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
if content:
print(content, end="")
formatted_content.append(content)
except Exception as e:
logger.warning(f"解析事件失败: {e}")
except ImportError:
# 手动解析SSE
logger.info("未安装sseclient，使用自定义SSE解析")
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = line[6:]
if data == "[DONE]":
break
try:
chunk = json.loads(data)
content = chunk.get("choices", [{}])[0].get("delta", {}).get("content", "")
if content:
print(content, end="")
formatted_content.append(content)
except Exception as e:
logger.warning(f"解析事件失败: {e}")
formatted_content = "".join(formatted_content)
print("\n\n生成完成！")
else:
# 非流式处理
formatted_content = self._call_api(messages, stream=False)
# 添加元数据
metadata = {
"source": self.config.get("rss", "url"),
"processed_date": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"article_count": len(entries),
"style": style
}
# 添加元数据到内容顶部作为YAML前置元数据
yaml_metadata = "---\n"
for key, value in metadata.items():
yaml_metadata += f"{key}: {value}\n"
yaml_metadata += "---\n\n"
return yaml_metadata + formatted_content
except Exception as e:
logger.error(f"内容生成失败: {e}")
logger.debug(traceback.format_exc())
return None
def save_to_file(self, content: Optional[str], filename: Optional[str] = None) -> Optional[str]:
"""
将内容保存到文件
参数:
content: 要保存的内容
filename: 文件名，如果为None则使用当前日期生成
返回:
保存的文件路径或None（如果失败）
"""
if not content:
logger.warning("没有内容可保存")
return None
# 确保输出目录存在
output_dir = self.config.get("output", "directory")
os.makedirs(output_dir, exist_ok=True)
# 如果未指定文件名，使用当前日期
if not filename:
today = datetime.datetime.now().strftime("%Y-%m-%d")
filename = f"ai_news_{today}.md"
# 确保文件路径
file_path = os.path.join(output_dir, filename)
try:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
logger.info(f"内容已保存到 {file_path}")
return file_path
except Exception as e:
logger.error(f"保存文件失败: {e}")
return None
def parse_arguments() -> argparse.Namespace:
"""解析命令行参数"""
parser = argparse.ArgumentParser(description="AI News生成器")
parser.add_argument("--config", "-c", type=str, help="配置文件路径")
parser.add_argument("--rss", "-r", type=str, help="RSS订阅地址")
parser.add_argument("--days", "-d", type=int, help="获取最近几天的内容")
parser.add_argument("--style", "-s", type=str, choices=["daily", "deep", "beginner"],
default="daily", help="生成内容的风格")
parser.add_argument("--stream", action="store_true", help="使用流式输出")
parser.add_argument("--output", "-o", type=str, help="输出文件路径")
parser.add_argument("--verbose", "-v", action="store_true", help="显示详细日志")
return parser.parse_args()
def main() -> None:
"""主函数"""
# 解析命令行参数
args = parse_arguments()
# 设置日志级别
if args.verbose:
logger.setLevel(logging.DEBUG)
# 加载配置
config = Config(args.config)
# 如果命令行参数提供了值，则覆盖配置
if args.rss:
config.config["rss"]["url"] = args.rss
if args.days:
config.config["rss"]["days"] = args.days
# 初始化RSS读取器和内容生成器
rss_reader = RssReader(config)
content_generator = ContentGenerator(config)
# 获取RSS内容
rss_url = config.get("rss", "url")
days = config.get("rss", "days")
logger.info(f"开始处理，获取 {rss_url} 最近 {days} 天的内容...")
print(f"正在获取 {rss_url} 最近 {days} 天的内容...")
try:
# 获取并解析RSS
content = rss_reader.fetch_rss_feed()
feed = rss_reader.parse_feed(content)
if not feed:
logger.error("无法解析RSS内容")
print("无法解析RSS内容")
return
# 获取最近的条目
recent_entries = rss_reader.get_recent_entries(feed, days)
rss_reader.display_entries(recent_entries)
if not recent_entries:
logger.warning("没有找到最近的文章")
return
# 如果命令行没有指定风格和流式输出，则交互式询问
style = args.style
stream = args.stream
if not args.style and not sys.argv[1:]:  # 如果没有提供任何命令行参数
print("\n选择内容润色风格:")
print("1. 日常简报风格 (默认，适合一般读者)")
print("2. 深度分析风格 (包含更多技术和市场分析)")
print("3. 小白友好风格 (零基础读者也能轻松理解)")
style_choice = input("请选择 (1-3，默认1): ").strip() or "1"
style_options = {
"1": "daily",
"2": "deep",
"3": "beginner"
}
style = style_options.get(style_choice, "daily")
print("\n是否使用流式输出? (实时显示生成过程)")
print("1. 是 - 实时显示生成过程")
print("2. 否 - 等待完整生成后显示")
stream_choice = input("请选择 (1-2，默认2): ").strip() or "2"
stream = stream_choice == "1"
# 使用OpenAI生成内容
logger.info(f"使用OpenAI进行内容润色 (风格: {style}, 流式输出: {stream})")
print(f"\n正在使用OpenAI进行内容润色 (风格: {style})...")
formatted_content = content_generator.format_with_openai(
recent_entries, style=style, stream=stream
)
if formatted_content:
if not stream:  # 只有非流式处理才需要显示预览
print("\n润色后内容预览 (前500字):")
print("-" * 80)
print(formatted_content[:500] + "...(更多内容已保存到文件)")
print("-" * 80)
# 确定输出文件名
output_file = args.output
if not output_file:
today = datetime.datetime.now().strftime("%Y-%m-%d")
output_file = f"ai_news_{style}_{today}.md"
# 保存到文件
saved_file = content_generator.save_to_file(formatted_content, output_file)
if saved_file:
print(f"完整内容已保存到 {saved_file}")
else:
logger.error("内容生成失败")
print("内容生成失败，请查看日志获取详细信息")
except Exception as e:
logger.error(f"处理过程中出错: {e}")
logger.debug(traceback.format_exc())
print(f"处理过程中出错: {e}")
print("请查看日志获取详细信息")
if __name__ == "__main__":
try:
import sys
main()
except KeyboardInterrupt:
logger.info("用户中断执行")
print("\n程序已中断")
except Exception as e:
logger.critical(f"未捕获的异常: {e}")
logger.debug(traceback.format_exc())
print(f"程序遇到错误: {e}")
sys.exit(1)