55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
"""
|
|
Gmail API integration — read-only access.
|
|
Fetches unread messages and supports search queries.
|
|
"""
|
|
from googleapiclient.discovery import build
|
|
from core.google_auth import get_credentials
|
|
|
|
def _service():
|
|
return build('gmail', 'v1', credentials=get_credentials())
|
|
|
|
def get_unread_emails(count: int = 10) -> list[dict]:
|
|
"""Fetch the most recent unread emails."""
|
|
svc = _service()
|
|
result = svc.users().messages().list(
|
|
userId='me', q='is:unread', maxResults=count
|
|
).execute()
|
|
messages = result.get('messages', [])
|
|
emails = []
|
|
for msg in messages:
|
|
detail = svc.users().messages().get(
|
|
userId='me', id=msg['id'], format='metadata',
|
|
metadataHeaders=['Subject', 'From', 'Date']
|
|
).execute()
|
|
headers = {h['name']: h['value'] for h in detail['payload']['headers']}
|
|
emails.append({
|
|
'id': msg['id'],
|
|
'subject': headers.get('Subject', '(No subject)'),
|
|
'from': headers.get('From', ''),
|
|
'date': headers.get('Date', ''),
|
|
'snippet': detail.get('snippet', ''),
|
|
})
|
|
return emails
|
|
|
|
def search_emails(query: str, count: int = 20) -> list[dict]:
|
|
"""Search Gmail using standard Gmail search syntax."""
|
|
svc = _service()
|
|
result = svc.users().messages().list(
|
|
userId='me', q=query, maxResults=count
|
|
).execute()
|
|
messages = result.get('messages', [])
|
|
emails = []
|
|
for msg in messages:
|
|
detail = svc.users().messages().get(
|
|
userId='me', id=msg['id'], format='metadata',
|
|
metadataHeaders=['Subject', 'From', 'Date']
|
|
).execute()
|
|
headers = {h['name']: h['value'] for h in detail['payload']['headers']}
|
|
emails.append({
|
|
'id': msg['id'],
|
|
'subject': headers.get('Subject', '(No subject)'),
|
|
'from': headers.get('From', ''),
|
|
'snippet': detail.get('snippet', ''),
|
|
})
|
|
return emails
|