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) loop = asyncio.new_event_loop() tasks = [loop.create_task(check_slots_and_notify(config, dry_run=dry_run))] loop.run_until_complete(asyncio.wait(tasks)) loop.close()