Merge branch 'master' of ssh://ssh.nx2.site:50022/nx2/dotfiles

This commit is contained in:
Lennart J. Kurzweg (Nx2)
2025-05-26 14:46:53 +02:00
4 changed files with 71 additions and 58 deletions

View File

@@ -47,104 +47,113 @@ in {
'';})
(writers.writePython3Bin "caldav_event" {
libraries = with pkgs.python3Packages; [ caldav ics pytz ];
flakeIgnore = [ "E302" "E305""E501" ];
flakeIgnore = [ "E302" "E305""E501" "E261" ];
} /* python */ ''
import os
import json
from caldav import DAVClient
from datetime import datetime, timezone
import json
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)
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)
json.dump(data, file, default=datetime_converter)
def get_ongoing_and_next_event(url, username, password):
def get_ongoing_or_next_event(url, username, password):
now = datetime.now(timezone.utc)
ongoing_events = []
upcoming_events = []
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:
events = calendar.events()
for event in events:
ical_data = event.data
calendar_parsed = Calendar(ical_data)
for event in calendar.events():
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)
for event in calendar_parsed.events:
event_name = event.name or "(No Title)"
start_time = event.begin.astimezone(timezone.utc)
end_time = 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
if start_time <= now <= end_time:
ongoing_events.append((event_name, start_time.timestamp(), end_time.timestamp()))
elif start_time > now:
upcoming_events.append((event_name, start_time.timestamp(), end_time.timestamp()))
except Exception as e:
print(f"Error accessing {url}: {e}")
upcoming_events.sort(key=lambda x: x[1]) # Sort by start time
return ongoing_events, upcoming_events[0] if upcoming_events else None
return None
if __name__ == "__main__":
password_file = "${config.sops.secrets."nx2site/radicale/password".path}" # Path to password file
password_file = "/home/nx2/.config/sops-nix/secrets/nx2site/radicale/password" # Path to password file
cache_file = "/tmp/caldav_event_cache.json" # Path to cache file
url = "https://dav.${domain}/"
username = "${user}"
password = get_password(password_file)
cache = load_cache(cache_file)
event_dict = load_cache(cache_file)
now = datetime.now(timezone.utc).timestamp()
if cache and cache.get("next_event_start") and now < cache["next_event_start"]:
ongoing_events = cache.get("ongoing_events", [])
next_event = (cache["next_event_name"], cache["next_event_start"], cache["next_event_end"]) if "next_event_name" in cache else None
else:
ongoing_events, next_event = get_ongoing_and_next_event(url, username, password)
if event_dict is None or event_dict['event_begin'].timestamp() <= now and now < event_dict['event_end'].timestamp():
event_dict = get_ongoing_or_next_event(url, username, password)
if event_dict is None:
print("No upcoming events found.")
exit(0)
cache_data = {
"ongoing_events": ongoing_events,
"next_event_name": next_event[0] if next_event else None,
"next_event_start": next_event[1] if next_event else None,
"next_event_end": next_event[2] if next_event else None
"event_name": event_dict['event_name'] if event_dict is not None else None,
"event_begin": event_dict['event_begin'] if event_dict is not None else None,
"event_end": event_dict['event_end'] if event_dict is not None else None
}
save_cache(cache_file, cache_data)
if ongoing_events:
for event_name, start_time, end_time in ongoing_events:
time_remaining = end_time - now
hours, rem = divmod(int(time_remaining), 3600)
minutes, _ = divmod(rem, 60)
if event_dict:
event_start = event_dict['event_begin'].timestamp()
event_end = event_dict['event_end'].timestamp()
if hours == 0:
print(f"{event_name} {minutes} minute{'s ' if minutes > 1 else ' '}left")
if event_start <= now <= event_end:
time_remaining = event_end - now
hours, rem = divmod(int(time_remaining), 3600)
minutes, _ = divmod(rem, 60)
print(f"{event_dict['event_name']} ends in {hours} hour{'s ' if hours != 1 else ' '}and {minutes} minute{'s ' if minutes != 1 else ' '}")
else:
print(f"{event_name} {hours} hour{'s ' if hours > 1 else ' '}and {minutes} minute{'s ' if minutes > 1 else ' '}left")
else:
if next_event:
event_name, start_time, end_time = next_event
time_until_start = start_time - now
hours, rem = divmod(int(time_until_start), 3600)
minutes, _ = divmod(rem, 60)
if hours == 0:
print(f"'{event_name}' starts in {minutes} minute{'s ' if minutes > 1 else ' '}")
else:
print(f"'{event_name}' starts in {hours} hour{'s ' if hours > 1 else ' '}and {minutes} minute{'s ' if minutes > 1 else ' '}")
time_until_start = event_start - now
hours, rem = divmod(int(time_until_start), 3600)
minutes, _ = divmod(rem, 60)
print(f"{event_dict['event_name']} starts in {hours} hour{'s ' if hours != 1 else ' '}and {minutes} minute{'s ' if minutes != 1 else ' '}")
else:
print("No upcoming events found.")
'')