diff --git a/.github/README.md b/.github/README.md index 5e8cda5..c22a98c 100644 --- a/.github/README.md +++ b/.github/README.md @@ -1,22 +1,26 @@ # 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`. @@ -24,14 +28,14 @@ Before continuing, make sure you have a Google Account. * 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: @@ -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 diff --git a/src/event.py b/src/event.py index 3d2b6f9..07b756d 100644 --- a/src/event.py +++ b/src/event.py @@ -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: @@ -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.") @@ -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() diff --git a/src/google_calendar_api.py b/src/google_calendar_api.py index 48a85f1..8ac49cd 100644 --- a/src/google_calendar_api.py +++ b/src/google_calendar_api.py @@ -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() @@ -48,7 +61,7 @@ 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, @@ -56,8 +69,7 @@ def get_events_this_month(self) -> List[Event]: ) .execute() ) - - logging.info( + self.logger.info( f"All Events: \n{json.dumps(events_result, indent=2)}" ) @@ -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() diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..92e9e07 --- /dev/null +++ b/src/main.py @@ -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) diff --git a/src/telegram_client.py b/src/telegram_client.py new file mode 100644 index 0000000..a3c579d --- /dev/null +++ b/src/telegram_client.py @@ -0,0 +1,73 @@ +import json +import logging +import os +from dataclasses import dataclass + +import requests + +BASE_URL = "https://api.telegram.org/bot{token}/{endpoint}" + + +@dataclass +class TelegramConfig: + bot_token: str + chat_id: str + + +class TelegramBot: + def __init__( + self, + bot_token: str = None, + chat_id: str = None, + logger=None, + ): + if not bot_token: + bot_token = os.environ.get("TELEGRAM_API") + if not bot_token: + raise ValueError( + "Bot token must be provided or set as an environment" + " variable." + ) + if not chat_id: + chat_id = os.environ.get("TELEGRAM_CHAT_ID") + if not chat_id: + raise ValueError( + "Chat ID must be provided or set as an environment" + " variable." + ) + self.config = TelegramConfig( + bot_token=bot_token, + chat_id=chat_id, + ) + logging.info(f"Config: chat_id={chat_id}") + + self.logger = logger or logging.getLogger(__name__) + + def send_message(self, message: str) -> None: + """ + Sends a message to the specified chat using the bot. + """ + endpoint = "sendMessage" + url = BASE_URL.format(token=self.config.bot_token, endpoint=endpoint) + payload = {"chat_id": self.config.chat_id, "text": message} + + response = requests.post(url, data=payload) + response_data = response.json() + self.logger.info( + f"Telegram Response: \n{json.dumps(response_data, indent=2)}" + ) + + if not response_data.get("ok"): + raise requests.RequestException( + f"Telegram API Error: {response_data.get('description')}" + ) + + +def send_telegram_message( + message: str, bot_token: str = None, chat_id: str = None, logger=None +) -> None: + """ + Convenience function to send message to Telegram + """ + bot = TelegramBot(bot_token, chat_id, logger=logger) + bot.send_message(message) diff --git a/src/templates/monthly_events_template.txt b/src/templates/monthly_events_template.txt new file mode 100644 index 0000000..1ce91e3 --- /dev/null +++ b/src/templates/monthly_events_template.txt @@ -0,0 +1,4 @@ +Event: {title} +Description: {description} +Starts: {start_time} - Ends: {end_time} +{optional_fields} diff --git a/tests/test_telegram_client.py b/tests/test_telegram_client.py new file mode 100644 index 0000000..857ebdf --- /dev/null +++ b/tests/test_telegram_client.py @@ -0,0 +1,48 @@ +import unittest +from unittest.mock import Mock, patch + +import requests + +from src.telegram_client import TelegramBot + + +class TestTelegramBot(unittest.TestCase): + def setUp(self): + self.bot_token = "TEST_BOT_TOKEN" + self.chat_id = "TEST_CHAT_ID" + self.message = "Test message" + + @patch("src.telegram_client.requests.post") + def test_send_message_success(self, mock_post): + mock_response = Mock() + mock_response.json.return_value = {"ok": True} + mock_post.return_value = mock_response + + bot = TelegramBot(self.bot_token, self.chat_id) + bot.send_message(self.message) + + @patch("src.telegram_client.requests.post") + def test_send_message_api_error(self, mock_post): + mock_response = Mock() + mock_response.json.return_value = { + "ok": False, + "description": "Test API Error", + } + mock_post.return_value = mock_response + + bot = TelegramBot(self.bot_token, self.chat_id) + + with self.assertRaises(requests.RequestException) as context: + bot.send_message(self.message) + self.assertEqual( + str(context.exception), "Telegram API Error: Test API Error" + ) + + @patch.dict("os.environ", {"TELEGRAM_API": "", "TELEGRAM_CHAT_ID": ""}) + def test_initialization_without_token_or_chatid(self): + with self.assertRaises(ValueError) as context: + self.bot = TelegramBot() + self.assertTrue( + "Bot token must be provided or set as an environment variable." + in str(context.exception) + )