diff --git a/vdirsyncer/cli/discover.py b/vdirsyncer/cli/discover.py index ebd4e161..f8389f63 100644 --- a/vdirsyncer/cli/discover.py +++ b/vdirsyncer/cli/discover.py @@ -165,6 +165,7 @@ def __init__(self, config, *, connector): "CalDAVStorage", "GoogleCalendarStorage", "GoogleContactsStorage", + "GoogleTasksStorage", ]: assert connector is not None config["connector"] = connector diff --git a/vdirsyncer/cli/utils.py b/vdirsyncer/cli/utils.py index 97f48edc..f8424073 100644 --- a/vdirsyncer/cli/utils.py +++ b/vdirsyncer/cli/utils.py @@ -39,6 +39,7 @@ def __init__(self): "singlefile": "vdirsyncer.storage.singlefile.SingleFileStorage", "google_calendar": "vdirsyncer.storage.google.GoogleCalendarStorage", "google_contacts": "vdirsyncer.storage.google.GoogleContactsStorage", + "google_tasks": "vdirsyncer.storage.google.GoogleTasksStorage", } def __getitem__(self, name: str) -> Storage: diff --git a/vdirsyncer/storage/google.py b/vdirsyncer/storage/google.py index 3109213a..f37cabf1 100644 --- a/vdirsyncer/storage/google.py +++ b/vdirsyncer/storage/google.py @@ -1,5 +1,8 @@ from __future__ import annotations +import pytz +from icalendar import Calendar, Todo +from datetime import datetime import json import logging import os @@ -17,14 +20,15 @@ from ..utils import checkdir from ..utils import expand_path from ..utils import open_graphical_browser +from ..vobject import Item from . import base from . import dav from .google_helpers import _RedirectWSGIApp from .google_helpers import _WSGIRequestHandler +from googleapiclient.errors import HttpError logger = logging.getLogger(__name__) - TOKEN_URL = "https://accounts.google.com/o/oauth2/v2/auth" REFRESH_URL = "https://www.googleapis.com/oauth2/v4/token" @@ -244,3 +248,197 @@ def __init__(self, token_file, client_id, client_secret, **kwargs): # docs here because the current way we autogenerate those docs are too # simple for our advanced argspec juggling in `vdirsyncer.storage.dav`. __init__._traverse_superclass = base.Storage # type: ignore + + +class GoogleTasksStorage(base.Storage): + SCOPES = ["https://www.googleapis.com/auth/tasks"] + read_only = False + no_delete = False + storage_name = "google_tasks" + authenticated_service = None + + def __init__(self, **kwargs): + super().__init__( + kwargs["collection"], + self.read_only, + self.no_delete, + kwargs["instance_name"], + ) + self.tasklist_id = kwargs["collection"] + self.service = GoogleTasksStorage.get_authenticated_service( + kwargs["token_file"], kwargs["client_id"], kwargs["client_secret"] + ) + + async def list(self): + tasks_result = ( + self.service.tasks() + .list(tasklist=self.tasklist_id, showCompleted=True, showHidden=True) + .execute() + ) + tasks = tasks_result.get("items", []) + for t in tasks: + yield t["id"], t["etag"] + + async def get(self, href: str): + tasks_result = ( + self.service.tasks().get(tasklist=self.tasklist_id, task=href).execute() + ) + ics = GoogleTasksStorage.task_to_ics(tasks_result, self.tasklist_id) + return (Item(ics), tasks_result["etag"]) + + async def upload(self, item: Item): + try: + task = GoogleTasksStorage.ics_to_task(item.raw) + tasks_result = ( + self.service.tasks() + .insert(tasklist=self.tasklist_id, body=task) + .execute() + ) + return tasks_result["id"], tasks_result["etag"] + except HttpError as e: + logger.debug(f"GTasks API error :\n {e.content}") + raise e + + async def update(self, href: str, item: Item, etag: str): + tasks_result = ( + self.service.tasks().get(tasklist=self.tasklist_id, task=href).execute() + ) + if tasks_result["etag"] != etag: + raise exceptions.WrongEtagError(etag, tasks_result["etag"]) + task = GoogleTasksStorage.ics_to_task(item.raw, tasks_result) + tasks_result = ( + self.service.tasks().insert(tasklist=self.tasklist_id, body=task).execute() + ) + return tasks_result["id"], tasks_result["etag"] + + async def delete(self, href: str, etag: str): + try: + self.service.tasks().delete(tasklist=self.tasklist_id, task=href).execute() + except HttpError as e: + logger.debug(f"GTasks API error:\n{e.content}") + raise e + + async def get_meta(self, key: str): + if key == "displayname": + tasks_result = ( + self.service.tasklists().get(tasklist=self.tasklist_id).execute() + ) + return tasks_result["title"] + else: + return None + + @classmethod + async def discover(cls, **kwargs): + service = GoogleTasksStorage.get_authenticated_service( + kwargs["token_file"], kwargs["client_id"], kwargs["client_secret"] + ) + tasks_result = service.tasklists().list().execute() + for tl in tasks_result["items"]: + print(f"tl: {tl['id']} / {tl['title']} etag: {tl['etag']}") + yield dict(collection=tl["id"], **kwargs) + + @staticmethod + def get_authenticated_service(token_file: str, client_id: str, client_secret: str): + from google.oauth2.credentials import Credentials + from google_auth_oauthlib.flow import InstalledAppFlow + from google.auth.transport.requests import Request + from googleapiclient.discovery import build + + creds = None + token_dir = os.path.dirname(token_file) + if not os.path.isdir(token_dir): + os.makedirs(token_dir) + if os.path.isfile(token_file): + creds = Credentials.from_authorized_user_file( + token_file, GoogleTasksStorage.SCOPES + ) + if not creds or not creds.valid: + if creds and creds.expired and creds.refresh_token: + creds.refresh(Request()) + else: + flow = InstalledAppFlow.from_client_config( + { + "installed": { + "client_id": client_id, + "client_secret": client_secret, + "auth_uri": TOKEN_URL, + "token_uri": REFRESH_URL, + "redirect_uris": [ + "urn:ietf:wg:oauth:2.0:oob", + "http://localhost", + ], + } + }, + GoogleTasksStorage.SCOPES, + ) + creds = flow.run_local_server(port=0) + with open(token_file, "w") as token: + token.write(creds.to_json()) + + service = build("tasks", "v1", credentials=creds) + return service + + @staticmethod + def task_to_ics(task, calendar_name: str): + """ + Convert a Google Task to an iCalendar (ICS) event. + + Args: + task (dict): A task returned from the Google Tasks API. + calendar_name (str): Name to assign to the calendar. + + Returns: + str: An ICS file content string. + """ + now = datetime.now(pytz.utc) + cal = Calendar() + cal.add("prodid", "-//Google Tasks to ICS//mxm.dk//") + cal.add("version", "2.0") + cal.add("X-WR-CALNAME", calendar_name) + + todo = Todo() + todo.add("uid", task.get("id")) + todo.add("summary", task.get("title", "Untitled Task")) + notes = task.get("notes") + if notes: + todo.add("description", notes) + + updated = task.get("updated") + if updated: + dt = datetime.fromisoformat(task.get("updated").rstrip("Z")) + todo.add("dtstart", dt) + else: + todo.add("dtstart", now) + due = task.get("due") + if due: + dt = datetime.fromisoformat(due.rstrip("Z")) + todo.add("due", dt) + + todo.add( + "status", + "COMPLETED" if task.get("status") == "completed" else "NEEDS-ACTION", + ) + + cal.add_component(todo) + return cal.to_ical().decode("utf-8") + + @staticmethod + def ics_to_task(ics: str, task_to_update=None): + task = {} + cal = Calendar.from_ical(ics) + todo = cal.walk("VTODO") + if todo is None: + raise Exception("ICS contains no VTODO") + todo = todo[0] + # task['id'] = str(todo['UID']) + task["title"] = str(todo["SUMMARY"]) + if "DESCRIPTION" in todo.keys(): + task["notes"] = str(todo.get("DESCRIPTION")) + task["status"] = ( + "needsAction" + if "STATUS" not in todo.keys() or todo["STATUS"] != "COMPLETED" + else "completed" + ) + if "DUE" in todo.keys(): + task["due"] = todo["DUE"].dt.isoformat().replace("+00:00", "Z") + return task