Uncheck all items in the given google task lists

gtasks-uncheck-all-items.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
#   "google-api-python-client",
#   "google-auth-oauthlib",
#   "setuptools",
# ]
# ///

"""
Login to google account, based on either the cached credentials or using interactive oAuth2
authentication via the browser. Then given the task list name, uncheck all the items in that
list.
"""

import logging
import pickle
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import Mapping, Sequence, cast
import tempfile

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


# borrowing this from the syncall project
CLIENT_SECRET_CONTS = {
    "installed": {
        "client_id": "56921267535-bc1haqq3ptgp2gahmtosgksiilnpeotd.apps.googleusercontent.com",
        "project_id": "taskwarrior-gtasks-sync-409014",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        "token_uri": "https://oauth2.googleapis.com/token",
        "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
        "client_secret": "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",
        "redirect_uris": ["http://localhost"],
    }
}


def parse_args() -> Namespace:
    parser = ArgumentParser(description=__doc__)
    parser.add_argument(
        "-li",
        "--lists",
        dest="task_lists",
        nargs="*",
        required=True,
        help="Names of the lists of tasks to uncheck their items",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Increase output verbosity",
        dest="is_verbose",
    )
    return parser.parse_args()


class GoogleTasksUnchecker:
    def __init__(
        self,
        scopes: Sequence[str],
        oauth_port: int,
        credentials_cache: Path,
        logger: logging.Logger,
    ):
        self._scopes = scopes
        self._oauth_port = oauth_port
        self._credentials_cache = credentials_cache
        self._logger = logger

        # If you modify this, delete your previously saved credentials
        self._service = None

    def start(self):
        """Initiate the connection with Google Tasks.

        :return: The obtained credentials.
        """
        creds = None
        credentials_cache = self._credentials_cache
        if credentials_cache.is_file():
            with credentials_cache.open("rb") as f:
                creds = pickle.load(f)  # noqa: S301

        if not creds or not creds.valid:
            self._logger.debug("Invalid credentials. Fetching again...")
            if creds and creds.expired and creds.refresh_token:
                creds.refresh(Request())
            else:
                flow = InstalledAppFlow.from_client_config(
                    client_config=CLIENT_SECRET_CONTS,
                    scopes=self._scopes,
                )
                try:
                    creds = flow.run_local_server(port=self._oauth_port)
                except OSError as e:
                    raise RuntimeError(
                        f"Port {self._oauth_port} is already in use, please specify a"
                        " different port or stop the process that's already using it.",
                    ) from e

            # Save the credentials for the next run
            with credentials_cache.open("wb") as f:
                pickle.dump(creds, f)
        else:
            self._logger.info("Using already cached credentials...")

        self._service = discovery.build("tasks", "v1", credentials=creds)

        return creds

    def fetch_task_list_id(self, title: str, must_exist: bool = True) -> str | None:
        """Return the id of the task list based on the given Title.

        :returns: id or None if that was not found
        """
        res = self._service.tasklists().list().execute()  # type: ignore
        task_lists_list: list[GTasksList] = res["items"]  # type: ignore

        matching_task_lists = [
            task_list["id"]
            for task_list in task_lists_list
            if task_list["title"] == title
        ]

        if len(matching_task_lists) == 0:
            if must_exist:
                raise ValueError(f'Task list with title "{title}" not found')

            return None

        if len(matching_task_lists) == 1:
            return cast(str, matching_task_lists[0])

        raise RuntimeError(
            f'Multiple matching task lists for title -> "{title}"',
        )

    def uncheck_all_items(self, list_id: str):
        """Uncheck all the items in the given list."""
        assert self._service is not None
        request = self._service.tasks().list(
            tasklist=list_id, showCompleted=True, showHidden=True
        )

        # Loop until all pages have been processed.
        while request is not None:
            response = request.execute()
            for task in response["items"]:
                if task["status"] != "completed":
                    continue

                self._logger.debug('Unchecking item "%s"', task["title"])

                task_id = task["id"]
                self._service.tasks().patch(
                    tasklist=list_id,
                    task=task_id,
                    body={"status": "needsAction"},
                ).execute()

            request = self._service.tasks().list_next(request, response)


def main():
    args = parse_args()
    logger = logging.getLogger("unchecker")
    logging.basicConfig(
        # format="[%(levelname)s] %(message)s",
        level=logging.DEBUG if args.is_verbose else logging.INFO,
    )
    logging.getLogger("googleapiclient").setLevel(logging.ERROR)

    # Initialize the google side instance -----------------------------------------------------
    unchecker = GoogleTasksUnchecker(
        scopes=["https://www.googleapis.com/auth/tasks"],
        credentials_cache=Path.home() / ".gtasks_uncheck_all_items_credentials.pickle",
        oauth_port=8081,
        logger=logger,
    )

    # connect with your credentials -----------------------------------------------------------
    logger.info("Connecting to Google Tasks...")
    unchecker.start()

    # fetch all the task list IDs and fail early if one of them is not found ------------------
    task_list_to_id: Mapping[str, str] = {}
    for task_list in args.task_lists:
        task_list_id = unchecker.fetch_task_list_id(title=task_list, must_exist=True)
        if task_list_id is None:
            raise ValueError(
                f'Task list "{task_list}" not found in Google Tasks, cannot proceed'
            )

        task_list_to_id[task_list] = task_list_id

    logger.info("Connected to Google Tasks and fetched all the task list IDs...")

    # uncheck all the items in the given task lists -------------------------------------------
    for task_list, task_list_id in task_list_to_id.items():
        logger.info('Unchecking all items in the list "%s" ...', task_list)
        unchecker.uncheck_all_items(list_id=task_list_id)

    logger.info("All done!")


if __name__ == "__main__":
    main()