fully implemented project

app/filter.py

@@ -0,0 +1,23 @@
+from ics import Calendar
+
+td_titles = ["TD"]
+sh_titles = ["ANGLAIS", "Anglais", "ETHIQUE"]
+
+def _filter_group(name: str, titles: list[str], group: str) -> bool:
+    for title in titles:
+        if title in name:
+            return group == "all" or group in name
+
+    return True
+
+def _filter_event(name: str, td_group: str, sh_group: str) -> bool:
+    return _filter_group(name, td_titles, td_group) and _filter_group(name, sh_titles, sh_group)
+
+def filter_calendar(calendar: Calendar, td_group: str, sh_group: str) -> Calendar:
+    filtered_calendar = Calendar()
+
+    for event in calendar.events:
+        if _filter_event(event.name, td_group, sh_group):
+            filtered_calendar.events.add(event)
+
+    return filtered_calendar

app/grabber.py

@@ -0,0 +1,18 @@
+from ics import Calendar
+import requests
+import logger
+
+def grab_calendar(ics_url: str):
+    response = requests.get(ics_url)
+
+    if response.status_code != 200:
+        logger.warning("Unable to fetch calendar")
+        return None
+
+    try:
+        calendar = Calendar(response.text)
+    except Exception:
+        logger.warning("Unable to parse calendar")
+        return None
+
+    return calendar

app/handler.py

@@ -0,0 +1,63 @@
+from http.server import BaseHTTPRequestHandler
+import urllib.parse as urllib
+import logger
+
+def generate_class(response_callback):
+    class Handler(BaseHTTPRequestHandler):
+        def _send_error(self, code: int, message: str):
+            self.send_response(code)
+            self.send_header("Content-Type", "text/plain")
+            self.end_headers()
+            self.wfile.write(message.encode("utf-8"))
+
+        def _handle_request(self):
+            host = self.client_address[0]
+            method = self.command
+            url = urllib.urlparse(self.path)
+            path = url.path
+            query = urllib.parse_qs(url.query)
+
+            if method != "GET" or path != "/calendar.ics":
+                logger.info("Invalid " + method + " request on " + path + " from " + host)
+                self._send_error(404, "Invalid endpoint")
+                return
+
+            logger.info("Received calendar request from " + host)
+
+            td_group = (query.get("td") or ["all"])[0]
+            sh_group = (query.get("sh") or ["all"])[0]
+
+            response = response_callback(td_group, sh_group)
+            if not response:
+                self._send_error(500, "Unable to generate calendar")
+                return
+
+            logger.info("Generated calendar for " + host)
+
+            self.send_response(200)
+            self.send_header("Content-Type", "text/calendar")
+            self.end_headers()
+            self.wfile.write(response.encode("utf-8"))
+
+        def do_GET(self):
+            self._handle_request()
+
+        def do_POST(self):
+            self._handle_request()
+
+        def do_PUT(self):
+            self._handle_request()
+
+        def do_DELETE(self):
+            self._handle_request()
+
+        def do_PATCH(self):
+            self._handle_request()
+
+        def do_OPTIONS(self):
+            self._handle_request()
+
+        def log_message(self, fmt, *args):
+            return
+
+    return Handler

app/logger.py

@@ -0,0 +1,18 @@
+import sys
+import threading
+
+_console_lock = threading.Lock()
+
+def _log(prefix: str, message: str):
+    with _console_lock:
+        print(prefix + ": " + message)
+
+def info(message: str):
+    _log("INFO", message)
+
+def warning(message: str):
+    _log("WARNING", message)
+
+def error(message: str):
+    _log("ERROR", message)
+    sys.exit(1)

app/main.py

@@ -1 +1,41 @@
-print("hello")
+import signal
+import sys
+import os
+import logger
+from server import Server
+
+def _handle_exit(sig, frame):
+    print("Exiting...")
+    sys.exit(0)
+
+def _get_environment_variable(key: str) -> str:
+    value = os.getenv(key)
+
+    if not value:
+        logger.error("Environment variable '" + key + "' must be set")
+
+    return value
+
+def _parse_environment_variable(key: str, value: str) -> int | None:
+    try:
+        return int(value)
+    except ValueError:
+        logger.error("Environment variable '" + key + "' must be an integer")
+
+def main():
+    signal.signal(signal.SIGTERM, _handle_exit)
+    signal.signal(signal.SIGINT, _handle_exit)
+
+    host = _get_environment_variable("HOST")
+    port = _get_environment_variable("PORT")
+    cache_duration = _get_environment_variable("CACHE_DURATION")
+    ics_url = _get_environment_variable("ICS_URL")
+
+    port_int = _parse_environment_variable("PORT", port)
+    cache_duration_int = _parse_environment_variable("CACHE_DURATION", cache_duration)
+
+    server = Server(host, port_int, cache_duration_int, ics_url)
+    server.serve()
+
+if __name__ == "__main__":
+    main()

app/requirements.txt

@@ -1 +1,2 @@
 ics==0.7.2
+requests==2.32.5

app/server.py

@@ -0,0 +1,40 @@
+import time
+import filter
+import handler
+from http.server import HTTPServer
+import logger
+import grabber
+
+class Server:
+    def __init__(self, host: str, port: int, cache_duration: int, ics_url: str):
+        self._host = host
+        self._port = port
+        self._cache_duration = cache_duration
+        self._ics_url = ics_url
+        self._cached_calendar = None
+        self._cached_time = -cache_duration
+
+    def _cache_calendar(self):
+        if not self._cached_calendar or self._cached_time + self._cache_duration < time.time():
+            self._cached_calendar = grabber.grab_calendar(self._ics_url)
+            if not self._cached_calendar:
+                return False
+
+            self._cached_time = time.time()
+            logger.info("Successfully cached calendar")
+
+        return True
+
+    def _response_callback(self, td_group: str, sh_group: str):
+        if not self._cache_calendar():
+            return None
+
+        filtered_calendar = filter.filter_calendar(self._cached_calendar, td_group, sh_group)
+
+        return filtered_calendar.serialize()
+
+    def serve(self):
+        handler_class = handler.generate_class(self._response_callback)
+        server = HTTPServer((self._host, self._port), handler_class)
+        logger.info("Listening on " + self._host + ":" + str(self._port))
+        server.serve_forever()