قراءة رسائل البريد وإرسالها برمجيًا

عرض الرسائل باستخدام Gmail API

def list_messages(service, user_id='me', query='', max_results=10):
    results = service.users().messages().list(
        userId=user_id,
        q=query,          # Gmail search query
        maxResults=max_results
    ).execute()

    messages = results.get('messages', [])
    print(f'Found {len(messages)} messages')
    return messages

# Examples of Gmail search queries:
# 'is:unread' — unread messages
# 'from:boss@company.com is:unread' — unread from boss
# 'subject:invoice label:inbox' — invoices in inbox
# 'after:2026/05/01 has:attachment' — recent with attachments
messages = list_messages(gmail_service, query='is:unread label:inbox')

جلب رسالة كاملة

def get_message(service, message_id, user_id='me'):
    message = service.users().messages().get(
        userId=user_id,
        id=message_id,
        format='full'  # 'full', 'metadata', or 'minimal'
    ).execute()
    return message

# Fetch the first unread message
messages = list_messages(gmail_service, query='is:unread', max_results=1)
if messages:
    msg = get_message(gmail_service, messages[0]['id'])
    print('Thread ID:', msg['threadId'])
    print('Labels:', msg['labelIds'])
    print('Snippet:', msg['snippet'][:100])

استخراج رؤوس البريد الإلكتروني

def get_header(message, name):
    headers = message.get('payload', {}).get('headers', [])
    for h in headers:
        if h['name'].lower() == name.lower():
            return h['value']
    return ''

def extract_email_meta(message):
    return {
        'id': message['id'],
        'from': get_header(message, 'From'),
        'to': get_header(message, 'To'),
        'subject': get_header(message, 'Subject'),
        'date': get_header(message, 'Date'),
        'snippet': message.get('snippet', '')
    }

meta = extract_email_meta(msg)
print(f'From: {meta["from"]}')
print(f'Subject: {meta["subject"]}')
print(f'Date: {meta["date"]}')

فك ترميز نص البريد الإلكتروني (base64)

import base64

def decode_body(data):
    if not data:
        return ''
    decoded_bytes = base64.urlsafe_b64decode(data + '==')
    return decoded_bytes.decode('utf-8', errors='replace')

def get_email_body(message):
    payload = message.get('payload', {})
    mime_type = payload.get('mimeType', '')

    # Simple (non-multipart) email
    if 'body' in payload and payload['body'].get('data'):
        return decode_body(payload['body']['data'])

    # Multipart email: find the text/plain or text/html part
    parts = payload.get('parts', [])
    for part in parts:
        if part.get('mimeType') == 'text/plain':
            return decode_body(part['body'].get('data', ''))

    # Fallback: try text/html
    for part in parts:
        if part.get('mimeType') == 'text/html':
            return decode_body(part['body'].get('data', ''))

    return message.get('snippet', '')

# --- demo ---
encoded = base64.urlsafe_b64encode(b'Hello from the agent!').decode().rstrip('=')
print('Decoded body:', decode_body(encoded))

message = {
    'payload': {
        'mimeType': 'multipart/alternative',
        'parts': [
            {'mimeType': 'text/plain', 'body': {'data': encoded}}
        ]
    }
}
print('Email body:', get_email_body(message))

معالجة رسائل البريد متعددة الأجزاء بشكل递归ي

import base64

def extract_parts(payload, target_mime='text/plain'):
    parts_text = []
    mime_type = payload.get('mimeType', '')

    if mime_type == target_mime:
        data = payload.get('body', {}).get('data', '')
        if data:
            decoded = base64.urlsafe_b64decode(data + '==').decode('utf-8', errors='replace')
            parts_text.append(decoded)

    # Recurse into sub-parts
    for part in payload.get('parts', []):
        parts_text.extend(extract_parts(part, target_mime))

    return parts_text

def get_plain_text(message):
    payload = message.get('payload', {})
    texts = extract_parts(payload, 'text/plain')
    return '\n\n'.join(texts) if texts else message.get('snippet', '')

body_text = get_plain_text(msg)
print(f'Body ({len(body_text)} chars):', body_text[:200])

وضع علامة مقروء على الرسائل

def mark_as_read(service, message_id, user_id='me'):
    service.users().messages().modify(
        userId=user_id,
        id=message_id,
        body={'removeLabelIds': ['UNREAD']}
    ).execute()
    print(f'Marked {message_id} as read')

def add_label(service, message_id, label_id, user_id='me'):
    service.users().messages().modify(
        userId=user_id,
        id=message_id,
        body={'addLabelIds': [label_id]}
    ).execute()

# Get label ID by name
def get_label_id(service, label_name, user_id='me'):
    labels = service.users().labels().list(userId=user_id).execute()
    for label in labels.get('labels', []):
        if label['name'].lower() == label_name.lower():
            return label['id']
    return None

# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
    def __init__(self, result):
        self._result = result
    def execute(self):
        return self._result

class _FakeUsers:
    def messages(self):
        return self
    def modify(self, **kwargs):
        print(f'[gmail api] messages.modify({kwargs})')
        return _Exec({'id': kwargs.get('id')})
    def labels(self):
        return self
    def list(self, **kwargs):
        return _Exec({'labels': [{'id': 'Label_1', 'name': 'Processed'}]})

class _FakeService:
    def users(self):
        return _FakeUsers()

service = _FakeService()
mark_as_read(service, 'msg_42')
add_label(service, 'msg_42', 'Label_1')
print('Label id for "Processed":', get_label_id(service, 'Processed'))

إنشاء بريد إلكتروني باستخدام MIMEText

import base64
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def create_message(sender, to, subject, body_text, body_html=None):
    if body_html:
        msg = MIMEMultipart('alternative')
        msg.attach(MIMEText(body_text, 'plain', 'utf-8'))
        msg.attach(MIMEText(body_html, 'html', 'utf-8'))
    else:
        msg = MIMEText(body_text, 'plain', 'utf-8')

    msg['From'] = sender
    msg['To'] = to
    msg['Subject'] = subject

    # Encode as base64url
    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
    return {'raw': raw}

message = create_message(
    sender='agent@yourcompany.com',
    to='recipient@example.com',
    subject='Weekly Summary',
    body_text='Hello,\n\nHere is your summary.\n\nBest,\nAgent'
)

# --- demo ---
print('Message keys:', list(message.keys()))
print('Base64 length:', len(message['raw']))

إرسال بريد إلكتروني باستخدام Gmail API

from googleapiclient.errors import HttpError

def send_message(service, message, user_id='me'):
    try:
        sent = service.users().messages().send(
            userId=user_id,
            body=message
        ).execute()
        print(f'Message sent! ID: {sent["id"]}')
        return sent
    except HttpError as e:
        import json
        body = json.loads(e.content.decode())
        print(f'Send failed ({e.resp.status}): {body.get("error", {}).get("message")}')
        return None

# Send the message
message = create_message(
    sender='me',
    to='team@company.com',
    subject='Agent Report',
    body_text='Processing complete. 42 tasks handled.'
)
send_message(gmail_service, message)

إنشاء مسودات البريد الإلكتروني وإرسالها

def create_draft(service, message, user_id='me'):
    draft = service.users().drafts().create(
        userId=user_id,
        body={'message': message}
    ).execute()
    print(f'Draft created: {draft["id"]}')
    return draft

def send_draft(service, draft_id, user_id='me'):
    sent = service.users().drafts().send(
        userId=user_id,
        body={'id': draft_id}
    ).execute()
    print(f'Draft sent as message: {sent["id"]}')
    return sent

# Create a draft for review
message = create_message(
    sender='me',
    to='client@example.com',
    subject='Proposal Follow-up',
    body_text='Dear Client,\n\nFollowing up on our proposal...'
)
draft = create_draft(gmail_service, message)
# Human reviews in Gmail, then agent sends:
# send_draft(gmail_service, draft['id'])

معالجة عدة رسائل بريد إلكتروني على دفعات

import time

def process_unread_emails(service, max_emails=20):
    messages = list_messages(
        service,
        query='is:unread label:inbox',
        max_results=max_emails
    )

    processed = []
    for i, msg_ref in enumerate(messages):
        # Rate-limit: process max 5 per second
        if i > 0 and i % 5 == 0:
            time.sleep(1)

        msg = get_message(service, msg_ref['id'])
        meta = extract_email_meta(msg)
        body = get_plain_text(msg)

        result = {
            'id': msg['id'],
            'from': meta['from'],
            'subject': meta['subject'],
            'body_preview': body[:200]
        }
        processed.append(result)
        mark_as_read(service, msg['id'])

    return processed

الرد على بريد إلكتروني (ضمن سلسلة الرسائل)

import base64
from email.mime.text import MIMEText

def create_reply(original_message, reply_text, sender='me'):
    original_msg_id = get_header(original_message, 'Message-ID')
    to = get_header(original_message, 'From')
    subject = get_header(original_message, 'Subject')
    if not subject.startswith('Re:'):
        subject = 'Re: ' + subject

    msg = MIMEText(reply_text, 'plain', 'utf-8')
    msg['From'] = sender
    msg['To'] = to
    msg['Subject'] = subject
    msg['In-Reply-To'] = original_msg_id
    msg['References'] = original_msg_id

    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode('utf-8')
    return {
        'raw': raw,
        'threadId': original_message['threadId']  # keeps it in thread
    }

تحقق سريع: نص البريد الإلكتروني بترميز base64

مراجعة قراءة البريد الإلكتروني وإرساله