google-calendar

#!/usr/bin/env python3
import argparse
import datetime
import itertools
import logging
import os
import runpy
import sys
import time
from configparser import ConfigParser
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Set, Tuple, TypedDict

import dateutil.parser
import pytz

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

SCOPES = ["https://www.googleapis.com/auth/calendar.events.readonly"]
HOME_DIR = os.path.expanduser("~")

# File containing user's access and refresh tokens for Google application requests.
# If it does not exist, e.g., first run, it is generated on user authorization.
TOKENS_FILE = "google-tokens.json"
TOKENS_PATH = os.path.join(HOME_DIR, TOKENS_FILE)

# The client secret file identifies the application requesting the client's data,
# and is required for the OAuth flow to fetch the tokens.
# It needs to be downloaded from Google, by the user.
CLIENT_SECRET_FILE = "client_secret.json"  # noqa: S105
CLIENT_SECRET_PATH = os.path.join(HOME_DIR, CLIENT_SECRET_FILE)


@dataclass
class GoogleCalendarOptions:
    calendar_id: str = "primary"
    interval: int = 30
    channel: Optional[str] = None
    topic: str = "calendar-reminders"
    noauth_local_webserver: bool = False
    tokens_path: str = TOKENS_PATH
    client_secret_path: str = CLIENT_SECRET_PATH
    format_message: Optional[str] = None


class Event(TypedDict):
    id: str
    start: datetime.datetime
    end: datetime.datetime
    html_link: str
    # The following fields are optional, and may not be present in all events.
    summary: str
    description: str
    location: str
    hangout_link: str


# Our cached view of the calendar, updated periodically.
events: List[Event] = []

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

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

usage = r"""google-calendar [--config-file PATH_TO_ZULIPRC_OF_BOT]
                       [--interval MINUTES] [--calendar CALENDAR_ID]
                       [--channel CHANNEL_NAME] [--topic TOPIC_NAME]
                       [--client-secret-file PATH_TO_CLIENT_SECRET_FILE]
                       [--tokens-file PATH_TO_GOOGLE_TOKENS_FILE]
                       [-n] [--noauth_local_webserver]
                       [-f MESSAGE_TEMPLATE] [--format-message MESSAGE_TEMPLATE]

    This integration can be used to send Zulip messages as reminders for upcoming events from your Google Calendar.

    Create a generic bot on Zulip, download its zuliprc file, and use the --config-file option to specify the path to your bot's zuliprc.

    For more information, see https://zulip.com/integrations/doc/google-calendar.
"""

parser = zulip.add_default_arguments(argparse.ArgumentParser(usage=usage), allow_provisioning=True)
parser.add_argument(
    "--interval",
    type=int,
    help="Minutes before event for reminder [default: 30]",
)
parser.add_argument(
    "--calendar",
    dest="calendar_id",
    help="The ID of the calendar you want to receive reminders from. By default, the primary calendar is used.",
)
parser.add_argument(
    "--channel",
    help="The channel to which to send the reminders to. By default, messages are sent to the DMs of the bot owner.",
)
parser.add_argument(
    "--topic",
    help="The topic to which to send the reminders to. Ignored if --channel is unspecified. 'calendar-reminders' is used as the default topic name.",
)
parser.add_argument(
    "--client-secret-file",
    help="The path to the file containing the client secret for the Google Calendar API. By default, the client secret file is assumed to be at {CLIENT_SECRET_PATH}.",
    dest="client_secret_path",
)
parser.add_argument(
    "--tokens-file",
    help=f"The path to the file containing the tokens for the Google Calendar API. By default, the tokens file is generated at {TOKENS_PATH} after the first run.",
    dest="tokens_path",
)
parser.add_argument(
    "-n",
    "--noauth_local_webserver",
    action="store_true",
    help="The default authorization process runs a local web server, which requires a browser on the same machine. For non-interactive environments and machines without browser access, e.g., remote servers, this option allows manual authorization. The authorization URL is printed, which the user can copy into a browser, copy the resulting authorization code, and paste back into the command line.",
)
parser.add_argument(
    "-f",
    "--format-message",
    help="A Python f-string to use to format the markdown message template. This option overrides the default message template. The f-string can use the following variables: start, end, title, description, calendar_link, location, google_meet_link.\nNote that the title, description, location, and google_meet_link variables are optional for Google Calendar events, and hence may be empty. Empty fields are displayed as {No title}, {No description}, {No location}, and {No link} in the message template.",
)
commandline_options = parser.parse_args()
if commandline_options.verbose:
    logging.getLogger().setLevel(logging.INFO)


valid_keys = list(GoogleCalendarOptions.__dataclass_fields__.keys())


def load_config_options(config_path: Optional[str]) -> GoogleCalendarOptions:
    if config_path is None:
        config_path = zulip.get_default_config_filename()
    assert config_path is not None
    if not os.path.exists(config_path):
        logging.info("No config file found at %s", config_path)
        return GoogleCalendarOptions()

    logging.info("Loading Google Calendar configuration from %s", config_path)
    config = ConfigParser()
    try:
        config.read(config_path)
    except Exception:
        logging.exception("Error reading config file %s", config_path)

    section = "google-calendar"
    config_values = {}
    if section in config:
        for key, value in config[section].items():
            if key in valid_keys:
                expected_type = GoogleCalendarOptions.__annotations__[key]
                config_values[key] = True if expected_type == bool else expected_type(value)
                logging.info("Setting key: %s to %s", key, config_values[key])
            else:
                logging.warning(
                    "Unknown key %s in section %s of config file %s", key, section, config_path
                )
    return GoogleCalendarOptions(**config_values)


def update_calendar_options_from_commandline_args(
    calendar_options: GoogleCalendarOptions, commandline_options: argparse.Namespace
) -> None:
    for key, value in commandline_options.__dict__.items():
        # Boolean arguments (store-true) have a default value of False when not passed in.
        # So, we ignore them, to prevent overwriting the config file option that is set.
        if key in valid_keys and value is not None and value is not False:
            setattr(calendar_options, key, value)


# Calendar options can be passed in from the command line or via zuliprc.
# The command line options override the zuliprc options.
calendar_options = load_config_options(commandline_options.zulip_config_file)
update_calendar_options_from_commandline_args(calendar_options, commandline_options)

zulip_client = zulip.init_from_options(commandline_options)

# Import dependencies only after parsing command-line args,
# as the --provision flag can be used to install the dependencies.
try:
    from google.oauth2.credentials import Credentials
    from googleapiclient.discovery import build
except ImportError:
    logging.exception(
        "You have unsatisfied dependencies. Install all missing dependencies with %(command)s --provision",
        {"command": sys.argv[0]},
    )
    sys.exit(1)


def get_credentials() -> Credentials:
    """Fetches credentials using the get-google-credentials script.

    Needs to call get-google-credentials everytime, because the auth token expires every hour,
    needing to be refreshed using the refresh token.
    """
    try:
        fetch_creds = runpy.run_path("./get-google-credentials")["get_credentials"]
        return fetch_creds(
            calendar_options.tokens_path,
            calendar_options.client_secret_path,
            SCOPES,
            calendar_options.noauth_local_webserver,
        )
    except Exception:
        logging.exception("Error getting google credentials")
        sys.exit(1)


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=calendar_options.calendar_id,
            timeMin=now,
            maxResults=5,
            singleEvents=True,
            orderBy="startTime",
        )
        .execute()
    )

    events.clear()
    for event in feed["items"]:

        def get_start_or_end(event: Dict[str, Any], field_name: str) -> datetime.datetime:
            try:
                field = dateutil.parser.parse(event[field_name]["dateTime"])
                # a time zone offset is required unless timeZone is explicitly specified.
                if field.tzinfo is None:
                    # pytz timezones include an extra localize method that's not part
                    # of the tzinfo base class.
                    event_timezone = pytz.timezone(event[field_name]["timeZone"])
                    field = event_timezone.localize(field)
            except KeyError:
                # All-day events can have only a date.
                field_naive = dateutil.parser.parse(event[field_name]["date"])
                # All-day events do not have a time zone offset; use the calendar's time zone.
                calendar_timezone = pytz.timezone(feed["timeZone"])
                field = calendar_timezone.localize(field_naive)
            return field

        events.append(
            {
                "id": event["id"],
                "start": get_start_or_end(event, "start"),
                "end": get_start_or_end(event, "end"),
                "summary": event.get("summary", "(No Title)"),
                "description": event.get("description", ""),
                "html_link": event["htmlLink"],
                "location": event.get("location", ""),
                "hangout_link": event.get("hangoutLink", ""),
            }
        )


def construct_message_from_event(event: Event) -> str:
    if calendar_options.format_message:
        message = calendar_options.format_message.format(
            start=event["start"].strftime("%Y-%m-%d %H:%M"),
            end=event["end"].strftime("%Y-%m-%d %H:%M"),
            title=event["summary"],
            description=event["description"] or "{No description}",
            calendar_link=event["html_link"],
            location=event["location"] or "{No location}",
            google_meet_link=event["hangout_link"] or "{No link}",
        )
        decoded_message = bytes(message, "utf-8").decode("unicode_escape")
        return decoded_message

    time_period = (
        "today"
        if event["start"].hour == 0 and event["start"].minute == 0
        else f"""scheduled from {event["start"].strftime('%H:%M')} to {event["end"].strftime('%H:%M')}"""
    )
    location = f""", at {event["location"]},""" if event["location"] else ""
    description = f"""\n> {event["description"]}\n""" if event["description"] else ""
    google_meet_link = (
        f"""\n[Join call]({event["hangout_link"]}).""" if event["hangout_link"] else ""
    )

    message = f"""[{event["summary"]}]({event["html_link"]}){location} is {time_period}.{description}{google_meet_link}"""

    return message


def send_reminder_message(message: str, key: Tuple[str, datetime.datetime]) -> None:
    user_profile = zulip_client.get_profile()
    if calendar_options.channel is not None:
        result = zulip_client.send_message(
            {
                "type": "stream",
                "to": calendar_options.channel,
                "topic": calendar_options.topic,
                "content": message,
            }
        )
    else:
        result = zulip_client.send_message(
            {
                "type": "direct",
                "to": [user_profile.get("bot_owner_id") or user_profile["email"]],
                "content": message,
            }
        )
    if result["result"] != "success":
        logging.error("Error sending zulip message: %s: %s", result.get("code"), result.get("msg"))
    sent.add(key)


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

    for event in events:
        dt = event["start"] - now
        if dt.days == 0 and dt.seconds < 60 * calendar_options.interval:
            # The unique key includes the start time due to repeating events.
            key = (event["id"], event["start"])
            if key not in sent:
                message = construct_message_from_event(event)
                logging.info("Sending reminder: %s", message)
                send_reminder_message(message, key)


# 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)