ical test

endpoints/api.py

@@ -8,6 +8,8 @@
 
 from flask import Blueprint, request, Response
 from flask_login import login_required, current_user
+
+import utils
 from endpoints.authorization import school_authorized
 import endpoints.webpush
 
@@ -143,6 +145,47 @@ def plan(school_num: str) -> Response:
     return send_success(data)
 
 
+@api.route(f"/api/v69.420/plan_ical/<token>", methods=["GET"])
+def plan_ical(token: str) -> Response:
+    out = utils.resolve_ical_token(token)
+    if out is None:
+        return Response(status=404)
+
+    user_id, favorite_index = out
+
+    user = utils.User(user_id)
+
+    favs = user.get_field("favourites", {})
+
+    try:
+        fav = favs[favorite_index]
+    except IndexError:
+        return Response(status=404)
+
+    preferences = fav.get("preferences", [])  # disabled class ids
+
+    from ical.calendar import Calendar
+    from ical.event import Event
+    from ical.calendar_stream import IcsCalendarStream
+
+    calendar = Calendar()
+    calendar.events.append(
+        Event(summary="Event summary", start=datetime.date(2022, 7, 3), end=datetime.date(2022, 7, 4)),
+    )
+
+    ical_str = IcsCalendarStream.calendar_to_ics(calendar)
+
+    return Response(ical_str, mimetype="text/calendar")
+
+
+@api.route(f"/api/v69.420/plan_ical_renew_links", methods=["GET"])
+@login_required
+def plan_ical_renew_links() -> Response:
+    current_user.generate_ical_tokens(overwrite=True)
+
+    return send_success()
+
+
 @api.route(f"{API_BASE_URL}/authorize", methods=["POST"])
 @login_required
 def authorize(school_num: str) -> Response:
@@ -201,7 +244,8 @@ def instant_authorize(school_num: str) -> Response:
 @login_required
 def favorites() -> Response:
     if request.method == "GET":
-        return send_success(current_user.get_user().get("favourites", []))
+        current_user.generate_ical_tokens()
+        return send_success(current_user.get_field("favourites", []))
     try:
         data = json.loads(request.data)
     except json.JSONDecodeError:

requirements.txt

@@ -12,4 +12,5 @@ discord-webhook~=1.3.0
 brotlicffi~=1.1.0.0
 pywebpush~=1.14.0
 py-vapid~=1.9.0
-aiohttp~=3.9.3
+aiohttp~=3.9.3
+ical~=8.1.1

utils.py

@@ -1,3 +1,4 @@
+import secrets
 import threading
 import os
 import json
@@ -25,6 +26,7 @@
 assert shared.mongodb.DATABASE is not None
 
 users = shared.mongodb.DATABASE.get_collection("users")
+ical_tokens = shared.mongodb.DATABASE.get_collection("ical_tokens")
 creds = shared.mongodb.DATABASE.get_collection("creds")
 VALID_SCHOOLS = [elem["_id"] for elem in list(creds.find({}))]
 meta = shared.mongodb.DATABASE.get_collection("meta")
@@ -103,11 +105,7 @@ def update_settings(self, user_settings=DEFAULT_SETTINGS) -> Response:
 
     # get setting for user, if setting not set get default setting
     def get_setting(self, setting_key):
-        self.get_user()
-        return self.user.get("settings", {}).get(setting_key, DEFAULT_SETTINGS.get(setting_key, None))
-
-    def get_favorites(self) -> Response:
-        return send_success(self.user.get("favourites", {}))
+        return self.get_field("settings", {}).get(setting_key, DEFAULT_SETTINGS.get(setting_key, None))
 
     """
         what does a favorite need:
@@ -159,6 +157,40 @@ def set_favorites(self, favorites) -> Response:
         users.update_one({'_id': ObjectId(self.mongo_id)}, {"$set": {'favourites': new_favorites}})
         return send_success(new_favorites)
 
+    def generate_ical_tokens(self, overwrite: bool = False):
+        if not overwrite:
+            if ical_tokens.find_one({"user_id": self.mongo_id}):
+                return
+
+        ical_tokens.delete_many({"user_id": self.mongo_id})
+
+        favs = self.get_field("favourites", [])
+        new_favs = []
+
+        for i, fav in enumerate(favs):
+            token = secrets.token_hex(32)
+            new_favs.append(
+                fav | {
+                    "token": token,
+                }
+            )
+
+            ical_tokens.update_one(
+                {"token": token},
+                {"$set": {"user_id": self.mongo_id, "favorite_index": i}},
+                upsert=True
+            )
+
+        users.update_one({'_id': ObjectId(self.mongo_id)}, {"$set": {'favourites': new_favs}})
+
+
+def resolve_ical_token(token: str) -> tuple[str, int] | None:
+    token_data = ical_tokens.find_one({"token": token})
+    if token_data is None:
+        return None
+    else:
+        return token_data["user_id"], token_data["favorite_index"]
+
 
 def get_user(user_id):
     try:
@@ -175,6 +207,7 @@ def wrapper(*args, **kwargs):
         if not current_user.user.get("admin"):
             return send_error("Du bist kein Admin")
         return func(*args, **kwargs)
+
     wrapper.__name__ = func.__name__
     return wrapper