如何基于特定邮件接收自动处理数据？Airflow是否适用？

当然可以实现！你的需求是典型的邮件触发型数据工作流，下面分几个部分帮你理清思路和具体实现方式：

完全合适！Airflow虽然没有内置的「邮件接收传感器」，但它的核心优势就是调度、监控和编排复杂工作流，刚好匹配你这种「周期性+触发式」的任务场景——不管是每周固定时间检查邮件，还是邮件到达后即时处理，都能通过Airflow实现。

这里分两种常见场景给出具体方案：

1. 周期性检查（比如每周一处理上周的邮件） #

你可以用Airflow的PythonOperator直接调用Python邮件读取脚本，或者自定义Sensor来检测目标邮件是否存在。下面是一个完整的示例代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import imaplib
import email
from email.header import decode_header
import os
import psycopg2  # 示例用PostgreSQL，根据你的数据库替换

def check_and_process_email():
    # 邮箱配置（建议存在Airflow的Variables里，不要硬编码）
    username = "your_work_email@company.com"
    password = "your_app_specific_password"  # 用应用密码而非登录密码更安全
    imap_server = "imap.company.com"
    target_sender = "data_sender@company.com"
    target_subject = "Weekly Data Report"
    download_folder = "/tmp/airflow_attachments"
    db_config = {
        "host": "db_host",
        "dbname": "your_db",
        "user": "db_user",
        "password": "db_pwd"
    }

    # 连接IMAP服务器
    imap = imaplib.IMAP4_SSL(imap_server)
    imap.login(username, password)
    imap.select("INBOX")

    # 搜索未读、指定发件人+主题的邮件
    search_query = f'(UNSEEN FROM "{target_sender}" SUBJECT "{target_subject}")'
    status, messages = imap.search(None, search_query)
    message_ids = messages[0].split()

    if message_ids:
        for msg_id in message_ids:
            status, msg_data = imap.fetch(msg_id, "(RFC822)")
            for response_part in msg_data:
                if isinstance(response_part, tuple):
                    msg = email.message_from_bytes(response_part[1])
                    # 解码邮件主题
                    subject, encoding = decode_header(msg["Subject"])[0]
                    subject = subject.decode(encoding) if isinstance(subject, bytes) else subject
                    
                    # 遍历并下载附件
                    for part in msg.walk():
                        if part.get_content_maintype() == 'multipart' or part.get('Content-Disposition') is None:
                            continue
                        filename = part.get_filename()
                        if filename:
                            os.makedirs(download_folder, exist_ok=True)
                            filepath = os.path.join(download_folder, filename)
                            with open(filepath, "wb") as f:
                                f.write(part.get_payload(decode=True))
                            
                            # 这里添加你的数据库导入和数据处理逻辑
                            with psycopg2.connect(**db_config) as conn:
                                with conn.cursor() as cur:
                                    # 示例：读取CSV并导入（根据你的文件格式调整）
                                    with open(filepath, "r") as csv_file:
                                        cur.copy_expert("COPY weekly_data FROM STDIN WITH CSV HEADER", csv_file)
                                    conn.commit()
                            print(f"成功处理邮件附件: {filename}")
            
            # 标记邮件为已读，避免重复处理
            imap.store(msg_id, '+FLAGS', '\\Seen')
    
    imap.close()
    imap.logout()

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': True,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'weekly_email_data_pipeline',
    default_args=default_args,
    description='自动处理每周带数据附件的邮件',
    schedule_interval='0 8 * * 1',  # 每周一早上8点执行
    catchup=False,
) as dag:
    process_email_task = PythonOperator(
        task_id='check_and_process_email',
        python_callable=check_and_process_email,
    )

2. 即时触发（邮件到达后立即处理） #

如果希望邮件一到就处理，有两种高效方式：

• 短间隔轮询：把Sensor的poke_interval设为5分钟左右，持续检查邮箱，发现目标邮件就触发后续任务；
• Webhook推送：如果你的企业邮箱支持Webhook（比如Office 365、Gmail通过Cloud Pub/Sub），可以配置当收到指定发件人的邮件时，自动调用Airflow的DAG触发API，实现零延迟处理。

• IMAP轮询：适合绝大多数场景，无需邮箱额外配置，兼容性强，唯一缺点是有少量延迟（取决于轮询间隔）；
• 邮箱Webhook推送：最优的即时方案，无延迟且资源消耗低，但需要邮箱和Airflow API的额外配置，部分小众邮箱可能不支持。

• 邮件处理后一定要标记为已读，或者移动到指定文件夹，避免重复处理；
• 附件下载后要做格式校验（比如检查是否为预期的CSV/Excel），加入错误捕获逻辑；
• 把敏感配置（邮箱密码、数据库信息）存在Airflow的Variables或Connections里，不要硬编码；
• 给数据处理步骤加入详细日志，方便Airflow监控和排查问题。

内容的提问来源于stack exchange，提问作者lizzie