Python新手求助：编写脚本读取Gmail特定发件人邮件并导出至CSV

Hey there! Since you're new to Python, let's walk through exactly how to build a script that reads your Gmail, filters emails from a specific sender, extracts the subject, currency pair data, and exports everything to a CSV file. Let's break this down step by step.

1. Prerequisites #

First off, you'll need to set up a few things before writing code:

• Install required libraries: Run these commands in your terminal:

  pip install google-api-python-client oauth2client python-dotenv
  

• Set up Gmail API access:
  • Head to the Google Cloud Console, create a new project, and enable the Gmail API.
  • Create OAuth 2.0 credentials (choose the "Desktop app" type), download the credentials.json file, and save it in your script's folder.
  • When you run the script for the first time, it'll open a browser window to authenticate your Google account—follow the prompts, and a token.json file will be created for future runs so you don't have to log in every time.

2. Full Script with Detailed Comments #

I've added plenty of comments so you can follow along easily:

import os
import re
import csv
from googleapiclient.discovery import build
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request

# Define the Gmail API access scope (we only need read access)
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']

def get_gmail_service():
    """Authenticate your account and return a Gmail API service object"""
    creds = None
    # Load saved authentication token if it exists
    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_file('token.json', SCOPES)
    # Refresh or re-authenticate if credentials are invalid/expired
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)
            creds = flow.run_local_server(port=0)
        # Save the new token for next time
        with open('token.json', 'w') as token:
            token.write(creds.to_json())
    # Build and return the service
    return build('gmail', 'v1', credentials=creds)

def extract_email_data(message):
    """Pull subject, currency pairs, prices, and timestamp from the email"""
    # Grab the email subject from headers
    headers = message['payload']['headers']
    subject = next(h['value'] for h in headers if h['name'] == 'Subject')
    
    # Extract the plain text body (matches your example email format)
    body = ""
    if 'parts' in message['payload']:
        for part in message['payload']['parts']:
            if part['mimeType'] == 'text/plain':
                # Decode the base64-encoded body content
                body = part['body']['data'].replace('-', '+').replace('_', '/')
                body = bytes(body, 'utf-8').decode('base64')
                break
    else:
        # Handle emails with no separate parts
        body = message['payload']['body']['data'].replace('-', '+').replace('_', '/')
        body = bytes(body, 'utf-8').decode('base64')
    
    # Use regex to pull currency pairs and their prices
    currency_pattern = re.compile(r'([A-Z]{3}/[A-Z]{3}) Price: (\d+\.\d+)')
    currency_data = currency_pattern.findall(body)
    
    # Extract the timestamp
    time_pattern = re.compile(r'Time: (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} \(CET\))')
    time_match = time_pattern.search(body)
    time = time_match.group(1) if time_match else "Timestamp not found"
    
    return {
        'subject': subject,
        'currency_pairs': currency_data,
        'time': time
    }

def export_to_csv(email_data_list, filename='currency_prices.csv'):
    """Write all extracted data to a CSV file"""
    with open(filename, 'w', newline='', encoding='utf-8') as csvfile:
        # Define CSV columns
        fieldnames = ['Subject', 'Currency Pair', 'Price', 'Time']
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        
        writer.writeheader()
        # Write a row for each currency pair in every email
        for email_data in email_data_list:
            subject = email_data['subject']
            time = email_data['time']
            for pair, price in email_data['currency_pairs']:
                writer.writerow({
                    'Subject': subject,
                    'Currency Pair': pair,
                    'Price': price,
                    'Time': time
                })
    print(f"Success! Data exported to {filename}")

def main():
    # Replace this with the sender email you want to filter
    target_sender = 'your-target-sender@example.com'
    
    # Get the authenticated Gmail service
    service = get_gmail_service()
    
    # Search Gmail for emails from the target sender
    results = service.users().messages().list(
        userId='me',
        q=f'from:{target_sender}'
    ).execute()
    messages = results.get('messages', [])
    
    if not messages:
        print(f"No emails found from {target_sender}.")
        return
    
    # Collect data from all matching emails
    email_data_list = []
    for msg in messages:
        message = service.users().messages().get(userId='me', id=msg['id'], format='full').execute()
        email_data = extract_email_data(message)
        email_data_list.append(email_data)
    
    # Export to CSV
    export_to_csv(email_data_list)

if __name__ == '__main__':
    main()

3. Quick Tips for You #

• Update the sender email: Make sure to replace your-target-sender@example.com in the main() function with the actual email address you want to filter.
• Adjust regex if needed: The regex patterns are built for your specific example email. If the sender changes their email format later, you'll need to tweak the regex to match the new structure.
• Handle edge cases: This script has basic error handling. As you get more comfortable with Python, you can add try-except blocks to handle things like missing email headers or failed API calls.

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