python Google API多处理

qkf9rpyu  于 2023-01-29  发布在  Python
关注(0)|答案(1)|浏览(140)

我正在尝试从Gmail帐户下的电子邮件中获取特定信息(主题,发件人,日期,消息正文),并能够成功地做到这一点使用谷歌API和相关库,然而,我注意到你有越多的电子邮件需要更长的时间来解析,以至于解析34封电子邮件需要近15秒,如果你试图扩展到解析1000封电子邮件,这是很糟糕的。我的目标是在parse_messages()函数上使用并发/多处理,但是,我没有运气,总是返回一个空列表。目标是处理所有的电子邮件,然后将它们全部附加到X1 M1 N1 X列表。
抱歉,太草率了,还没有清理干净,总共不到100行。

from __future__ import print_function
import os.path
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from concurrent.futures import ProcessPoolExecutor
import base64
import re

combined = []

def authenticate():
    # If modifying these scopes, delete the file token.json.
    SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']

    creds = None

    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_file('token.json', SCOPES)

    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(
                'creds.json', SCOPES)
            creds = flow.run_local_server(port=0)

        with open('token.json', 'w') as token:
            token.write(creds.to_json())
    return creds

def get_messages(creds):
    # Get the messages
    days = 31
    service = build('gmail', 'v1', credentials=creds)
    results = service.users().messages().list(userId='me', q=f'newer_than:{days}d, in:inbox').execute()
    messages = results.get('messages', [])
    message_count = len(messages)
    print(f"You've received {message_count} email(s) in the last {days} days")
    if not messages:
        print(f'No Emails found in the last {days} days.')
    return messages

def parse_message(msg):
    # Call the Gmail API
    service = build('gmail', 'v1', credentials=creds)
    txt = service.users().messages().get(userId='me', id=msg['id']).execute()
    payload = txt['payload']
    headers = payload['headers']

    #Grab the Subject Line, From and Date from the Email
    for d in headers:
        if d['name'] == 'Subject':
            subject = d['value']
        if d['name'] == 'From':
            sender = d['value']
            try:
                match = re.search(r'<(.*)>', sender).group(1)
            except:
                match = sender
        if d['name'] == "Date":
            date_received = d['value']

    def get_body(payload):
        if 'body' in payload and 'data' in payload['body']:
            return payload['body']['data']
        elif 'parts' in payload:
            for part in payload['parts']:
                data = get_body(part)
                if data:
                    return data
        else:
            return None

    data = get_body(payload)

    data = data.replace("-","+").replace("_","/")
    decoded_data = base64.b64decode(data).decode("UTF-8")
    decoded_data = (decoded_data.encode('ascii', 'ignore')).decode("UTF-8")
    decoded_data = decoded_data.replace('\n','').replace('\r','').replace('\t', '')

    # Append parsed message to shared list
    return combined.append([date_received, subject, match, decoded_data])

if __name__ == '__main__':
    creds = authenticate()
    messages = get_messages(creds)
    # Create a process pool with 4 worker processes
    with ProcessPoolExecutor(max_workers=4) as executor:
        # Submit the parse_message function for each message in the messages variable
        executor.map(parse_message, messages)
   
    print(f"Combined: {combined}")

当运行脚本时,我的输出通常是。

You've received 34 email(s) in the last 31 days
combined: []
sqxo8psd

sqxo8psd1#

多亏了SimpleApp的帮助,我和其他几个人沿着做了他们的修改,才让这个工作起来。

# Append parsed message to shared list
    return [date_received, subject, match, decoded_data]

if __name__ == '__main__':
    creds = authenticate()
    messages, service = get_messages(creds)
    # Create a process pool with default worker processes
    with ProcessPoolExecutor() as executor:
        combined = []
        # Submit the parse_message function for each message in the messages variable
        all_pools = executor.map(parse_message, messages, [service]*len(messages))
        for e_p in all_pools:
            combined.append(e_p)

相关问题