google-calendar

#!/usr/bin/env python3
#
# This script depends on python-dateutil and python-pytz for properly handling
# times and time zones of calendar events.
import argparse
import datetime
import itertools
import logging
import os
import sys
import time
from typing import List, Optional, Set, Tuple

import dateutil.parser
import pytz
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

sys.path.append(os.path.join(os.path.dirname(__file__), "../../"))
import zulip

SCOPES = ["https://www.googleapis.com/auth/calendar.readonly"]
CLIENT_SECRET_FILE = "client_secret.json"  # noqa: S105
APPLICATION_NAME = "Zulip"
HOME_DIR = os.path.expanduser("~")

# Our cached view of the calendar, updated periodically.
events: List[Tuple[int, datetime.datetime, str]] = []

# Unique keys for events we've already sent, so we don't remind twice.
sent: Set[Tuple[int, datetime.datetime]] = set()

sys.path.append(os.path.dirname(__file__))

parser = zulip.add_default_arguments(
    argparse.ArgumentParser(
        r"""

google-calendar --calendar calendarID@example.calendar.google.com

    This integration can be used to send yourself reminders, on Zulip, of Google Calendar Events.

    Specify your Zulip API credentials and server in a ~/.zuliprc file or using the options.

    Before running this integration make sure you run the get-google-credentials file to give Zulip
    access to certain aspects of your Google Account.

    This integration should be run on your local machine. Your API key and other information are
    revealed to local users through the command line.

    Depends on: google-api-python-client
"""
    )
)


parser.add_argument(
    "--interval",
    dest="interval",
    default=30,
    type=int,
    action="store",
    help="Minutes before event for reminder [default: 30]",
    metavar="MINUTES",
)

parser.add_argument(
    "--calendar",
    dest="calendarID",
    default="primary",
    type=str,
    action="store",
    help="Calendar ID for the calendar you want to receive reminders from.",
)

options = parser.parse_args()

if not options.zulip_email:
    parser.error("You must specify --user")

zulip_client = zulip.init_from_options(options)


def get_credentials() -> Credentials:
    """Gets valid user credentials from storage.

    If nothing has been stored, or if the stored credentials are invalid,
    the user will be prompted to authenticate.

    Returns:
        Credentials, the obtained credential.
    """
    credential_path = os.path.join(HOME_DIR, "google-credentials.json")
    creds = None

    # Load credentials from file if they exist
    if os.path.exists(credential_path):
        creds = Credentials.from_authorized_user_file(credential_path, SCOPES)

    # If there are no (valid) credentials available, prompt the user to log in.
    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(CLIENT_SECRET_FILE, SCOPES)
            creds = flow.run_local_server(port=0)

        # Save the credentials for the next run
        with open(credential_path, "w") as token:
            token.write(creds.to_json())

    return creds


def populate_events() -> Optional[None]:
    credentials = get_credentials()
    service = build("calendar", "v3", credentials=credentials)

    now = datetime.datetime.now(pytz.utc).isoformat()
    feed = (
        service.events()
        .list(
            calendarId=options.calendarID,
            timeMin=now,
            maxResults=5,
            singleEvents=True,
            orderBy="startTime",
        )
        .execute()
    )

    events.clear()
    for event in feed["items"]:
        try:
            start = dateutil.parser.parse(event["start"]["dateTime"])
            # According to the API documentation, a time zone offset is required
            # for start.dateTime unless a time zone is explicitly specified in
            # start.timeZone.
            if start.tzinfo is None:
                event_timezone = pytz.timezone(event["start"]["timeZone"])
                # pytz timezones include an extra localize method that's not part
                # of the tzinfo base class.
                start = event_timezone.localize(start)
        except KeyError:
            # All-day events can have only a date.
            start_naive = dateutil.parser.parse(event["start"]["date"])

            # All-day events don't have a time zone offset; instead, we use the
            # time zone of the calendar.
            calendar_timezone = pytz.timezone(feed["timeZone"])
            # pytz timezones include an extra localize method that's not part
            # of the tzinfo base class.
            start = calendar_timezone.localize(start_naive)

        try:
            events.append((event["id"], start, event["summary"]))
        except KeyError:
            events.append((event["id"], start, "(No Title)"))


def send_reminders() -> Optional[None]:
    messages = []
    keys = set()
    now = datetime.datetime.now(tz=pytz.utc)

    for id, start, summary in events:
        dt = start - now
        if dt.days == 0 and dt.seconds < 60 * options.interval:
            # The unique key includes the start time, because of
            # repeating events.
            key = (id, start)
            if key not in sent:
                if start.hour == 0 and start.minute == 0:
                    line = f"{summary} is today."
                else:
                    line = "{} starts at {}".format(summary, start.strftime("%H:%M"))
                print("Sending reminder:", line)
                messages.append(line)
                keys.add(key)

    if not messages:
        return

    if len(messages) == 1:
        message = "Reminder: " + messages[0]
    else:
        message = "Reminder:\n\n" + "\n".join("* " + m for m in messages)

    zulip_client.send_message(
        dict(type="private", to=options.zulip_email, sender=options.zulip_email, content=message)
    )

    sent.update(keys)


# Loop forever
for i in itertools.count():
    try:
        # We check reminders every minute, but only
        # download the calendar every 10 minutes.
        if not i % 10:
            populate_events()
        send_reminders()
    except Exception:
        logging.exception("Couldn't download Google calendar and/or couldn't post to Zulip.")
    time.sleep(60)