如何通过OAuth认证从Gmail提取CSV并合并为Python DataFrame

问题分析 #

你遇到的核心问题包括：原ChatGPT代码缺少必要库导入、日期查询格式不符合Gmail API要求、未处理邮件列表分页、未递归处理嵌套邮件部分，以及使用了已弃用的DataFrame.append方法。以下是修正后的完整可运行实现。

完整解决方案代码 #

import datetime
import base64
import io
import pandas as pd
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

def get_message_ids(service, start_date, end_date):
    # 转换为Gmail查询支持的YYYY/MM/DD格式
    start_str = datetime.datetime.strptime(start_date, '%Y-%m-%d').strftime('%Y/%m/%d')
    end_str = datetime.datetime.strptime(end_date, '%Y-%m-%d').strftime('%Y/%m/%d')
    # 组合查询条件：日期范围 + CSV附件过滤
    query = f'after:{start_str} before:{end_str} has:attachment filename:csv'
    
    message_ids = []
    page_token = None
    # 处理邮件列表分页
    while True:
        try:
            results = service.users().messages().list(
                userId='me',
                q=query,
                pageToken=page_token
            ).execute()
            message_ids.extend([msg['id'] for msg in results.get('messages', [])])
            page_token = results.get('nextPageToken')
            if not page_token:
                break
        except HttpError as error:
            print(f'获取邮件列表失败: {error}')
            break
    return message_ids

def extract_csv_attachments(service, message_id):
    csv_attachments = []
    try:
        # 获取完整邮件内容（包含payload结构）
        message = service.users().messages().get(
            userId='me',
            id=message_id,
            format='full'
        ).execute()
        payload = message.get('payload', {})
        parts = payload.get('parts', [])
        
        # 递归遍历所有邮件部分，处理嵌套附件
        def traverse_parts(parts_list):
            for part in parts_list:
                # 判断是否为CSV附件且存在附件ID
                if part.get('filename', '').endswith('.csv') and part.get('body', {}).get('attachmentId'):
                    attachment = service.users().messages().attachments().get(
                        userId='me',
                        messageId=message_id,
                        id=part['body']['attachmentId']
                    ).execute()
                    # 解码URL安全的Base64数据
                    file_data = base64.urlsafe_b64decode(attachment['data'].encode('UTF-8'))
                    csv_attachments.append((part['filename'], file_data))
                # 递归处理子部分
                elif part.get('parts'):
                    traverse_parts(part['parts'])
        
        traverse_parts(parts)
    except HttpError as error:
        print(f'处理邮件ID {message_id} 失败: {error}')
    return csv_attachments

def merge_csv_to_dataframe(csv_attachments):
    df_list = []
    for filename, data in csv_attachments:
        try:
            temp_df = pd.read_csv(io.BytesIO(data))
            # 可选：添加来源文件名字段，便于数据溯源
            temp_df['source_file'] = filename
            df_list.append(temp_df)
        except Exception as e:
            print(f'解析CSV文件 {filename} 失败: {e}')
    # 使用concat替代已弃用的append方法
    return pd.concat(df_list, ignore_index=True) if df_list else pd.DataFrame()

def main(start_date, end_date):
    # 加载OAuth2凭证
    creds = Credentials.from_authorized_user_file('token.json')
    service = build('gmail', 'v1', credentials=creds)
    
    # 获取符合条件的邮件ID列表
    message_ids = get_message_ids(service, start_date, end_date)
    if not message_ids:
        print('未找到符合条件的邮件')
        return pd.DataFrame()
    
    # 提取所有CSV附件
    all_csvs = []
    for msg_id in message_ids:
        all_csvs.extend(extract_csv_attachments(service, msg_id))
    
    # 合并为单一DataFrame
    return merge_csv_to_dataframe(all_csvs)

# 调用示例
if __name__ == '__main__':
    merged_data = main('2024-01-01', '2024-06-30')
    print(merged_data.head())
    # 可选：保存合并结果到本地
    # merged_data.to_csv('merged_gmail_data.csv', index=False)

关键修正说明 #

1. 日期格式适配：Gmail API的after/before查询参数仅支持YYYY/MM/DD格式，替换原代码的ISO格式避免查询失效。
2. 分页处理：邮件列表超过100条时会分页，通过循环获取nextPageToken确保拿到所有目标邮件。
3. 嵌套附件处理：部分邮件的附件嵌套在多级parts中，通过递归遍历确保不遗漏任何CSV附件。
4. 弃用方法替换：用pd.concat替代已弃用的DataFrame.append，提升性能并避免警告。
5. 错误捕获：添加HttpError捕获API调用异常，以及CSV解析异常，增强代码稳定性。

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