如何用Python遍历Outlook邮件批量更新指定数据库？

核心思路 #

通过读取数据库的最后同步时间，筛选指定邮件文件夹中该时间点之后的所有目标邮件，批量提取CSV附件并更新数据库，解决中断同步导致的数据缺失问题。

分步实现代码 #

1. 获取数据库最后同步日期 #

从数据库中查询记录的最后同步时间，首次同步则返回一个初始日期：

import sqlite3  # 根据实际数据库替换为pymysql/psycopg2等

def get_last_sync_date():
    conn = sqlite3.connect('your_db.db')
    cursor = conn.cursor()
    # 假设用sync_log表存储最后同步时间，或从业务表取最大更新时间
    cursor.execute("SELECT MAX(sync_time) FROM sync_log")
    last_date = cursor.fetchone()[0]
    conn.close()
    return last_date if last_date else "1970-01-01 00:00:00"

2. 筛选目标日期范围内的邮件 #

以IMAP协议为例，连接邮箱并切换到存放"XYZ"主题邮件的文件夹，搜索发送时间晚于最后同步日期的邮件：

import imaplib
import email
from datetime import datetime

def fetch_target_emails(last_sync_date):
    # 替换为你的邮箱服务器配置
    imap_host = 'imap.yourmail.com'
    email_user = 'your_account@domain.com'
    email_pass = 'your_password_or_app_token'

    mail = imaplib.IMAP4_SSL(imap_host)
    mail.login(email_user, email_pass)
    mail.select('XYZ_Special_Folder')  # 替换为你的目标文件夹名

    # 转换日期为IMAP搜索格式（如1-OCT-2024）
    sync_dt = datetime.strptime(last_sync_date, "%Y-%m-%d %H:%M:%S")
    imap_date = sync_dt.strftime("%d-%b-%Y").upper()
    # 搜索条件：主题为XYZ，发送时间晚于指定日期
    search_rule = f'(SUBJECT "XYZ" SENTSINCE "{imap_date}")'
    _, email_ids = mail.search(None, search_rule)

    target_emails = []
    for e_id in email_ids[0].split():
        _, raw_data = mail.fetch(e_id, '(RFC822)')
        msg = email.message_from_bytes(raw_data[0][1])
        sent_dt = email.utils.parsedate_to_datetime(msg['Date'])
        target_emails.append({'msg': msg, 'sent_time': sent_dt})
    
    mail.logout()
    # 按发送时间排序，保证处理顺序正确
    return sorted(target_emails, key=lambda x: x['sent_time'])

3. 提取CSV附件并批量更新数据库 #

遍历筛选出的邮件，提取CSV内容并批量写入数据库，同时更新同步日志：

import csv
from io import BytesIO

def process_attachments_and_sync_db(emails):
    conn = sqlite3.connect('your_db.db')
    cursor = conn.cursor()

    for item in emails:
        msg = item['msg']
        sent_time = item['sent_time']

        # 遍历邮件部件，提取CSV附件
        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 and filename.lower().endswith('.csv'):
                # 解析CSV内容
                csv_bytes = part.get_payload(decode=True)
                csv_file = BytesIO(csv_bytes)
                reader = csv.DictReader(csv_file)

                # 批量插入/更新数据，根据业务表结构调整字段
                data_rows = []
                for row in reader:
                    data_rows.append((row['field1'], row['field2'], sent_time))
                
                # 使用executemany提升批量处理效率
                cursor.executemany(
                    "INSERT OR REPLACE INTO your_business_table (col1, col2, update_time) VALUES (?, ?, ?)",
                    data_rows
                )
                # 更新同步日志为当前邮件的发送时间
                cursor.execute("REPLACE INTO sync_log (sync_time) VALUES (?)", (sent_time.strftime("%Y-%m-%d %H:%M:%S"),))
                conn.commit()
    
    conn.close()

4. 主执行流程 #

整合上述函数，形成完整的同步逻辑：

def main():
    last_sync = get_last_sync_date()
    target_emails = fetch_target_emails(last_sync)
    if target_emails:
        process_attachments_and_sync_db(target_emails)
        print(f"完成同步：共处理{len(target_emails)}封邮件")
    else:
        print("无需要同步的邮件")

if __name__ == "__main__":
    main()

关键注意事项 #

• 日期兼容性：不同邮箱的发送日期格式可能存在差异，若解析失败可添加异常捕获和格式兼容处理。
• 数据库适配：根据实际使用的数据库类型（MySQL/PostgreSQL等）替换对应的连接和查询代码，务必使用参数化查询防止SQL注入。
• 重复数据处理：使用INSERT OR REPLACE或先查询再更新的逻辑，避免重复数据插入。
• 邮箱权限：部分邮箱（如Gmail）需开启IMAP权限，开启两步验证的需使用应用专用密码登录。

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