Permanently bulk-delete messages from gmail

delete_gmail.py

#! /usr/bin/env python3
"""Permanently delete messages from Gmail

WARNING: THIS SCRIPT CAN DO GREAT DAMAGE TO YOUR GMAIL ACCOUNT.
USE AT YOUR OWN RISK!

Setup guide:

1.  Create and activate a new virtualenv with Python 3

2.  Turn on Gmail API
    Follow steps 1 and 2 in the quick start guide to enable the Gmail API:

    https://developers.google.com/gmail/api/quickstart/python

3.  Save the credentials.json file

4.  Run this script (use --help for more details)

5.  Delete the token.pickle file in the current directory when you are finished 
    using this script. This will force a new Gmail authentication the next time
    the script is run.

Where to find the most recent version of this script:
https://gist.github.com/millerdev/0d65dafba0b866dfd81a77da996c6092
"""

import os.path
import pickle
import sys
from argparse import ArgumentParser, RawDescriptionHelpFormatter

from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build

SCOPES = ['https://mail.google.com/']


def main():
    parser = ArgumentParser(
        description=__doc__,
        formatter_class=RawDescriptionHelpFormatter,
    )
    parser.add_argument("query",
        help="Message query. Example: 'in:some-label older_than:2y'")
    parser.add_argument("-b", "--batch-size", type=int, default=500,
        help="Batch size (default: 500)")
    parser.add_argument("-c", "--credentials",
        default="./credentials.json",
        help="Path to credentials.json file (default: ./credentials.json)")
    args = parser.parse_args()

    creds = load_credentials(args.credentials)

    print("Permanently deleting messages matching: %s" % args.query)
    if input("\nType 'delete' to confirm: ").lower() != "delete":
        print("abort")
        sys.exit(1)

    service = build('gmail', 'v1', credentials=creds)
    msgsvc = service.users().messages()
    total = 0
    try:
        while True:
            deleted = delete_messages(args.query, msgsvc, args.batch_size)
            if not deleted:
                break
            total += deleted
            print(".", end="", flush=True)
    except KeyboardInterrupt:
        pass
    finally:
        print("")
        print("deleted %s messages" % total)


def delete_messages(query, msgsvc, batch_size):
    response = msgsvc.list(userId="me", q=query, maxResults=batch_size).execute()
    ids = [msg["id"] for msg in response.get("messages", [])]
    if ids:
        msgsvc.batchDelete(userId="me", body={'ids': ids}).execute()
    return len(ids)


def load_credentials(credentials_file):
    creds = None
    if os.path.exists('token.pickle'):
        with open('token.pickle', 'rb') as token_file:
            creds = pickle.load(token_file)
    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_file, SCOPES)
            creds = flow.run_local_server()
        with open('token.pickle', 'wb') as token_file:
            pickle.dump(creds, token_file)
    return creds


if __name__ == '__main__':
    main()