127 lines
4.3 KiB
Nix
127 lines
4.3 KiB
Nix
{ pkgs, ... }@all: with all; {
|
|
sops.secrets = {
|
|
"nx2site/radicale/password" = { };
|
|
};
|
|
home.packages = [
|
|
(pkgs.writers.writePython3Bin "caldav_event" {
|
|
libraries = with pkgs.python3Packages; [ caldav ics pytz ];
|
|
flakeIgnore = [ "E302" "E305" "E501" "E261" ];
|
|
} /* python */ ''
|
|
import os
|
|
import json
|
|
from caldav import DAVClient
|
|
from datetime import datetime, timezone
|
|
from ics import Calendar
|
|
from pytz import UTC
|
|
|
|
def get_password(password_file):
|
|
with open(password_file, "r") as file:
|
|
return file.read().strip()
|
|
|
|
def datetime_converter(obj):
|
|
if isinstance(obj, datetime):
|
|
return obj.isoformat()
|
|
return obj
|
|
|
|
def datetime_parser(dct):
|
|
for key, value in dct.items():
|
|
if isinstance(value, str):
|
|
try:
|
|
dct[key] = datetime.fromisoformat(value)
|
|
except ValueError:
|
|
pass
|
|
return dct
|
|
|
|
def load_cache(cache_file):
|
|
if os.path.exists(cache_file):
|
|
with open(cache_file, "r") as file:
|
|
return json.load(file, object_hook=datetime_parser)
|
|
return None
|
|
|
|
def save_cache(cache_file, data):
|
|
with open(cache_file, "w") as file:
|
|
json.dump(data, file, default=datetime_converter, indent=4)
|
|
|
|
|
|
def get_ongoing_or_next_event(url, username, password):
|
|
now = datetime.now(timezone.utc)
|
|
|
|
try:
|
|
client = DAVClient(url, username=username, password=password)
|
|
principal = client.principal()
|
|
calendars = principal.calendars()
|
|
|
|
next_event_dict = {
|
|
'event_name': "fake",
|
|
'event_begin': datetime(9000, 1, 1, tzinfo=UTC), # in the year 9000
|
|
'event_end': datetime(9000, 1, 1, 8, tzinfo=UTC),
|
|
}
|
|
|
|
for calendar in calendars:
|
|
for event in calendar.search(start=now):
|
|
calendar_parsed = Calendar(event.data)
|
|
for ics_event in calendar_parsed.events:
|
|
event_dict = {}
|
|
event_dict['event_name'] = ics_event.name or "(No Title)"
|
|
event_dict['event_begin'] = ics_event.begin.astimezone(timezone.utc)
|
|
event_dict['event_end'] = ics_event.end.astimezone(timezone.utc)
|
|
|
|
if event_dict['event_begin'] <= now and now <= event_dict['event_end']:
|
|
return event_dict
|
|
elif event_dict['event_begin'] >= now and next_event_dict['event_begin'] > event_dict['event_begin']:
|
|
next_event_dict = event_dict
|
|
return next_event_dict
|
|
|
|
except Exception as e:
|
|
print(f"Error accessing {url}: {e}")
|
|
return None
|
|
|
|
def is_expired(event_dict: dict):
|
|
now = datetime.now(timezone.utc).timestamp()
|
|
event_end = event_dict['event_end'].timestamp()
|
|
return not (now <= event_end)
|
|
|
|
if __name__ == "__main__":
|
|
password_file = "${config.sops.secrets."nx2site/radicale/password".path}" # Path to password file
|
|
cache_file = "/tmp/caldav_event_cache.json" # Path to cache file
|
|
url = "https://dav.${hyper.domain}/"
|
|
username = "nx2"
|
|
password = get_password(password_file)
|
|
now = datetime.now(timezone.utc).timestamp()
|
|
|
|
event_dict = load_cache(cache_file)
|
|
|
|
if (event_dict is None) or (is_expired(event_dict)):
|
|
event_dict = get_ongoing_or_next_event(url, username, password)
|
|
save_cache(cache_file, event_dict)
|
|
|
|
if event_dict is None: # none were found
|
|
print("* zen *")
|
|
exit(0)
|
|
|
|
event_start = event_dict['event_begin'].timestamp()
|
|
event_end = event_dict['event_end'].timestamp()
|
|
|
|
if event_start <= now <= event_end: # is currently ongoing
|
|
action_string = "ends"
|
|
t = event_end - now # time_remaining
|
|
else: # is in the future
|
|
action_string = "starts"
|
|
t = event_start - now # time_remaining
|
|
|
|
hours, rem = divmod(int(t), 3600)
|
|
minutes, _ = divmod(rem, 60)
|
|
hour_string = f"{hours} hour{'s ' if hours != 1 else ' '}" if hours > 0 else ""
|
|
minu_string = f"{minutes} minute{'s ' if minutes != 1 else ' '}" if minutes > 0 else ""
|
|
if hour_string == "" and minu_string == "":
|
|
time_string = "now"
|
|
elif hour_string == "" or minu_string == "":
|
|
time_string = "in " + hour_string + minu_string
|
|
else:
|
|
time_string = "in " + hour_string + "and " + minu_string
|
|
|
|
print(f"{event_dict['event_name']} {action_string} {time_string}")
|
|
'')
|
|
];
|
|
}
|