Python 实现邮件订阅链接自动提取与 AI 分析报告生成

更新时间:2026-10-10 09:59:30 来源:本站整理 浏览:0

架构设计

整个系统由四个核心模块组成:Config(配置加载与校验)、Fetcher(IMAP 拉取邮件)、Parser(多策略 URL 提取)、Analyzer(调用 LLM 分析)以及 Reporter(输出报告)。数据流为:Email → URL → AnalysisResult → Report。模块间高度解耦,可独立替换或扩展,新增解析策略无需改动主流程。

架构图(Mermaid)

架构图

模块职责

  • Config:加载 TOML 配置,校验过滤器规则。
  • Fetcher:通过 IMAP 搜索未读邮件,并按发件人、关键词进行双层过滤。
  • Parser:根据邮件格式选择解析策略,提取链接并去除跟踪参数。
  • Analyzer:构造 Prompt,调用 Gemini CLI 生成结构化分析结果。
  • Reporter:将结果渲染为 Markdown 或 JSON 报告。

核心流程

1. 启动与初始化

输入 CLI 参数和 config.toml,加载配置并初始化各模块,输出 Config 对象。

2. 邮件抓取

根据 FilterConfig 构造 IMAP 搜索条件(如 UNSEEN 和 FROM),再通过 Python 对标题关键词进行二次过滤,最终得到邮件列表。

3. URL 提取

根据邮件内容类型(HTML/Text)选择解析器,提取链接并去除跳转参数(如 redirectUrl、url、target),输出 ExtractedURL[]。

4. 分析阶段

为每个 URL 构造 Prompt,调用 Gemini CLI 生成分析结果,并清洗输出,得到 AnalysisResult。

5. 输出生成

将 URL 与分析结果组合,渲染为 Markdown 或 JSON 报告文件。

关键实现解析

配置系统:类型安全与多过滤器

使用 dataclass 定义 FilterConfig,支持多个过滤器,避免硬编码,并校验重复名称。

@dataclasses.dataclass
class FilterConfig:
    name: str
    sender: str
    title_keywords: List[str]
    max_emails: int
    extract_format: str = "name_url"

for f in filters_data:
    if name in names:
        raise ValueError("Duplicate filter name")

Fetcher:IMAP 与双层过滤

先通过 IMAP 搜索条件粗筛,再用 Python 对标题关键词精筛,减少无效解析。

search_criteria = ["UNSEEN"]
if filter_config.sender:
    search_criteria.extend(["FROM", filter_config.sender])
if not filter_config.title_keywords:
    return True

Parser:策略模式适配多格式

通过 ParserFactory 根据 extract_format 创建对应解析器(如 MediumParser),支持 HTML 与 Text 双路径,并按 base URL 去重。

class ParserFactory:
    def create(extract_format):
        if extract_format == "name_url":
            return MediumParser()

URL 解析与清洗:抗噪核心

解析 query 参数获取真实 URL,并过滤包含噪声关键词的链接,提高数据质量。

redirect_params = ["redirectUrl", "url", "target"]
if any(keyword in lower_title for keyword in _NOISY_TITLE_KEYWORDS):
    return False

Analyzer:LLM 调用工程化

使用临时文件传递参数,避免 shell 转义问题,并实现重试与退避机制,增强稳定性。

with tempfile.NamedTemporaryFile(...) as tmp_user:
    env["GEMINI_SYSTEM_MD"] = tmp_sys_path

增量输出:防止数据丢失

在分析循环中逐步生成报告,确保长任务中断时已处理结果不丢失。

for result in analyzer.analyze_articles(...):
    reporter.generate_summaries_report(...)

通过以上设计,系统实现了从邮件抓取到报告生成的全自动化,具备高扩展性和可靠性。

原文链接:本站整理

本文来自互联网用户投稿,该文章观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。

如若内容造成侵权/违法违规/事实不符,请将相关资料发送至 service@2cto.cc 进行投诉反馈,一经查实,立即处理!