Criação de conteúdo orientado por IA: encontrando o equilíbrio entre regulamentação e criatividade

Para muitos criadores de conteúdo, transformar um feed de notícias RSS bruto e entediante em um artigo de opinião claramente estruturado, perspicaz e com estilo consistente é uma tarefa demorada. A enxurrada de informações chega diariamente, mas a luta para transformá-las em conteúdo envolvente costuma ser uma luta entre "movê-las mecanicamente" e "ficar sem ideias".

Neste artigo, exploraremos a ideia de criar uma ferramenta automatizada que torne esse processo eficiente e inteligente. O objetivo principal dessa ferramenta é claro: obter informações não processadas de feeds RSS e transformá-las em artigos de alta qualidade adequados para publicação por meio do processamento de Modelagem de Linguagem Ampla (LLM).

Controle da "temperatura" da criatividade

Um feed RSS é como um mensageiro fiel que fornece informações com precisão, mas carece de interpretação e emoção. Uma boa ferramenta automatizada, por outro lado, deve assumir o papel de um colunista, não apenas apresentando os fatos, mas dando-lhes perspectiva e vida.

A chave para conseguir isso é o controle preciso sobre a criatividade do modelo de IA. Ao ajustar o temperature parâmetro que permite ao usuário manipular o estilo de saída do modelo. Esse é um valor que normalmente varia entre 0 e 1 e é usado para regular a aleatoriedade das previsões do modelo. Mais baixo temperature Um valor como 0,2 torna a saída mais determinística e consistente, o que é adequado para relatórios rigorosos e orientados por fatos, enquanto um valor mais alto (por exemplo, 0,8) estimula a "imaginação" do modelo e produz apresentações mais variadas e criativas, o que é adequado para conteúdo científico popularizado que requer metáforas vívidas e explicações para leigos.

A "trindade" dos assistentes de redação

Para atender às necessidades de diferentes cenários, a ferramenta tem três "personalidades de redação" integradas, cada uma representando um equilíbrio diferente de rigor e criatividade:

• Personalidade do Briefing DiárioSimula o tom amigável de um blogueiro de tecnologia, reunindo notícias diárias sobre IA em um briefing descontraído. Ele usa metáforas realistas e emojis apropriados para aproximar os leitores e, ao mesmo tempo, garantir que a mensagem principal seja precisa.
• Análise aprofundada da personalidadeA perspectiva de um analista do setor é aprofundada na natureza técnica e no impacto sobre o mercado por trás das notícias. Tende a ser mais baixo temperatureO objetivo é garantir que o resultado seja profissional e lógico, ao mesmo tempo em que desenvolve percepções dentro da estrutura.
• Personalidade do tradutor brancoSua principal missão é simplificar. Por exemplo, quando se trata do termo "Mixture of Experts (MoE)" (mistura de especialistas), ele não o joga diretamente no leitor, mas o compara a "uma equipe de especialistas trabalhando juntos, cada um deles responsável por apenas uma pequena parte do problema em que é melhor, a fim de melhorar a eficiência e a precisão gerais", de modo que até mesmo um leitor novato possa entendê-lo facilmente. eficiência e precisão", para que leitores sem conhecimento básico possam entender facilmente.

Essas três personalidades garantem que o resultado possa ser estilisticamente flexível, mas consistente em termos de estrutura do artigo, ajudando os leitores a desenvolver expectativas de leitura estáveis.

Fundamentos técnicos para uma operação robusta

Uma ferramenta realmente útil não pode ser apenas uma simples chamada para uma API, ela requer um conjunto robusto de engenharia por trás para garantir um processo suave e confiável.

• mecanismo de repetição automáticaFlutuações de rede ou falhas temporárias de API são problemas comuns. Com a introdução de um mecanismo de repetição de backoff exponencial, o sistema pode esperar automaticamente por um determinado período de tempo após uma falha antes de tentar novamente, o que melhora muito a taxa de sucesso das tarefas.
• sistema de cachePara feeds RSS que são atualizados com pouca frequência, rastrear o mesmo conteúdo várias vezes é um desperdício de recursos. Um sistema de cache simples pode armazenar temporariamente os dados obtidos e lê-los diretamente da área local em um período de tempo definido, reduzindo efetivamente as solicitações de rede e o tempo de processamento.
• Saída de streamingSaída de streaming: em vez de esperar que o modelo gere todo o conteúdo, a saída de streaming pode exibir o texto gerado literalmente ou frase por frase em tempo real, como em um carregamento de vídeo. Isso não apenas permite que o usuário visualize a qualidade do conteúdo instantaneamente, mas também facilita a interrupção da tarefa se ela estiver fora do curso.
• Configuração flexívelPara se adaptar a diferentes ambientes, a ferramenta suporta configurações por meio de parâmetros de linha de comando, variáveis de ambiente e arquivos de configuração, etc. Os usuários podem ajustar modelos, chaves de API ou diretórios de saída, etc., sem modificar o código-fonte.
• Tratamento abrangente de exceçõesResposta graciosa a surpresas, como chaves de API ausentes, endereços RSS inacessíveis etc. Assegure-se de que o sistema informe claramente os erros em todos os casos, em vez de apenas travar.

Eficácia na prática

Imagine uma época em que Google O lançamento foi chamado de Gemini 2.5 Pro incorporando DeepThink Quando um novo modelo de função de raciocínio é processado pela "Personalidade Tradutora Branca", ele pode ser interpretado como:

Gemini 2.5 Pro É como um parceiro de IA extremamente inteligente que pode ajudá-lo em várias tarefas. E a adição de DeepThink O modo, por outro lado, dá a ele a capacidade de pensar profundamente. Isso significa que ele não apenas responde "o quê", mas também explica "por quê", como um mentor que pode levá-lo ao cerne da questão.

Esse tipo de tratamento mantém o conhecimento técnico e alcança a popularização da expressão. Isso prova que a criatividade não é uma imaginação selvagem, mas uma inovação ordenada sob a estrutura de regras claras, assim como a improvisação do jazz, que parece fluir livremente, mas, na verdade, cada passo não se desviou da trilha da harmonia.

Quando os criadores de conteúdo não precisarem mais fazer uma escolha difícil entre "profundidade técnica" e "ciência popular", o valor da tecnologia poderá ser realmente demonstrado.

#!/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)