feat(users/Profpatsch/sync-abfall-ics-aichach-friedberg): init

A small script that fetches calendar files for our local trash
provider. First step towards integrating ics files into my calendar
setup.

Change-Id: I0e8915a00c19349104cb6256e9dc87c17620fcae
Reviewed-on: https://cl.tvl.fyi/c/depot/+/5883
Tested-by: BuildkiteCI
Reviewed-by: Profpatsch <mail@profpatsch.de>
Autosubmit: Profpatsch <mail@profpatsch.de>

users/Profpatsch/sync-abfall-ics-aichach-friedberg/default.nix

@@ -0,0 +1,14 @@
+{ depot, pkgs, ... }:
+
+let
+  sync-to-dir = depot.users.Profpatsch.writers.python3
+    {
+      name = "sync-ics-to-dir";
+      libraries = (py: [
+        py.httpx
+        py.icalendar
+      ]);
+    } ./sync-ics-to-dir.py;
+
+in
+sync-to-dir

users/Profpatsch/sync-abfall-ics-aichach-friedberg/sync-ics-to-dir.py

@@ -0,0 +1,133 @@
+# horrible little module that fetches ICS files for the local trash public service.
+#
+# It tries its best to not overwrite existing ICS files in case the upstream goes down
+# or returns empty ICS files.
+import sys
+import httpx
+import asyncio
+import icalendar
+from datetime import datetime
+import syslog
+import os.path
+
+# Internal id for the street (extracted from the ics download url)
+ortsteil_id = "e9c32ab3-df25-4660-b88e-abda91897d7a"
+
+# They are using a numeric encoding to refer to different kinds of trash
+fraktionen = {
+    "restmüll": "1",
+    "bio": "5",
+    "papier": "7",
+    "gelbe_tonne": "13",
+    "problemmüllsammlung": "20"
+}
+
+def ics_url(year):
+  frakt = ','.join(fraktionen.values())
+  return f'https://awido.cubefour.de/Customer/aic-fdb/KalenderICS.aspx?oid={ortsteil_id}&jahr={year}&fraktionen={frakt}&reminder=1.12:00'
+
+def fetchers_for_years(start_year, no_of_years_in_future):
+    """given a starting year, and a number of years in the future,
+    return the years for which to fetch ics files"""
+    current_year = datetime.now().year
+    max_year = current_year + no_of_years_in_future
+    return {
+        "passed_years": range(start_year, current_year),
+        "this_and_future_years": range(current_year, 1 + max_year)
+    }
+
+async def fetch_ics(c, url):
+    """fetch an ICS file from an URL"""
+    try:
+        resp = await c.get(url)
+    except Exception as e:
+        return { "ics_does_not_exist_exc": e }
+
+    if resp.is_error:
+        return { "ics_does_not_exist": resp }
+    else:
+        try:
+            ics = icalendar.Calendar.from_ical(resp.content)
+            return { "ics": { "ics_parsed": ics, "ics_bytes": resp.content } }
+        except ValueError as e:
+            return { "ics_cannot_be_parsed": e }
+
+def ics_has_events(ics):
+    """Determine if there is any event in the ICS, otherwise we can assume it’s an empty file"""
+    for item in ics.walk():
+      if isinstance(item, icalendar.Event):
+        return True
+    return False
+
+async def write_nonempty_ics(directory, year, ics):
+    # only overwrite if the new ics has any events
+    if ics_has_events(ics['ics_parsed']):
+        path = os.path.join(directory, f"{year}.ics")
+        with open(path, "wb") as f:
+            f.write(ics['ics_bytes'])
+            info(f"wrote ics for year {year} to file {path}")
+    else:
+        info(f"ics for year {year} was empty, skipping")
+
+
+def main():
+    ics_directory = os.getenv("ICS_DIRECTORY", None)
+    if not ics_directory:
+        critical("please set ICS_DIRECTORY")
+    start_year = int(os.getenv("ICS_START_YEAR", 2022))
+    future_years = int(os.getenv("ICS_FUTURE_YEARS", 2))
+
+    years = fetchers_for_years(start_year, no_of_years_in_future=future_years)
+
+
+    async def go():
+        async with httpx.AsyncClient(follow_redirects=True) as c:
+            info(f"fetching ics for passed years: {years['passed_years']}")
+            for year in years["passed_years"]:
+                match await fetch_ics(c, ics_url(year)):
+                    case { "ics_does_not_exist_exc": error }:
+                       warn(f"The ics for the year {year} is gone, error when requesting: {error} for url {ics_url(year)}")
+                    case { "ics_does_not_exist": resp }:
+                       warn(f"The ics for the year {year} is gone, server returned status {resp.status} for url {ics_url(year)}")
+                    case { "ics_cannot_be_parsed": error }:
+                       warn(f"The returned ICS could not be parsed: {error} for url {ics_url(year)}")
+                    case { "ics": ics }:
+                       info(f"fetched ics from {ics_url(year)}")
+                       await write_nonempty_ics(ics_directory, year, ics)
+                    case _:
+                       critical("unknown case for ics result")
+
+
+            info(f"fetching ics for current and upcoming years: {years['this_and_future_years']}")
+            for year in years["this_and_future_years"]:
+                match await fetch_ics(c, ics_url(year)):
+                    case { "ics_does_not_exist_exc": error }:
+                       critical(f"The ics for the year {year} is not available, error when requesting: {error} for url {ics_url(year)}")
+                    case { "ics_does_not_exist": resp }:
+                       critical(f"The ics for the year {year} is not available, server returned status {resp.status} for url {ics_url(year)}")
+                    case { "ics_cannot_be_parsed": error }:
+                       critical(f"The returned ICS could not be parsed: {error} for url {ics_url(year)}")
+                    case { "ics": ics }:
+                       info(f"fetched ics from {ics_url(year)}")
+                       await write_nonempty_ics(ics_directory, year, ics)
+                    case _:
+                       critical("unknown case for ics result")
+
+    asyncio.run(go())
+
+def info(msg):
+    syslog.syslog(syslog.LOG_INFO, msg)
+
+def critical(msg):
+    syslog.syslog(syslog.LOG_CRIT, msg)
+    sys.exit(1)
+
+def warn(msg):
+    syslog.syslog(syslog.LOG_WARNING, msg)
+
+def debug(msg):
+    syslog.syslog(syslog.LOG_DEBUG, msg)
+
+
+if __name__ == "__main__":
+    main()

users/Profpatsch/writers/default.nix

@@ -15,12 +15,18 @@ let
       string;
   Libraries = defun [ (attrs any) (list drv) ];
 
+  pythonPackages = pkgs.python310Packages;
+  python = pythonPackages.python;
+
   python3 =
     { name
     , libraries ? (_: [ ])
     , flakeIgnore ? [ ]
-    }: pkgs.writers.writePython3 name {
-      libraries = Libraries libraries pkgs.python3Packages;
+    }:
+    let
+    in
+    pkgs.writers.makePythonWriter python pythonPackages name {
+      libraries = Libraries libraries pythonPackages;
       flakeIgnore =
         let
           ignoreTheseErrors = [
@@ -37,6 +43,10 @@ let
             # … between functions
             "E302"
             "E305"
+            # … if there’s too many of them
+            "E303"
+            # or lines that are too long
+            "E501"
           ];
         in
         list FlakeError (ignoreTheseErrors ++ flakeIgnore);
@@ -80,10 +90,10 @@ let
         ]
       ];
     in
-    pkgs.python3Packages.buildPythonPackage {
+    pythonPackages.buildPythonPackage {
       inherit name;
       src = srcTree;
-      propagatedBuildInputs = libraries pkgs.python3Packages;
+      propagatedBuildInputs = libraries pythonPackages;
       doCheck = false;
     };