# 1. 问题 军事科技每日摘报的 `main()` 函数(位于 `scripts/military_daily_report_v3.py` 第38-246行)长达208行,承担了整个业务流程的11个步骤,严重违反单一职责原则。 ## 1.1. **职责过多,难以理解** 当前 `main()` 函数包含了从RSS抓取到报告输出的完整业务流程,涉及数据获取、清洗、翻译、AI分析、格式化等多个不同层次的职责。开发者需要在200多行代码中理解整个业务逻辑,认知负担过重。 **问题代码片段(第38-246行)**: ```python 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. **函数拆分设计** ```mermaid 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:提取数据获取和过滤逻辑** **问题代码**: ```python 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) ``` **重构后代码**: ```python 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:提取翻译和关键词筛选逻辑** **问题代码**: ```python 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 ``` **重构后代码**: ```python 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分析和排序逻辑** **问题代码**: ```python 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) ``` **重构后代码**: ```python 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:提取网摘生成逻辑** **问题代码**: ```python 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) ``` **重构后代码**: ```python 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:提取分类整理和报告生成逻辑** **问题代码**: ```python 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拼接代码 ``` **重构后代码**: ```python 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:重构后的主函数** **重构后的主函数**: ```python 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. 主链路 **完整业务流程**: 1. RSS抓取 → 时间过滤 → 去重 → 排序 → 翻译 → 关键词筛选 → AI分析 → 评分排序 → 网摘生成 → 报告输出 **关键验证点**: * RSS抓取能够正常获取12个订阅源的文章 * 时间过滤能够正确过滤出24小时内的文章 * 去重逻辑能够正确去除重复标题的文章 * 翻译功能能够正确翻译外文文章 * 关键词筛选在文章数 > 15 时能够正确执行 * AI分析能够正确生成评分、摘要和分类 * 网摘生成能够正确生成TOP3网摘的TXT和PNG文件 * 报告生成能够正确生成Markdown报告文件 ## 4.2. 边界情况 **边界场景**: 1. **无新文章场景**:当24小时内没有新文章时,程序应该正常退出并输出日志 2. **筛选后无文章场景**:当关键词筛选后没有符合条件的文章时,程序应该正常退出并输出日志 3. **AI配置缺失场景**:当未配置OPENAI\_API\_KEY时,程序应该正常退出并输出错误日志 4. **文章数 <= 15场景**:当文章数 <= 15时,应该跳过关键词筛选,直接进入AI分析 5. **分类为空场景**:当某个分类(如"地区冲突")没有文章时,报告中应该不显示该分类 6. **翻译失败场景**:当翻译失败时,应该使用原标题和原文内容,不影响后续流程 7. **AI分析失败场景**:当AI分析失败时,应该使用默认评分和分类,不影响后续流程 **验证方法**: * 通过修改配置文件模拟各种边界场景 * 检查日志输出是否符合预期 * 检查生成的文件内容是否正确 * 检查程序是否正常退出或抛出预期的异常