From a3c265b22e940e091633129ada3dc877d4739126 Mon Sep 17 00:00:00 2001 From: Guilherme Bufolo Date: Tue, 20 Aug 2024 23:33:41 +0200 Subject: [PATCH] Working version Allows to query the Thekendienst calendar and report empty slots to telegram. --- cfg/check_next_7_days.json | 5 + cfg/config_template.json | 21 +++ requirements.txt | 3 + src/reportMissingThekendienst.py | 273 +++++++++++++++++++++++++++++++ 4 files changed, 302 insertions(+) create mode 100644 cfg/check_next_7_days.json create mode 100644 cfg/config_template.json create mode 100644 requirements.txt create mode 100644 src/reportMissingThekendienst.py diff --git a/cfg/check_next_7_days.json b/cfg/check_next_7_days.json new file mode 100644 index 0000000..66fde0c --- /dev/null +++ b/cfg/check_next_7_days.json @@ -0,0 +1,5 @@ +{ + "start_date": null, + "end_date": null, + "days_to_check": 7 +} diff --git a/cfg/config_template.json b/cfg/config_template.json new file mode 100644 index 0000000..a575217 --- /dev/null +++ b/cfg/config_template.json @@ -0,0 +1,21 @@ +{ + "teamup_api_key": "", + "calendar_id": "ksp4hsa93c1nt5kmym", + "telegram_bot_token": "", + "telegram_channels": [{"name": "ThekendienstBotTest", "id": -4558981107}], + "subcalendars_to_check": ["flexibler Thekendienst", "regelmäßiger Thekendienst"], + "timezone": "Europe/Berlin", + "header": "Es gibt noch offene Thekendienste für diese Woche\\!\nWer kann bitte im [Kalender](https://teamup.com/ksp4hsa93c1nt5kmym) eintragen und im 'HOBBYHIMMEL Thekenhelden' chat melden\\.\n\n", + "footer": "\\- Der freundliche Theckendiensterinnerungsbot", + "no_open_slots": "Juhu\\! Diese Woche gibt es keine offenen Slots\\!", + "appointment_motivator": "Da mache ich Theke\\.", + "time_slots": { + "Monday": {"start": "17:00", "end": "22:00"}, + "Tuesday": {"start": "17:00", "end": "22:00"}, + "Wednesday": {"start": "17:00", "end": "22:00"}, + "Thursday": {"start": "17:00", "end": "22:00"}, + "Friday": {"start": "15:00", "end": "22:00"}, + "Saturday": {"start": "12:00", "end": "22:00"}, + "Sunday": {"start": "12:00", "end": "20:00"} + } +} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e79fac5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +python-telegram-bot==21.4 +requests==2.32.3 +urllib3==2.2.2 diff --git a/src/reportMissingThekendienst.py b/src/reportMissingThekendienst.py new file mode 100644 index 0000000..1e888a8 --- /dev/null +++ b/src/reportMissingThekendienst.py @@ -0,0 +1,273 @@ +from typing import List +import urllib.parse +import requests +import json +import sys +from datetime import datetime, timedelta, time +from zoneinfo import ZoneInfo +from typing import NamedTuple +import locale +import urllib +import telegram +import asyncio + + +class Subcalendar(NamedTuple): + name: str + id: int + + +class TimeSlot(NamedTuple): + start: datetime + end: datetime + + def covered(this, event): + event_start = datetime.strptime(event["start_dt"], "%Y-%m-%dT%H:%M:%S%z") + event_end = datetime.strptime(event["end_dt"], "%Y-%m-%dT%H:%M:%S%z") + is_covered = this.start >= event_start and this.end <= event_end + return is_covered + + +def load_config(config_files): + config = {} + for config_file in config_files: + try: + with open(config_file, "r") as file: + config_content = json.load(file) + config.update(config_content) + except Exception as e: + print(f"Error loading configuration {config_file}: {e}") + sys.exit(1) + + return config + + +def fetch_subcalendar_ids(api_key, calendar_id): + headers = {"Teamup-Token": api_key} + url = f"https://api.teamup.com/{calendar_id}/subcalendars" + response = requests.get(url, headers=headers) + response.raise_for_status() + subcalendars = response.json().get("subcalendars", []) + return {sub["name"]: sub["id"] for sub in subcalendars} + + +def fetch_events( + api_key: str, + calendar_id: str, + start_date: datetime, + end_date: datetime, + sub_calendars: List[Subcalendar], +): + subcalendar_query = [ + ("subcalendarId[]", sub_calendar.id) for sub_calendar in sub_calendars + ] + headers = {"Teamup-Token": api_key} + params = [ + ("startDate", start_date.strftime("%Y-%m-%d")), + ("endDate", end_date.strftime("%Y-%m-%d")), + *subcalendar_query, + ] + url = f"https://api.teamup.com/{calendar_id}/events" + response = requests.get(url, headers=headers, params=params) + response.raise_for_status() + return response.json().get("events", []) + + +def create_available_slots( + start_datetime: datetime, end_datetime: datetime +) -> List[TimeSlot]: + available_slots = [] + current_time = start_datetime + time_inc = timedelta(minutes=15) + while current_time < end_datetime: + available_slots.append(TimeSlot(current_time, current_time + time_inc)) + current_time += time_inc + return available_slots + + +def merge_slots(slots: List[TimeSlot]) -> List[TimeSlot]: + new_slots = [] + if not slots: + return new_slots + + current_slot = slots[0]._replace() + for slot in slots[1:]: + if current_slot.end == slot.start: + current_slot = current_slot._replace(end=slot.end) + else: + new_slots.append(current_slot) + current_slot = slot._replace() + + new_slots.append(current_slot) + + return new_slots + + +def get_free_time_slots( + events, date: datetime, start_time: time, end_time: time +) -> List[TimeSlot]: + start_datetime = date.replace( + hour=start_time.hour, + minute=start_time.minute, + second=start_time.second, + microsecond=0, + ) + end_datetime = date.replace( + hour=end_time.hour, + minute=end_time.minute, + second=end_time.second, + microsecond=0, + ) + + available_slots = create_available_slots(start_datetime, end_datetime) + + for event in events: + available_slots = [ + available_slot + for available_slot in available_slots + if not available_slot.covered(event) + ] + if not available_slots: + break + + return merge_slots(available_slots) + + +async def send_telegram_message(bot_token, channels, message): + bot = telegram.Bot(bot_token) + lpo = telegram.LinkPreviewOptions(is_disabled=True) + for channel in channels: + await bot.sendMessage( + text=message, + parse_mode=telegram.constants.ParseMode.MARKDOWN_V2, + chat_id=channel["id"], + link_preview_options=lpo, + ) + print("Message sent successfully!") + + +def fetch_subcalendar_id_from_name(config) -> List[Subcalendar]: + subcalendar_ids = fetch_subcalendar_ids( + config["teamup_api_key"], config["calendar_id"] + ) + interesting_calendars = config["subcalendars_to_check"] + subcalendars_to_check = [ + Subcalendar(name, int(subcalendar_ids[name])) + for name in interesting_calendars + if name in subcalendar_ids + ] + + calendar_not_found = False + for name in interesting_calendars: + if not name in subcalendar_ids: + print(f"Calendar {name} not found in response.") + calendar_not_found = True + + if calendar_not_found: + print(f"Known calendars: {subcalendar_ids}") + sys.exit(1) + + return subcalendars_to_check + + +def convert_to_date(zone, text, days_to_check=None) -> datetime: + date = datetime.strptime(text, "%Y-%m-%d") if text else None + + if date: + date = date.replace(tzinfo=zone) + else: + date = datetime.now(zone) + if days_to_check: + date += timedelta(days=days_to_check) + + return date + + +def parse_time(text: str) -> time: + return time.fromisoformat(text) + + +def create_teamup_event_link(config, start: datetime, end: datetime) -> str: + format_string = "%Y-%m-%d %H:%M:%S" + url_start = urllib.parse.quote(start.strftime(format_string)) + url_end = urllib.parse.quote(end.strftime(format_string)) + return f"https://teamup.com/{config['calendar_id']}/events/new?start_dt={url_start}&end_dt={url_end}" + + +async def check_slots_and_notify(config: map, dry_run: bool = False) -> None: + + sub_calendars = fetch_subcalendar_id_from_name(config) + + tzone = ZoneInfo(config["timezone"]) + start_date = convert_to_date(tzone, config.get("start_date")).replace( + hour=0, minute=0, second=0, microsecond=0 + ) + timedelta(days=1) + end_date = convert_to_date( + tzone, config.get("end_date"), config.get("days_to_check", 7) + ).replace(hour=23, minute=59, second=59, microsecond=999) + + print( + f"Checking Thekendienst between ({start_date:%A}) {start_date:%Y-%m-%d} and ({end_date:%A}) {end_date:%Y-%m-%d}." + ) + + events = fetch_events( + config["teamup_api_key"], + config["calendar_id"], + start_date, + end_date, + sub_calendars, + ) + + print( + f"Found {len(events)} events in the time range for calendars '{"', '".join(sub_calendar.name for sub_calendar in sub_calendars)}'." + ) + + message = "" + for i in range((end_date - start_date).days + 1): + check_date = start_date + timedelta(days=i) + day_of_week = check_date.strftime("%A") # Get the day name, e.g., 'Monday' + + if day_of_week in config["time_slots"]: + start_time = parse_time(config["time_slots"][day_of_week]["start"]) + end_time = parse_time(config["time_slots"][day_of_week]["end"]) + + free_time_slots = get_free_time_slots( + events, check_date, start_time, end_time + ) + + # kind of a hack but I don't want to install any packages + old_locale = locale.getlocale() + locale.setlocale(locale.LC_ALL, "de_DE.utf8") + + if free_time_slots: + free_slots = "\n ".join( + [ + f"`{slot.start:%H:%M} \\- {slot.end:%H:%M}` \\- [{config["appointment_motivator"]}]({create_teamup_event_link(config, slot.start, slot.end)}) 💪" + for slot in free_time_slots + ] + ) + message += f"🚨 `{free_time_slots[0].start:%a}, {free_time_slots[0].start:%d\\.%m\\.} `{free_slots}\n" + locale.setlocale(locale.LC_ALL, old_locale) + + if message: + message = config["header"] + message + "\n" + config["footer"] + else: + message = config["no_open_slots"] + + if dry_run: + print("Messsage that would be sent on Telegram:") + print(message) + else: + await send_telegram_message( + config["telegram_bot_token"], config["telegram_channels"], message + ) + + +if __name__ == "__main__": + default_config_file = "reportMissingThekendienstConfig.json" + dry_run = "--dry-run" in sys.argv + config_files = [arg for arg in sys.argv[1:] if arg != "--dry-run"] + if not config_files: + config_files = [default_config_file] + config = load_config(config_files) + asyncio.run(check_slots_and_notify(config, dry_run))