25 KiB
1. 问题
军事科技每日摘报的 main() 函数(位于 scripts/military_daily_report_v3.py 第38-246行)长达208行,承担了整个业务流程的11个步骤,严重违反单一职责原则。
1.1. 职责过多,难以理解
当前 main() 函数包含了从RSS抓取到报告输出的完整业务流程,涉及数据获取、清洗、翻译、AI分析、格式化等多个不同层次的职责。开发者需要在200多行代码中理解整个业务逻辑,认知负担过重。
问题代码片段(第38-246行):
def main():
now = datetime.now()
cutoff = now - timedelta(hours=TIME_WINDOW_HOURS)
date_str = now.strftime('%Y%m%d')
# 1. RSS 抓取
all_articles = fetch_all_feeds()
# 2. 24小时时间过滤
recent_articles = [a for a in all_articles if a['published'] >= cutoff]
# 3. 标题去重
seen = set()
unique_articles = []
for a in recent_articles:
key = a['title'].strip()
if key not in seen and len(key) > 5:
seen.add(key)
unique_articles.append(a)
# 4. 排序
all_time_valid = sorted(unique_articles, key=lambda x: x['published'], reverse=True)
# 5. 加载AI配置 + 批量翻译
api_key, base_url, model, webzine_model = load_ai_config()
translated_articles = batch_translate_articles(all_time_valid, api_key, base_url, model, max_concurrent=2)
# 6. 关键词筛选
if len(translated_articles) > 15:
filtered_articles = keyword_filter(translated_articles)
else:
filtered_articles = translated_articles
# 7. AI分析
for i, article in enumerate(filtered_articles, 1):
ai_result = score_and_classify_article(article, api_key, base_url, model)
article['ai_result'] = ai_result
article['final_score'] = calculate_weighted_score(ai_result)
article['summary'] = ai_result.get('summary', article['translated_title'])
article['category'] = ai_result.get('category', '装备动态')
article['scores'] = ai_result.get('scores', {})
# 8. 按评分排序
filtered_articles.sort(key=lambda x: x['final_score'], reverse=True)
# 9. 生成TOP3网摘
today_top3 = filtered_articles[:3]
# ... 网摘生成代码
# 10. 分类整理 + 生成分类洞察
# ... 分类整理代码
# 11. 生成Markdown报告
# ... 报告生成代码
问题分析:
-
11个业务步骤混在一个函数中,每个步骤的逻辑细节都暴露在主流程中
-
变量名重复使用(如
filtered_articles),容易产生混淆 -
中间状态(如
today_top3、equipment_articles等)难以追踪 -
无法快速定位某个具体步骤的代码位置
1.2. 难以测试,无法复用
由于所有逻辑都在 main() 函数中,无法对单个业务步骤进行独立的单元测试。例如,想要测试"去重逻辑"或"分类整理逻辑",必须运行整个流程,这大大增加了测试成本和复杂度。
问题分析:
-
去重逻辑(第62-71行)嵌入在主函数中,无法单独测试
-
分类整理逻辑(第175-185行)无法独立验证
-
网摘生成逻辑(第127-159行)与文件IO耦合,难以mock测试
-
想要复用某个步骤(如只做RSS抓取和去重)无法实现
1.3. 维护困难,容易出错
当需要修改某个业务步骤时,开发者必须在200多行代码中定位相关逻辑,容易遗漏或引入错误。例如,如果需要调整"关键词筛选"的触发条件,需要在第88-93行找到相关代码,但很容易在复杂的流程中迷失。
问题分析:
-
修改某个步骤时,需要理解整个上下文
-
容易在修改时影响其他步骤的逻辑
-
代码审查时难以快速定位修改点
-
新人上手成本高,需要阅读大量代码才能理解业务流程
2. 收益
通过将 main() 函数拆分为多个职责明确的子函数,可以显著提升代码的可读性、可测试性和可维护性。
2.1. 提升代码可读性
重构后,每个业务步骤都有独立的函数,函数名清晰表达了其职责。主函数变成一个简洁的流程编排器,一眼就能看出整个业务流程的11个步骤。
预期改进:
-
主函数从208行缩减到约30行
-
每个子函数平均15-30行,职责单一
-
函数名直接表达业务含义(如
filter_by_time_window、deduplicate_articles) -
代码结构清晰,新人可以在5分钟内理解整体流程
2.2. 提升可测试性
每个业务步骤都可以独立进行单元测试,无需运行整个流程。可以mock输入数据,验证每个步骤的输出是否符合预期。
预期改进:
-
可以为每个子函数编写独立的单元测试
-
测试覆盖率可以从当前的接近0%提升到80%以上
-
测试执行时间从分钟级降低到秒级
-
可以快速定位和修复bug
2.3. 提升可维护性
当需要修改某个业务步骤时,只需要定位到对应的子函数,修改范围明确,不会影响其他步骤。代码审查时也可以快速定位修改点。
预期改进:
-
修改某个步骤时,只需要关注对应的子函数
-
代码审查时可以快速定位修改的函数
-
降低引入新bug的风险
-
提高开发效率
2.4. 提升代码复用性
业务步骤可以独立使用,例如只做RSS抓取和去重,或者只做AI分析。这为后续的功能扩展提供了灵活性。
预期改进:
-
可以单独调用某个业务步骤
-
可以组合不同的步骤实现新的功能
-
为后续的功能扩展提供基础
3. 方案
将 main() 函数拆分为11个职责明确的子函数,每个函数负责一个独立的业务步骤。主函数变成一个简洁的流程编排器,负责调用这些子函数。
3.1. 函数拆分设计
flowchart TD
A[main] --> B[fetch_and_filter_articles]
B --> C[translate_articles]
C --> D[filter_by_keywords]
D --> E[analyze_articles]
E --> F[generate_webzine]
F --> G[generate_report]
B --> B1[fetch_all_feeds]
B --> B2[filter_by_time_window]
B --> B3[deduplicate_articles]
B --> B4[sort_by_publish_time]
E --> E1[ai_score_and_classify]
E --> E2[sort_by_score]
G --> G1[categorize_articles]
G --> G2[generate_category_overviews]
G --> G3[write_markdown_report]
style A fill:#e1f5e1
style B fill:#fff4e1
style C fill:#fff4e1
style D fill:#fff4e1
style E fill:#fff4e1
style F fill:#fff4e1
style G fill:#fff4e1
图表说明:
-
绿色节点表示主函数,负责流程编排
-
黄色节点表示拆分后的子函数,每个函数负责一个独立的业务步骤
-
箭头表示调用关系
-
通过这种拆分,主函数变得简洁,每个子函数职责单一
3.2. 函数拆分实现
步骤1:提取数据获取和过滤逻辑
问题代码:
def main():
# 1. RSS 抓取
all_articles = fetch_all_feeds()
# 2. 24小时时间过滤
recent_articles = [a for a in all_articles if a['published'] >= cutoff]
# 3. 标题去重
seen = set()
unique_articles = []
for a in recent_articles:
key = a['title'].strip()
if key not in seen and len(key) > 5:
seen.add(key)
unique_articles.append(a)
# 4. 排序
all_time_valid = sorted(unique_articles, key=lambda x: x['published'], reverse=True)
重构后代码:
def filter_by_time_window(articles, cutoff):
"""过滤出指定时间窗口内的文章"""
return [a for a in articles if a['published'] >= cutoff]
def deduplicate_articles(articles):
"""根据标题去重文章"""
seen = set()
unique_articles = []
for a in articles:
key = a['title'].strip()
if key not in seen and len(key) > 5:
seen.add(key)
unique_articles.append(a)
return unique_articles
def sort_by_publish_time(articles, reverse=True):
"""按发布时间排序文章"""
return sorted(articles, key=lambda x: x['published'], reverse=reverse)
def fetch_and_filter_articles(cutoff):
"""获取并过滤文章:抓取 -> 时间过滤 -> 去重 -> 排序"""
all_articles = fetch_all_feeds()
recent_articles = filter_by_time_window(all_articles, cutoff)
unique_articles = deduplicate_articles(recent_articles)
return sort_by_publish_time(unique_articles)
改进点:
-
每个函数职责单一,函数名清晰表达其功能
-
可以独立测试每个步骤
-
可以单独复用某个步骤
步骤2:提取翻译和关键词筛选逻辑
问题代码:
def main():
# 5. 加载AI配置 + 批量翻译
api_key, base_url, model, webzine_model = load_ai_config()
translated_articles = batch_translate_articles(all_time_valid, api_key, base_url, model, max_concurrent=2)
# 6. 关键词筛选
if len(translated_articles) > 15:
filtered_articles = keyword_filter(translated_articles)
else:
filtered_articles = translated_articles
重构后代码:
def filter_by_keywords(articles):
"""根据关键词筛选文章(仅当文章数 > 15 时执行)"""
if len(articles) > 15:
return keyword_filter(articles)
return articles
def translate_articles(articles, api_key, base_url, model):
"""批量翻译外文文章"""
return batch_translate_articles(articles, api_key, base_url, model, max_concurrent=2)
改进点:
-
将关键词筛选的条件判断封装在函数内部
-
翻译逻辑独立,可以单独测试
步骤3:提取AI分析和排序逻辑
问题代码:
def main():
# 7. AI分析
for i, article in enumerate(filtered_articles, 1):
ai_result = score_and_classify_article(article, api_key, base_url, model)
article['ai_result'] = ai_result
article['final_score'] = calculate_weighted_score(ai_result)
article['summary'] = ai_result.get('summary', article['translated_title'])
article['category'] = ai_result.get('category', '装备动态')
article['scores'] = ai_result.get('scores', {})
# 8. 按评分排序
filtered_articles.sort(key=lambda x: x['final_score'], reverse=True)
重构后代码:
def ai_score_and_classify(article, api_key, base_url, model):
"""对单篇文章进行AI分析:评分、摘要、分类"""
ai_result = score_and_classify_article(article, api_key, base_url, model)
article['ai_result'] = ai_result
article['final_score'] = calculate_weighted_score(ai_result)
article['summary'] = ai_result.get('summary', article['translated_title'])
article['category'] = ai_result.get('category', '装备动态')
article['scores'] = ai_result.get('scores', {})
return article
def analyze_articles(articles, api_key, base_url, model):
"""对所有文章进行AI分析并按评分排序"""
for i, article in enumerate(articles, 1):
logger.info("处理中 %d/%d: %s...", i, len(articles), article['translated_title'][:35])
ai_score_and_classify(article, api_key, base_url, model)
# 按评分排序
return sorted(articles, key=lambda x: x['final_score'], reverse=True)
改进点:
-
将单篇文章的AI分析和批量分析分离
-
排序逻辑封装在函数内部
-
可以单独测试AI分析逻辑
步骤4:提取网摘生成逻辑
问题代码:
def main():
# 9. 提取TOP3 + 生成《参考消息》风格网摘
today_top3 = filtered_articles[:3]
webzine_content = []
webzine_content.append(f"═{'═'*60}═")
webzine_content.append(f" 🔥 军事科技每日摘报 - 今日必看 TOP3 网摘")
webzine_content.append(f" 📅 {now.strftime('%Y年%m月%d日')}")
webzine_content.append(f"═{'═'*60}═")
webzine_content.append("")
webzine_texts = []
for i, article in enumerate(today_top3, 1):
webzine = generate_webzine_for_article(article, api_key, base_url, webzine_model)
webzine_texts.append(webzine)
webzine_content.append(f"{'='*62}")
webzine_content.append(f" 📌 第 {i} 篇")
webzine_content.append(f"{'='*62}")
webzine_content.append("")
webzine_content.append(webzine)
webzine_content.append("")
webzine_content.append(f"🔗 原文链接:{article['link']}")
webzine_content.append("")
webzine_content.append("")
# 保存网摘TXT文件
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
webzine_file = OUTPUT_DIR / f"military_webzine_{date_str}.txt"
with open(webzine_file, 'w', encoding='utf-8') as f:
f.write('\n'.join(webzine_content))
# 生成合并长图
webzine_image = OUTPUT_DIR / f"military_webzine_{date_str}.png"
create_combined_webzine_image(webzine_texts, str(webzine_image), now)
重构后代码:
def generate_webzine_content(top3_articles, now):
"""生成网摘文本内容"""
webzine_content = []
webzine_content.append(f"═{'═'*60}═")
webzine_content.append(f" 🔥 军事科技每日摘报 - 今日必看 TOP3 网摘")
webzine_content.append(f" 📅 {now.strftime('%Y年%m月%d日')}")
webzine_content.append(f"═{'═'*60}═")
webzine_content.append("")
webzine_texts = []
for i, article in enumerate(top3_articles, 1):
webzine = generate_webzine_for_article(article, api_key, base_url, webzine_model)
webzine_texts.append(webzine)
webzine_content.append(f"{'='*62}")
webzine_content.append(f" 📌 第 {i} 篇")
webzine_content.append(f"{'='*62}")
webzine_content.append("")
webzine_content.append(webzine)
webzine_content.append("")
webzine_content.append(f"🔗 原文链接:{article['link']}")
webzine_content.append("")
webzine_content.append("")
return webzine_content, webzine_texts
def save_webzine_files(webzine_content, webzine_texts, date_str, now):
"""保存网摘文件(TXT和PNG)"""
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# 保存TXT文件
webzine_file = OUTPUT_DIR / f"military_webzine_{date_str}.txt"
with open(webzine_file, 'w', encoding='utf-8') as f:
f.write('\n'.join(webzine_content))
logger.info("网摘文件已保存: %s", webzine_file)
# 生成PNG文件
webzine_image = OUTPUT_DIR / f"military_webzine_{date_str}.png"
create_combined_webzine_image(webzine_texts, str(webzine_image), now)
def generate_webzine(articles, api_key, base_url, webzine_model, date_str, now):
"""生成TOP3网摘并保存文件"""
top3 = articles[:3]
webzine_content, webzine_texts = generate_webzine_content(top3, now)
save_webzine_files(webzine_content, webzine_texts, date_str, now)
改进点:
-
将网摘内容生成和文件保存分离
-
可以独立测试网摘内容生成逻辑
-
文件IO逻辑集中,易于mock
步骤5:提取分类整理和报告生成逻辑
问题代码:
def main():
# 10. 分类整理 + 生成各分类深度洞察
equipment_articles = []
conflict_articles = []
strategy_articles = []
for article in filtered_articles[3:]:
cat = article.get('category', '装备动态')
if '冲突' in cat or '战争' in cat or '地区' in cat:
conflict_articles.append(article)
elif '战略' in cat or '政策' in cat or '外交' in cat:
strategy_articles.append(article)
else:
equipment_articles.append(article)
today_overview = generate_category_overview("今日必看", today_top3, api_key, base_url, model)
equipment_overview = generate_category_overview("装备动态", equipment_articles, api_key, base_url, model)
conflict_overview = generate_category_overview("地区冲突", conflict_articles, api_key, base_url, model)
strategy_overview = generate_category_overview("战略政策", strategy_articles, api_key, base_url, model)
# 11. 生成 Markdown 报告文件
md_lines = []
md_lines.append(f"# 🔥 {now.year}年{now.month}月{now.day}日 军事科技每日摘报\n")
# ... 大量Markdown拼接代码
重构后代码:
def categorize_articles(articles):
"""将文章按分类整理"""
equipment_articles = []
conflict_articles = []
strategy_articles = []
for article in articles:
cat = article.get('category', '装备动态')
if '冲突' in cat or '战争' in cat or '地区' in cat:
conflict_articles.append(article)
elif '战略' in cat or '政策' in cat or '外交' in cat:
strategy_articles.append(article)
else:
equipment_articles.append(article)
return equipment_articles, conflict_articles, strategy_articles
def generate_category_overviews(top3, equipment, conflict, strategy, api_key, base_url, model):
"""生成各分类的总体介绍"""
today_overview = generate_category_overview("今日必看", top3, api_key, base_url, model)
equipment_overview = generate_category_overview("装备动态", equipment, api_key, base_url, model)
conflict_overview = generate_category_overview("地区冲突", conflict, api_key, base_url, model)
strategy_overview = generate_category_overview("战略政策", strategy, api_key, base_url, model)
return {
'today': today_overview,
'equipment': equipment_overview,
'conflict': conflict_overview,
'strategy': strategy_overview
}
def write_markdown_report(articles, overviews, date_str, now):
"""生成Markdown报告文件"""
top3 = articles[:3]
equipment, conflict, strategy = categorize_articles(articles[3:])
md_lines = []
md_lines.append(f"# 🔥 {now.year}年{now.month}月{now.day}日 军事科技每日摘报\n")
md_lines.append(f"📊 今日共 {len(articles)} 篇精选文章 | ⏰ {now.strftime('%Y-%m-%d %H:%M')}\n")
# 今日必看
md_lines.append(f"## 🔥 今日必看\n")
if overviews['today']:
md_lines.append(f"> {overviews['today']}\n")
for i, article in enumerate(top3, 1):
md_lines.append(f"### {i}. {article['translated_title']}\n")
md_lines.append(f"- 📡 来源: {article['source']}")
md_lines.append(f"- ⏰ 时间: {article['published'].strftime('%m-%d %H:%M')}")
md_lines.append(f"- ⭐ 评分: {article['final_score']}/10")
md_lines.append(f"- 📝 {article['summary']}")
md_lines.append(f"- 🔗 [阅读原文]({article['link']})\n")
# 装备动态
if equipment:
md_lines.append(f"## ⚙️ 装备动态\n")
if overviews['equipment']:
md_lines.append(f"> {overviews['equipment']}\n")
for i, article in enumerate(equipment, 1):
md_lines.append(f"### {i}. {article['translated_title']}\n")
md_lines.append(f"- 📡 {article['source']} | ⏰ {article['published'].strftime('%m-%d %H:%M')}")
md_lines.append(f"- 📝 {article['summary']}")
md_lines.append(f"- 🔗 [阅读原文]({article['link']})\n")
# 地区冲突
if conflict:
md_lines.append(f"## 💥 地区冲突\n")
if overviews['conflict']:
md_lines.append(f"> {overviews['conflict']}\n")
for i, article in enumerate(conflict, 1):
md_lines.append(f"### {i}. {article['translated_title']}\n")
md_lines.append(f"- 📡 {article['source']} | ⏰ {article['published'].strftime('%m-%d %H:%M')}")
md_lines.append(f"- 📝 {article['summary']}")
md_lines.append(f"- 🔗 [阅读原文]({article['link']})\n")
# 战略政策
if strategy:
md_lines.append(f"## 🎯 战略政策\n")
if overviews['strategy']:
md_lines.append(f"> {overviews['strategy']}\n")
for i, article in enumerate(strategy, 1):
md_lines.append(f"### {i}. {article['translated_title']}\n")
md_lines.append(f"- 📡 {article['source']} | ⏰ {article['published'].strftime('%m-%d %H:%M')}")
md_lines.append(f"- 📝 {article['summary']}")
md_lines.append(f"- 🔗 [阅读原文]({article['link']})\n")
# 保存文件
report_file = OUTPUT_DIR / f"military_report_{date_str}.md"
with open(report_file, 'w', encoding='utf-8') as f:
f.write('\n'.join(md_lines))
logger.info("报告文件已保存: %s", report_file)
def generate_report(articles, api_key, base_url, model, date_str, now):
"""生成完整报告:分类整理 -> 生成分类洞察 -> 写入Markdown"""
equipment, conflict, strategy = categorize_articles(articles[3:])
top3 = articles[:3]
overviews = generate_category_overviews(
top3, equipment, conflict, strategy, api_key, base_url, model
)
write_markdown_report(articles, overviews, date_str, now)
改进点:
-
将分类整理、分类洞察生成、报告写入分离
-
每个函数职责单一,易于测试
-
Markdown生成逻辑集中,易于维护
步骤6:重构后的主函数
重构后的主函数:
def main():
"""主函数:流程编排"""
now = datetime.now()
cutoff = now - timedelta(hours=TIME_WINDOW_HOURS)
date_str = now.strftime('%Y%m%d')
logger.info("开始抓取 %d 个军事科技 RSS 订阅源...", len(RSS_FEEDS))
logger.info("时间窗口: %s ~ %s", cutoff.strftime('%m-%d %H:%M'), now.strftime('%m-%d %H:%M'))
# 1. 获取并过滤文章
articles = fetch_and_filter_articles(cutoff)
logger.info("总共抓取 %d 篇文章", len(articles))
if not articles:
logger.info("今日暂无新的军事科技文章")
return
# 2. 翻译文章
api_key, base_url, model, webzine_model = load_ai_config()
if not api_key:
logger.error("未找到OPENAI_API_KEY,退出")
return
translated_articles = translate_articles(articles, api_key, base_url, model)
logger.info("翻译完成,共 %d 篇文章", len(translated_articles))
# 3. 关键词筛选
filtered_articles = filter_by_keywords(translated_articles)
logger.info("筛选后: %d 篇", len(filtered_articles))
if not filtered_articles:
logger.info("筛选后无符合条件文章")
return
# 4. AI分析
logger.info("开始 AI 智能分析(使用 %s)...", model)
analyzed_articles = analyze_articles(filtered_articles, api_key, base_url, model)
logger.info("AI 评分完成,最高分: %.1f/10", analyzed_articles[0]['final_score'])
# 5. 生成网摘
logger.info("生成今日必看 TOP3 网摘(《参考消息》风格)...")
generate_webzine(analyzed_articles, api_key, base_url, webzine_model, date_str, now)
# 6. 生成报告
logger.info("生成 Markdown 报告文件...")
generate_report(analyzed_articles, api_key, base_url, model, date_str, now)
改进点:
-
主函数从208行缩减到约50行
-
流程清晰,一眼就能看出整个业务流程
-
每个步骤都有清晰的注释
-
易于理解和维护
4. 回归范围
本次重构主要是代码结构调整,不改变业务逻辑,因此回归测试的重点是验证重构后的功能与重构前完全一致。
4.1. 主链路
完整业务流程:
- RSS抓取 → 时间过滤 → 去重 → 排序 → 翻译 → 关键词筛选 → AI分析 → 评分排序 → 网摘生成 → 报告输出
关键验证点:
-
RSS抓取能够正常获取12个订阅源的文章
-
时间过滤能够正确过滤出24小时内的文章
-
去重逻辑能够正确去除重复标题的文章
-
翻译功能能够正确翻译外文文章
-
关键词筛选在文章数 > 15 时能够正确执行
-
AI分析能够正确生成评分、摘要和分类
-
网摘生成能够正确生成TOP3网摘的TXT和PNG文件
-
报告生成能够正确生成Markdown报告文件
4.2. 边界情况
边界场景:
- 无新文章场景:当24小时内没有新文章时,程序应该正常退出并输出日志
- 筛选后无文章场景:当关键词筛选后没有符合条件的文章时,程序应该正常退出并输出日志
- AI配置缺失场景:当未配置OPENAI_API_KEY时,程序应该正常退出并输出错误日志
- 文章数 <= 15场景:当文章数 <= 15时,应该跳过关键词筛选,直接进入AI分析
- 分类为空场景:当某个分类(如"地区冲突")没有文章时,报告中应该不显示该分类
- 翻译失败场景:当翻译失败时,应该使用原标题和原文内容,不影响后续流程
- AI分析失败场景:当AI分析失败时,应该使用默认评分和分类,不影响后续流程
验证方法:
-
通过修改配置文件模拟各种边界场景
-
检查日志输出是否符合预期
-
检查生成的文件内容是否正确
-
检查程序是否正常退出或抛出预期的异常