|
|
|
|
@ -6,6 +6,7 @@ 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
|
|
|
|
|
@ -186,16 +187,22 @@ def convert_to_date(zone, text, days_to_check=None) -> datetime:
|
|
|
|
|
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, start: datetime, end: datetime) -> str:
|
|
|
|
|
|
|
|
|
|
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))
|
|
|
|
|
return f"https://teamup.com/{config['calendar_id']}/events/new?start_dt={url_start}&end_dt={url_end}"
|
|
|
|
|
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 check_slots(events, start_date, end_date):
|
|
|
|
|
message = ""
|
|
|
|
|
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'
|
|
|
|
|
@ -208,27 +215,46 @@ def check_slots(events, start_date, end_date):
|
|
|
|
|
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")
|
|
|
|
|
open_slots += free_time_slots
|
|
|
|
|
|
|
|
|
|
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)
|
|
|
|
|
|
|
|
|
|
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, message):
|
|
|
|
|
|
|
|
|
|
def finalize_message(config, sub_calendars, open_slots: List[TimeSlot])-> str:
|
|
|
|
|
message = make_slots_message(config, sub_calendars, open_slots)
|
|
|
|
|
|
|
|
|
|
if message:
|
|
|
|
|
message = config["header"] + message + "\n" + config["footer"]
|
|
|
|
|
message = config["header"] + message
|
|
|
|
|
else:
|
|
|
|
|
message = config["no_open_slots"]
|
|
|
|
|
|
|
|
|
|
message += "\n" + config["footer"]
|
|
|
|
|
return message
|
|
|
|
|
|
|
|
|
|
async def check_slots_and_notify(config: map, dry_run: bool = False) -> None:
|
|
|
|
|
@ -259,9 +285,9 @@ async def check_slots_and_notify(config: map, dry_run: bool = False) -> None:
|
|
|
|
|
f"Found {len(events)} events in the time range for calendars '{"', '".join(sub_calendar.name for sub_calendar in sub_calendars)}'."
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
message = check_slots(events, start_date, end_date)
|
|
|
|
|
open_slots = find_open_slots(config, events, start_date, end_date)
|
|
|
|
|
|
|
|
|
|
message = finalize_message(config, message)
|
|
|
|
|
message = finalize_message(config, sub_calendars, open_slots)
|
|
|
|
|
|
|
|
|
|
if dry_run:
|
|
|
|
|
print("Messsage that would be sent on Telegram:")
|
|
|
|
|
|