from typing import List, Tuple import urllib.parse import requests import json import sys from datetime import datetime, timedelta, time from zoneinfo import ZoneInfo from typing import NamedTuple from collections import defaultdict import locale import urllib import telegram import asyncio import dateparser 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: if text: date = dateparser.parse(text).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 fetch_sub_calander_ids_for_new_event(config, sub_calendars: List[Subcalendar]): print(sub_calendars) return [entry.id for entry in sub_calendars if entry.name in config['default_subcalendar_for_new_event']] def create_teamup_event_link(config, sub_calendars, 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)) sub_calendars_id = fetch_sub_calander_ids_for_new_event(config, sub_calendars) sub_calendars_string = f"&subcalendar_ids=[{','.join([str(sub_calendar_id) for sub_calendar_id in sub_calendars_id])}]" return f"https://teamup.com/{config['calendar_id']}/events/new?start_dt={url_start}&end_dt={url_end}{sub_calendars_string}" def find_open_slots(config, events, start_date, end_date)-> List[TimeSlot]: open_slots = [] 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 ) open_slots += free_time_slots return open_slots def group_by_dow(open_slots: List[TimeSlot]) -> dict[str, list[str]]: res = defaultdict(list) for v in open_slots: res[v.start.strftime("%a")].append(v) return res def make_slots_message(config, sub_calendars, open_slots: List[TimeSlot])-> str: # 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") message = "" if open_slots: slots_by_dow = group_by_dow(open_slots) for dow, slots in slots_by_dow.items(): free_slots = "\n ".join( [ f"`{slot.start:%H:%M} \\- {slot.end:%H:%M}` \\- [{config["appointment_motivator"]}]({create_teamup_event_link(config, sub_calendars, slot.start, slot.end)}) 💪" for slot in slots ] ) message += f"🚨 `{dow}, {slots[0].start:%d\\.%m\\.} `{free_slots}\n" locale.setlocale(locale.LC_ALL, old_locale) return message def finalize_message(config, sub_calendars, open_slots: List[TimeSlot])-> Tuple[str, bool]: message = make_slots_message(config, sub_calendars, open_slots) if message: message = config["header"] + message else: message = config["no_open_slots"] had_message_without_footer = True if message else False message += "\n" + config["footer"] return (message, had_message_without_footer) 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 ) 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)}'." ) open_slots = find_open_slots(config, events, start_date, end_date) (message, had_message_without_footer) = finalize_message(config, sub_calendars, open_slots) if dry_run: print("Messsage that would be sent on Telegram:") if not had_message_without_footer: print("No message would be sent.") else: print(message) else: if had_message_without_footer: 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))