Code: Select all
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import pickle
import os
from datetime import datetime, timedelta
import logging
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# Define scopes for Gmail API
SCOPES = ['https://www.googleapis.com/auth/gmail.modify']
def main():
creds = None
# Load existing credentials
if os.path.exists('token.pickle'):
with open('token.pickle', 'rb') as token:
creds = pickle.load(token)
# Authenticate if no valid credentials
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)
with open('token.pickle', 'wb') as token:
pickle.dump(creds, token)
# Verify granted scopes
verify_scopes(creds)
service = build('gmail', 'v1', credentials=creds)
delete_all_emails_with_buffer(service)
def verify_scopes(creds):
"""Verify the scopes granted in the OAuth token."""
if creds and creds.scopes:
logging.info(f"Granted Scopes: {creds.scopes}")
required_scopes = set(SCOPES)
granted_scopes = set(creds.scopes)
if required_scopes.issubset(granted_scopes):
logging.info("All required scopes are granted.")
else:
missing_scopes = required_scopes - granted_scopes
logging.error(f"Missing required scopes: {missing_scopes}")
else:
logging.error("No scopes found in the token. Authentication might have failed.")
def delete_all_emails_with_buffer(service, dry_run=False):
try:
buffer_date = (datetime.now() - timedelta(days=14)).strftime('%Y/%m/%d')
query = f'label:inbox before:{buffer_date}'
messages = []
response = service.users().messages().list(userId='me', q=query).execute()
if 'messages' in response:
messages.extend(response['messages'])
while 'nextPageToken' in response:
response = service.users().messages().list(userId='me', q=query,
pageToken=response['nextPageToken']).execute()
messages.extend(response['messages'])
logging.info(f"Found {len(messages)} emails to delete.")
if dry_run:
logging.info("Dry run mode: No emails will be deleted.")
for msg in messages[:10]: # Show a sample of emails
try:
email_details = service.users().messages().get(userId='me', id=msg['id']).execute()
logging.info(f"Email ID: {msg['id']}, Subject: {email_details.get('snippet')}")
except Exception as e:
logging.error(f"Failed to fetch email details for ID {msg['id']}: {e}")
else:
logging.info("Deleting emails...")
for msg in messages:
try:
service.users().messages().delete(userId='me', id=msg['id']).execute()
logging.info(f"Deleted email ID: {msg['id']}")
except Exception as e:
logging.error(f"Failed to delete email ID {msg['id']}: {e}")
except Exception as e:
logging.error(f"Error during email deletion process: {e}")
if __name__ == '__main__':
main()
Code: Select all
2025-01-11 17:45:25,628 - WARNING - Encountered 403 Forbidden with reason "insufficientPermissions"
2025-01-11 17:45:25,628 - ERROR - Failed to delete email ID 194080c9712fd742: