main.py

import webbrowser
import datetime
from zoneinfo import ZoneInfo
import os.path
from pathlib import Path

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


class GCalMeet:
    """A class to hold data an functionality for the GCal event fetching and evaluating."""

    def __init__(self) -> None:
        # If modifying these scopes, delete the file token.json.
        self.SCOPES = ["https://www.googleapis.com/auth/calendar.readonly"]
        self.PWD = Path(__file__).resolve().parent

    def login(self) -> None:
        """Logs into Google API"""
        self.creds = None
        # The file token.json stores the user's access and refresh tokens, and is
        # created automatically when the authorization flow completes for the first
        # time.
        if os.path.exists(self.PWD / "token.json"):
            self.creds = Credentials.from_authorized_user_file(
                self.PWD / "token.json", self.SCOPES
            )
        # If there are no (valid) credentials available, let the user log in.
        if not self.creds or not self.creds.valid:
            if self.creds and self.creds.expired and self.creds.refresh_token:
                self.creds.refresh(Request())
            else:
                flow = InstalledAppFlow.from_client_secrets_file(
                    self.PWD / "credentials.json", self.SCOPES
                )
                self.creds = flow.run_local_server(port=0)
            # Save the credentials for the next run
            with open(self.PWD / "token.json", "w") as token:
                token.write(self.creds.to_json())

    def fetch(self) -> None:
        try:
            service = build("calendar", "v3", credentials=self.creds)

            # Call the Calendar API
            now = datetime.datetime.now(tz=datetime.timezone.utc).isoformat()
            print("Getting the upcoming 10 events")
            events_result = (
                service.events()
                .list(
                    calendarId="primary",
                    timeMin=now,
                    maxResults=10,
                    singleEvents=True,
                    orderBy="startTime",
                )
                .execute()
            )
            self.events = events_result.get("items", [])
        except HttpError as error:
            raise ValueError(f"Fetching from GCal API failed. + {error}") from error

    def _convert_start_to_datetime_object(self, start: dict) -> datetime.datetime:
        naive_dt = datetime.datetime.fromisoformat(start["dateTime"])
        aware_dt = naive_dt.astimezone(ZoneInfo(start["timeZone"]))
        return aware_dt

    def _event_declined(self, event: dict) -> bool:
        """Check in attendees for having declined a meeting."""

        if "attendees" not in event:
            return False

        for attendee in event["attendees"]:
            if attendee.get("self"):
                if attendee["responseStatus"] == "declined":
                    return True
        return False

    def _event_starting_now(
        self, aware_now: datetime.datetime, aware_dt: datetime.datetime
    ) -> bool:
        timediff = aware_now - aware_dt
        if datetime.timedelta(minutes=4) >= timediff >= datetime.timedelta(minutes=-1):
            return True
        return False

    def evaluate_events(self):
        if not self.events:
            print("No upcoming events found.")
            return

        # Prints the start and name of the next 10 events
        for event in self.events:
            if "dateTime" not in event["start"]:
                # full day events dont have `dateTime` but `date` only
                print(event["summary"], "skipped (full day event)")
                continue

            if self._event_declined(event=event):
                continue

            start_aware_dt = self._convert_start_to_datetime_object(event["start"])
            aware_now = datetime.datetime.now(tz=ZoneInfo("Europe/Berlin"))
            event_starting_now = self._event_starting_now(
                aware_dt=start_aware_dt, aware_now=aware_now
            )

            print(start_aware_dt, aware_now, event_starting_now, event["summary"])

            if event_starting_now:
                print("launch in browser")
                webbrowser.open(event["hangoutLink"], new=0, autoraise=True)
            else:
                pass
                # print("event not starting now...")


if __name__ == "__main__":
    gcalmeet = GCalMeet()
    gcalmeet.login()
    gcalmeet.fetch()
    gcalmeet.evaluate_events()