Source code for todays_paper_web.widgets.calendar.google

from __future__ import print_function

import os
import datetime
from dateutil.parser import parse
import logging

from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

from PIL import Image, ImageDraw
from pydantic import Field

from todays_paper_web.fonts import Fonts
from todays_paper_web.widgets import DataSource, registerDataSource, registerView, View
from todays_paper_web.widgets.datasource import DataSourceParams

_LOGGER = logging.getLogger(__name__)
SCOPES = ["https://www.googleapis.com/auth/calendar.readonly"]




class GoogleCalendarParams(DataSourceParams):
    calendarId: str = Field(
        ..., description="The calendar to read, e.g. `primary` or a shared calendar's id."
    )
    maxResults: int = Field(..., description="Max number of upcoming events to return.")



@registerDataSource("calendar.google")
class GoogleCalendarDataSource(DataSource):
    Params = GoogleCalendarParams

    def getService(self):
        if not os.path.exists("token.json"):
            return None
        creds = Credentials.from_authorized_user_file("token.json", SCOPES)
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
            with open("token.json", "w") as token:
                token.write(creds.to_json())
        return build("calendar", "v3", credentials=creds)

    def fetch(self):
        service = self.getService()
        if not service:
            return {}
        try:
            now = datetime.datetime.utcnow().isoformat() + "Z"  # 'Z' indicates UTC time
            events_result = (
                service.events()
                .list(
                    calendarId=self.params.calendarId,
                    timeMin=now,
                    maxResults=self.params.maxResults,
                    singleEvents=True,
                    orderBy="startTime",
                )
                .execute()
            )
            events = events_result.get("items", [])
            for event in events:
                self.convertStrToDate(event["start"], "dateTime")

            if not events:
                return {}
            return events
        except HttpError as error:
            _LOGGER.error("An error occurred: %s" % error)

        return {}

    @staticmethod
    def convertStrToDate(obj: object, key: str) -> None:
        if key not in obj:
            return
        obj[key] = parse(obj[key])