728 lines
25 KiB
Markdown
728 lines
25 KiB
Markdown
# 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分析失败时，应该使用默认评分和分类，不影响后续流程
|
||
|
||
**验证方法**：
|
||
|
||
* 通过修改配置文件模拟各种边界场景
|
||
|
||
* 检查日志输出是否符合预期
|
||
|
||
* 检查生成的文件内容是否正确
|
||
|
||
* 检查程序是否正常退出或抛出预期的异常
|
||
|