refactor

system-modules/calendar/dicos.nix

@@ -0,0 +1,140 @@
+{ pkgs, ... }@all: with all;
+{
+  systemd.timers."nx_cal_dicos" = {
+    enable = true;
+    wantedBy = [ "timers.target" ];
+    timerConfig = {
+      OnBootSec = "40m";
+      OnUnitActiveSec = "12h";
+      Unit = "nx_cal_dicos.service";
+    };
+  };
+
+  systemd.services."nx_cal_dicos" = {
+    script = let
+      nx_cal_dicos = (pkgs.writers.writePython3Bin "nx_cal_dicos" {
+        libraries = with pkgs.python3Packages; [
+          ics
+        ];
+        flakeIgnore = [ "E302" "E305" "E226" "E501" ];
+      } /* python */ ''
+import os
+from glob import glob
+from ics import Calendar
+from ics.event import datetime
+
+NETTO_STUNDE = 18.46
+WEEKLY = 12
+
+# week_dict = {}
+# latest_week = 0
+# latest_goal = WEEKLY
+deficit = 0
+
+def fraction_to_unicode(frac):
+    div, rem = divmod(frac, 1)
+    if rem == 0.5:
+        unicode = "½"
+    elif rem == 0.25:
+        unicode = "¼"
+    elif rem == 0.75:
+        unicode = "¾"
+    elif rem == 0:
+        unicode = ""
+    else:
+        unicode = rem
+    if div == 0:
+        h = ""
+    else:
+        h = int(div)
+    return f"{h}{unicode}"
+
+def modify_event(event):
+    """Modify the event if it contains 'DICOS' in the SUMMARY."""
+    # global week_dict
+    # global latest_goal
+    # global latest_week
+    # global deficit
+
+    if event.name is not None and "DICOS" in event.name:
+        length = (event.end - event.begin).seconds / 3600
+        money_made = divmod(length * NETTO_STUNDE, 1)
+
+        # Calculate total hours for DICOS events in the same week
+        year, week, _ = event.begin.isocalendar()
+
+        # if week != latest_week:
+        #     try:
+        #         deficit = latest_goal - week_dict[f"{year}_{latest_week}"]
+        #     except KeyError:
+        #         deficit = 0
+
+        # week_dict[f"{year}_{week}"] = length + (week_dict[f"{year}_{week}"] if f"{year}_{week}" in week_dict else 0)
+
+        # progress = week_dict[f"{year}_{week}"]
+        # goal = WEEKLY + deficit
+
+        # if week != latest_week:
+        #     latest_goal = goal
+        #     latest_week = week
+
+        try:
+            new_description = [event.description.split("\n")[0]]
+        except AttributeError:
+            new_description = ["::"]
+        new_description.append("")
+        new_description.append(f"Netto: {money_made[0]:.0f},{int(money_made[1] * 10):02d}€")
+        # new_description.append(f"This weeks porgress: ({fraction_to_unicode(progress)}/{fraction_to_unicode(goal)})")
+        # new_description.append(f"You're {fraction_to_unicode(abs(deficit))}h in the {'plus' if deficit < 0 else 'minus'} this week.")
+
+        event.description = "\n".join(new_description)
+
+        event.name = f"DICOS {fraction_to_unicode(length)}"
+    return event
+
+def process_ics_file(filepath):
+    """Read, modify, and overwrite an ICS file."""
+    with open(filepath, 'r') as f:
+        calendar = Calendar(f.read())
+
+    modified = False
+
+    for event in calendar.events:
+        if event.name is not None and 'DICOS' in event.name:
+            event = modify_event(event)
+            modified = True
+
+    if modified:
+        with open(filepath, 'w') as f:
+            f.writelines(calendar.serialize_iter())
+
+def get_event_start_time(filepath):
+    """Extract the event's start time from an ICS file."""
+    with open(filepath, 'r') as f:
+        calendar = Calendar(f.read())
+
+    for event in calendar.events:
+        return event.begin.datetime
+    else:
+        return datetime(year=1, month=1, day=1)
+
+
+if __name__ == "__main__":
+    directory = "/var/lib/radicale/collections/collection-root/nx2/experience"
+    ics_files = glob(os.path.join(directory, "*.ics"))
+    if not ics_files:
+        print("No ICS files found in the directory.")
+    sorted_files = sorted(ics_files, key=get_event_start_time)
+    for ics_file in sorted_files:
+        process_ics_file(ics_file)
+    print("Processing complete.")
+'');
+    in ''
+      ${nx_cal_dicos}/bin/nx_cal_dicos
+    '';
+    serviceConfig = {
+      Type = "oneshot";
+      User = "radicale";
+    };
+  };
+}

system-modules/calendar/lec.nix

@@ -0,0 +1,89 @@
+{ pkgs, ... }@all: with all;
+{
+  systemd.timers."nx_cal_lec" = {
+    enable = true;
+    wantedBy = [ "timers.target" ];
+    timerConfig = {
+      OnBootSec = "40m";
+      OnUnitActiveSec = "24h";
+      Unit = "nx_cal_lec.service";
+    };
+  };
+
+  systemd.services."nx_cal_lec" = {
+    script = let 
+      nx_cal_lec = (pkgs.writers.writePython3Bin "nx_cal_lec" {
+      libraries = with pkgs.python3Packages; [
+        ical
+        ics
+        requests
+        dateutils
+      ];
+      flakeIgnore = [ "E302" "E305" "E226" "E501" ];
+    } /*python */ ''
+from ics import Calendar
+import requests
+from datetime import timedelta
+
+def adjust_events(events):
+    """
+    Adjust overlapping events to ensure they do not conflict.
+    """
+    sorted_events = sorted(events, key=lambda e: e.begin)
+    for i in range(1, len(sorted_events)):
+        previous_event = sorted_events[i - 1]
+        current_event = sorted_events[i]
+
+        if current_event.begin < previous_event.end:
+            # Adjust the start time of the current event to just after the previous event
+            current_event.begin = previous_event.end + timedelta(minutes=1)
+            print(f"Adjusted event '{current_event.name}' to start at {current_event.begin} and end at {current_event.end}")
+    return sorted_events
+
+def fetch_and_save_ical_events(ical_url, save_path):
+    """
+    Fetch events from an iCal URL and save them as a single combined calendar.
+    """
+    try:
+        # Fetch the iCal data
+        response = requests.get(ical_url)
+        response.raise_for_status()
+
+        # Parse the iCal data
+        calendar = Calendar(response.text)
+
+        # Adjust events
+        adjusted_events = adjust_events(list(calendar.events))
+
+        # Create a new combined calendar
+        combined_calendar = Calendar()
+        for event in adjusted_events:
+            combined_calendar.events.add(event)
+
+        # Save the combined calendar to a single .ics file
+        with open(save_path, 'w') as file:
+            file.writelines(combined_calendar.serialize_iter())
+
+        print(f"Saved combined calendar to {save_path}")
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching iCal data: {e}")
+    except Exception as e:
+        print(f"Error processing iCal data: {e}")
+
+if __name__ == "__main__":
+    # Replace with your iCal URL and target file path
+    ICAL_URL = "https://zlypher.github.io/lol-events/cal/league-of-legends-lec.ical"
+    SAVE_PATH = "${config.services.nginx.virtualHosts."${hyper.domain}".root}/lec.ics"
+
+    fetch_and_save_ical_events(ICAL_URL, SAVE_PATH)
+'');
+    in ''
+      ${nx_cal_lec}/bin/nx_cal_lec
+    '';
+    serviceConfig = {
+      Type = "oneshot";
+      User = hyper.user;
+    };
+  };
+}

system-modules/calendar/lr.nix

@@ -0,0 +1,79 @@
+{ pkgs, ... }@all: with all;
+{
+  systemd.timers."nx_cal_lr" = {
+    enable = true;
+    wantedBy = [ "timers.target" ];
+    timerConfig = {
+      OnBootSec = "41m";
+      OnUnitActiveSec = "24h";
+      Unit = "nx_cal_lr.service";
+    };
+  };
+
+  systemd.services."nx_cal_lr" = {
+    script = let 
+      nx_cal_lr = (pkgs.writers.writePython3Bin "nx_cal_lr" {
+      libraries = with pkgs.python3Packages; [
+        ics
+        requests
+      ];
+      flakeIgnore = [ "E302" "E305" "E226" "E501" ];
+    } /*python */ ''
+from ics import Calendar
+import requests
+
+def filter_events(events):
+    return [event for event in events if ("LR" in event.name) or ("TBD" in event.name)]
+
+def fetch_and_save_ical_events(ical_urls, save_path):
+    """
+    Fetch events from an iCal URL and save them as a single combined calendar.
+    """
+    try:
+        # Create a new combined calendar
+        combined_calendar = Calendar()
+
+        for url in ical_urls:
+            # Fetch the iCal data
+            response = requests.get(url)
+            response.raise_for_status()
+
+            # Parse the iCal data
+            calendar = Calendar(response.text)
+
+            # Adjust events
+            adjusted_events = filter_events(list(calendar.events))
+
+            for event in adjusted_events:
+                combined_calendar.events.add(event)
+
+        # Save the combined calendar to a single .ics file
+        with open(save_path, 'w') as file:
+            file.writelines(combined_calendar.serialize_iter())
+
+        print(f"Saved combined calendar to {save_path}")
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching iCal data: {e}")
+    except Exception as e:
+        print(f"Error processing iCal data: {e}")
+
+if __name__ == "__main__":
+    # Replace with your iCal URL and target file path
+    ICAL_URLS = [
+        "https://zlypher.github.io/lol-events/cal/league-of-legends-nlc.ical",
+        "https://zlypher.github.io/lol-events/cal/league-of-legends-emea-masters.ical"
+    ]
+    SAVE_PATH = "${config.services.nginx.virtualHosts."${hyper.domain}".root}/lr.ics"
+
+    fetch_and_save_ical_events(ICAL_URLS, SAVE_PATH)
+'');
+    in ''
+      ${nx_cal_lr}/bin/nx_cal_lr
+    '';
+    serviceConfig = {
+      Type = "oneshot";
+      User = hyper.user;
+    };
+  };
+}

system-modules/calendar/publish.nix

@@ -0,0 +1,136 @@
+{ pkgs, hyper, ... }@all: with all; let 
+  radicale-root = "/var/lib/radicale";
+  web-root = "/var/nginx/webroot";
+in {
+  systemd.timers."nx_cal_publish" = {
+    enable = true;
+    wantedBy = [ "timers.target" ];
+    timerConfig = {
+      OnBootSec = "2m";
+      OnUnitActiveSec = "6h";
+      Unit = "nx_cal_publish.service";
+    };
+  };
+
+  systemd.services."nx_cal_publish" = {
+    script = with pkgs; let 
+      nx_cal_publish = (writers.writePython3Bin "nx_cal_publish" {
+        libraries = with python3Packages; [ 
+          ical
+          ics
+          requests
+          dateutils
+        ];
+        flakeIgnore = [ "E302" "E305" "E226" "E501" ];
+      } /*python */ ''
+import pytz
+import os
+from ics import Calendar, Event
+from ics.grammar.parse import ContentLine
+from dateutil.rrule import rrulestr
+from ics.event import datetime, timedelta
+
+def combine_ics_from_directories(directories, output_file):
+    """
+    Combine all .ics events from a list of directories into one .ics file, supporting recurring events.
+
+    :param directories: List of directories containing .ics files.
+    :param output_file: Path to the output .ics file.
+    """
+    combined_calendar = Calendar()
+
+    for directory in directories:
+        if not os.path.exists(directory):
+            print(f"Directory '{directory}' does not exist. Skipping.")
+            continue
+
+        for filename in os.listdir(directory):
+            if filename.endswith(".ics"):
+                file_path = os.path.join(directory, filename)
+                try:
+                    with open(file_path, 'r') as file:
+                        calendar = Calendar(file.read())
+                        for event in calendar.events:
+                            # Handle recurring events
+                            rrule_line = None
+                            for line in event.extra:
+                                if isinstance(line, ContentLine) and line.name == "RRULE":
+                                    rrule_line = line
+                                    break
+
+                            if rrule_line:
+                                # Convert UNTIL to UTC if DTSTART is timezone-aware
+                                rrule_params = rrule_line.value.split(";")
+                                rrule_dict = {}
+                                for param in rrule_params:
+                                    key, value = param.split("=")
+                                    rrule_dict[key] = value
+
+                                if "UNTIL" in rrule_dict and event.begin.tzinfo:
+                                    until = datetime.fromisoformat(rrule_dict["UNTIL"])
+                                    if until.tzinfo is None:  # If UNTIL is naive, make it UTC
+                                        until = until.astimezone(pytz.UTC)
+                                    rrule_dict["UNTIL"] = until.astimezone(pytz.UTC).strftime("%Y%m%dT%H%M%SZ")
+
+                                # Reconstruct RRULE string
+                                rrule_fixed = ";".join(f"{key}={value}" for key, value in rrule_dict.items())
+                                rrule = rrulestr(rrule_fixed, dtstart=event.begin.astimezone(pytz.timezone('CET')))
+
+                                # Expand recurring events and filter based on the date
+                                for occurrence in rrule:
+                                    notTooOld = occurrence.date() >= (datetime.now().astimezone(pytz.UTC) - timedelta(days=1)).date()
+                                    notTooFuturisic = occurrence.date() < (datetime.now().astimezone(pytz.UTC) + timedelta(days=60)).date()
+                                    if notTooOld and notTooFuturisic:
+                                        new_event = Event(
+                                            name="",
+                                            begin=occurrence,
+                                            end=occurrence + (event.end - event.begin),
+                                            transparent=event.transparent or True,
+                                        )
+                                        combined_calendar.events.add(new_event)
+                            else:
+                                # Regular events, directly add if within date range
+                                if event.begin.astimezone(pytz.timezone('CET')).date() >= (datetime.now().astimezone(pytz.timezone('CET')) - timedelta(days=1)).date():
+                                    new_event = Event(
+                                        name="",
+                                        begin=event.begin,
+                                        end=event.end,
+                                        transparent=event.transparent or True,
+                                    )
+                                    combined_calendar.events.add(new_event)
+
+                except Exception as e:
+                    print(f"Error reading file '{file_path}': {e}")
+                    exit(1)
+
+    try:
+        with open(output_file, 'w') as file:
+            file.writelines(combined_calendar.serialize_iter())
+        print(f"Combined .ics file saved to '{output_file}'")
+    except Exception as e:
+        print(f"Error saving combined .ics file: {e}")
+
+if __name__ == "__main__":
+    # List of directories containing .ics files
+    DIRECTORIES = [
+        "${radicale-root}/collections/collection-root/${hyper.user}/preservation",
+        "${radicale-root}/collections/collection-root/${hyper.user}/effort",
+        "${radicale-root}/collections/collection-root/${hyper.user}/experience",
+        "${radicale-root}/collections/collection-root/${hyper.user}/exposure",
+        "${radicale-root}/collections/collection-root/${hyper.user}/engagement",
+    ]
+
+    # Path to the output .ics file
+    OUTPUT_FILE = "${web-root}/schedule.ics"
+
+    combine_ics_from_directories(DIRECTORIES, OUTPUT_FILE)
+''); 
+    in ''
+      ${nx_cal_publish}/bin/nx_cal_publish
+    '';
+    serviceConfig = {
+      Type = "oneshot";
+      User = hyper.user;
+    };
+  };
+}