Files
my-daily/01主函数过长问题.md
T

25 KiB
Raw Blame History

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_top3equipment_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_windowdeduplicate_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. 主链路

完整业务流程

  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分析失败时,应该使用默认评分和分类,不影响后续流程

验证方法

  • 通过修改配置文件模拟各种边界场景

  • 检查日志输出是否符合预期

  • 检查生成的文件内容是否正确

  • 检查程序是否正常退出或抛出预期的异常