Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Telegram integration #2

Merged
merged 5 commits into from
Sep 21, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 37 additions & 6 deletions .github/README.md
Original file line number Diff line number Diff line change
@@ -1,37 +1,41 @@
# daily-event-publisher

## Connect To Google Calendar Using A Service Account
This project integrates Google Calendar with Telegram, notifying users of their upcoming events for the month.

## Prerequisites

### Connect To Google Calendar Using A Service Account

Before continuing, make sure you have a Google Account.

### Project Creation
#### Project Creation

* Go to the [Google Cloud Console](https://console.cloud.google.com/).
* Click on the project drop-down and then click on NEW `PROJECT`.
* Give your project a name and click `CREATE`.

### Enable Google Calendar API
#### Enable Google Calendar API

* In the left sidebar, navigate to `APIs & Services` > `Library`.
* Search for `Google Calendar API` and select it.
* Click `ENABLE`.

### Service Account Creation
#### Service Account Creation

* In the left sidebar, navigate to `IAM & Admin` > `Service accounts`.
* Click on `CREATE SERVICE ACCOUNT`.
* Provide a name and description for the service account. Click `CREATE`.
* Grant the service account the required permissions. For basic calendar access, you can choose `Role` > `Google Calendar API` > `Calendar Viewer`. Click `CONTINUE`.
* Click `DONE`.

### Download Credentials
#### Download Credentials

* In the service accounts list, find the service account you just created.
* Click on its name to view details.
* In the `KEYS` tab, click on `ADD KEY` > `JSON`.
* A `credentials.json` file will be downloaded. This file contains the credentials your application will use to authenticate its API requests.

### Share Your Google Calendar
#### Share Your Google Calendar

If you're intending to access a specific Google Calendar, make sure to share it with your service account:

Expand All @@ -41,6 +45,33 @@ If you're intending to access a specific Google Calendar, make sure to share it
* Click on `+ Add people` and enter the email address of the service account (found in the service account details in Google Cloud Console).
* Set the desired permissions (e.g., `See all event details`) and click `Send`.

### Telegram Bot Token and Channel ID

* [Telegram Bot Tutorial](https://core.telegram.org/bots/tutorial).

## Configuration

* Google Calendar:
* credentials.json

* Telegram:
* Environment Tokens.

## Customization

* Event Formatting:
* Modify the monthly_events_template.txt found in the templates directory to adjust the appearance of event messages.
* Extended Fields in Google Calendar:
* When creating events in Google Calendar, you can specify additional details in the description as follows:

```txt
Tickets: [Ticket Info]
Location: [Location Info]
Website: [Website URL]
```

The application will parse these details and include them in the Telegram message.

## Local Development

### Prerequisites
Expand Down
83 changes: 62 additions & 21 deletions src/event.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

TICKETS_PREFIX = "Tickets:"
LOCATION_PREFIX = "Location:"
WEBSITE_PREFIX = "Website:"


@dataclass
class Event:
Expand All @@ -14,6 +19,16 @@ class Event:
location: Optional[str] = None
website: Optional[str] = None

TEMPLATE_PATH = (
Path(__file__).parent / "templates/monthly_events_template.txt"
)

@property
def template_content(self) -> str:
if not hasattr(self, "_template_content"):
self._template_content = self._read_template()
return self._template_content

def __post_init__(self):
if not self.title:
raise ValueError("Title is required.")
Expand All @@ -23,26 +38,52 @@ def __post_init__(self):
raise ValueError("Start time is required.")
if not self.end_time:
raise ValueError("End time is required.")
self._parse_description(self.description)

def _read_template(self) -> str:
"""Read the event template file."""
try:
with self.TEMPLATE_PATH.open() as template_file:
return template_file.read()
except Exception as e:
raise IOError(f"Error reading template file: {e}")

def _parse_description(self, description: str) -> None:
"""Parse the description to extract extended fields."""
lines = description.split("\n")

for line in lines:
if line.startswith(TICKETS_PREFIX):
self.tickets = line.replace(TICKETS_PREFIX, "").strip()
elif line.startswith(LOCATION_PREFIX):
self.location = line.replace(LOCATION_PREFIX, "").strip()
elif line.startswith(WEBSITE_PREFIX):
self.website = line.replace(WEBSITE_PREFIX, "").strip()

# Remove extracted details from the main description.
self.description = "\n".join(
[
line
for line in lines
if not line.startswith(
(TICKETS_PREFIX, LOCATION_PREFIX, WEBSITE_PREFIX)
)
]
).strip()

def pretty(self) -> str:
"""Pretty print the event."""
base_str = (
f"Event: {self.title}\n"
f"Description: {self.description}\n"
f"Starts: {self.start_time} - Ends: {self.end_time}"
)

additional_str = []
if self.tickets:
additional_str.append(f"Tickets: {self.tickets}")
if self.location:
additional_str.append(f"Location: {self.location}")
if self.website:
additional_str.append(f"Website: {self.website}")

optional_str = " | ".join(additional_str)

if optional_str:
return f"{base_str}\n{optional_str}"
else:
return base_str
"""Pretty print the event using a template."""
optional_fields = [
f"{TICKETS_PREFIX} {self.tickets}" if self.tickets else None,
f"{LOCATION_PREFIX} {self.location}" if self.location else None,
f"{WEBSITE_PREFIX} {self.website}" if self.website else None,
]
optional_str = " | ".join(filter(None, optional_fields))

return self.template_content.format(
title=self.title,
description=self.description,
start_time=self.start_time,
end_time=self.end_time,
optional_fields=optional_str,
).rstrip()
62 changes: 36 additions & 26 deletions src/google_calendar_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,30 +10,43 @@

from event import Event

# Setup logging
logging.basicConfig(level=logging.INFO)


class GoogleCalendarClient:
"""A class to handle operations related to the Google Calendar."""

CALENDAR_ID: str = os.environ.get("GOOGLE_CALENDAR_ID", "0")
SERVICE_NAME = "calendar"
SERVICE_VERSION = "v3"
SCOPES = ["https://www.googleapis.com/auth/calendar.readonly"]

def __init__(self, credentials_file: str = None):
def __init__(
self,
credentials_file: str = None,
calendar_id: str = None,
logger=None,
):
"""Initializes the GoogleCalendarClient class with credentials."""

if not credentials_file:
credentials_file = os.environ.get(
"GOOGLE_CALENDAR_CREDENTIALS", "credentials.json"
)

if not calendar_id:
calendar_id = os.environ.get("GOOGLE_CALENDAR_ID", "0")

self.calendar_id = calendar_id

self.credentials: Credentials = Credentials.from_service_account_file(
credentials_file,
scopes=["https://www.googleapis.com/auth/calendar.readonly"],
credentials_file, scopes=self.SCOPES
)
self.service: Resource = build(
"calendar", "v3", credentials=self.credentials
self.SERVICE_NAME,
self.SERVICE_VERSION,
credentials=self.credentials,
)

self.logger = logger or logging.getLogger(__name__)

def get_events_this_month(self) -> List[Event]:
"""Returns the events for the current month in 'Event' format."""
today = datetime.datetime.utcnow()
Expand All @@ -48,16 +61,15 @@ def get_events_this_month(self) -> List[Event]:
events_result = (
self.service.events()
.list(
calendarId=self.CALENDAR_ID,
calendarId=self.calendar_id,
timeMin=first_day_of_month.isoformat() + "Z",
timeMax=last_day_of_month.isoformat() + "Z",
singleEvents=True,
orderBy="startTime",
)
.execute()
)

logging.info(
self.logger.info(
f"All Events: \n{json.dumps(events_result, indent=2)}"
)

Expand All @@ -80,22 +92,20 @@ def get_events_this_month(self) -> List[Event]:
]

return events_list

except Exception as e:
logging.error(f"Error fetching events: {e}")
self.logger.error(f"Error fetching events: {e}")
return []

def get_events(self) -> List[Event]:
"""Displays the events for the current month."""
events = self.get_events_this_month()
if not events:
logging.info("No events found for this month.")
else:
for event in events:
logging.info(event)
return events


if __name__ == "__main__":
gc = GoogleCalendarClient("credentials.json")
gc.get_events()
def get_this_month_events(
credentials_file: str = None, calendar_id: str = None, logger=None
) -> List[Event]:
"""
Convenience function to fetch events for the current month.
"""
client = GoogleCalendarClient(
credentials_file=credentials_file,
calendar_id=calendar_id,
logger=logger,
)
return client.get_events_this_month()
35 changes: 35 additions & 0 deletions src/main.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
import logging

from google_calendar_api import get_this_month_events
from telegram_client import send_telegram_message

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)


def main(credentials_path: str):
try:
events = get_this_month_events(credentials_file=credentials_path)

if not events:
logger.warning("No events found for this month.")
return

messages = [event.pretty() for event in events]
events_message = "\n\n".join(messages)

send_telegram_message(events_message, logger=logger)

except FileNotFoundError:
logger.error(f"Credentials file '{credentials_path}' not found.")
except Exception as e:
logger.error(f"Error: {e}")


if __name__ == "__main__":
# argparse?
credentials_path = "credentials.json"
main(credentials_path)
Loading