Kutten Test

app/main.py

@@ -42,9 +42,17 @@ AVATAR_DIR = os.path.join(
     "avatars"
 )
 
+PATCH_DIR = os.path.join(
+    BASE_DIR,
+    "static",
+    "uploads",
+    "patches"
+)
+
 os.makedirs(UPLOAD_DIR, exist_ok=True)
 os.makedirs(PHOTO_DIR, exist_ok=True)
 os.makedirs(AVATAR_DIR, exist_ok=True)
+os.makedirs(PATCH_DIR, exist_ok=True)
 
 SESSION_COOKIE = "pingu_session"
 SESSION_DAYS = 30
@@ -55,11 +63,22 @@ INITIAL_ADMIN_PASSWORD = os.environ.get("INITIAL_ADMIN_PASSWORD")
 INITIAL_ADMIN_EMAIL = os.environ.get("INITIAL_ADMIN_EMAIL")
 
 BADGE_DEFINITIONS = (
-    ("first_gig", "Erster Gig", "🎸", 1, "Dein erstes besuchtes Konzert"),
-    ("regular", "Stammgast", "🤘", 5, "5 Konzerte besucht"),
-    ("ten_gigs", "Zehnerrunde", "🔥", 10, "10 Konzerte besucht"),
-    ("tour_veteran", "Tourveteran", "⚡", 25, "25 Konzerte besucht"),
+    ("beta_tester", "Beta Tester", "🧪", None, "In der Beta dabei", "beta"),
+    ("first_gig", "Erster Gig", "🎸", 1, "Dein erstes besuchtes Konzert", "attendance"),
+    ("regular", "Stammgast", "🤘", 5, "5 Konzerte besucht", "attendance"),
+    ("ten_gigs", "Zehnerrunde", "🔥", 10, "10 Konzerte besucht", "attendance"),
+    ("tour_veteran", "Tourveteran", "⚡", 25, "25 Konzerte besucht", "attendance"),
 )
+ATTENDANCE_BADGE_CODES = tuple(
+    badge_code
+    for badge_code, _name, _icon, _threshold, _description, category in BADGE_DEFINITIONS
+    if category == "attendance"
+)
+BETA_REGISTRATION_DEADLINE = datetime(2026, 9, 16)
+BADGE_BY_CODE = {
+    badge_code: (name, icon, threshold, description, category)
+    for badge_code, name, icon, threshold, description, category in BADGE_DEFINITIONS
+}
 
 
 def get_db_connection():
@@ -147,6 +166,14 @@ def ensure_schema():
             PRIMARY KEY (user_id, badge_code)
         )
         """,
+        """
+        CREATE TABLE IF NOT EXISTS badge_assets (
+            badge_code VARCHAR(50) PRIMARY KEY,
+            path TEXT NOT NULL,
+            updated_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
+            updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+        )
+        """,
     ]
 
     with get_db_connection() as connection:
@@ -496,22 +523,48 @@ def attended_concert_count(user_id: int) -> int:
             return cursor.fetchone()[0]
 
 
-def grant_earned_badges(user_id: int, attended_count: int):
+def grant_earned_badges(user_id: int, attended_count: int, registered_at):
+    highest_attendance_badge = None
+
+    for badge_code, _name, _icon, threshold, _description, category in BADGE_DEFINITIONS:
+        if category == "attendance" and attended_count >= threshold:
+            highest_attendance_badge = badge_code
+
     with get_db_connection() as connection:
         with connection.cursor() as cursor:
-            for badge_code, _name, _icon, threshold, _description in BADGE_DEFINITIONS:
-                if attended_count >= threshold:
-                    cursor.execute(
-                        """
-                        INSERT INTO user_badges (user_id, badge_code)
-                        VALUES (%s, %s)
-                        ON CONFLICT (user_id, badge_code) DO NOTHING
-                        """,
-                        (user_id, badge_code),
-                    )
+            if registered_at < BETA_REGISTRATION_DEADLINE:
+                cursor.execute(
+                    """
+                    INSERT INTO user_badges (user_id, badge_code)
+                    VALUES (%s, 'beta_tester')
+                    ON CONFLICT (user_id, badge_code) DO NOTHING
+                    """,
+                    (user_id,),
+                )
+
+            if highest_attendance_badge:
+                cursor.execute(
+                    "DELETE FROM user_badges WHERE user_id = %s AND badge_code = ANY(%s)",
+                    (user_id, list(ATTENDANCE_BADGE_CODES)),
+                )
+                cursor.execute(
+                    """
+                    INSERT INTO user_badges (user_id, badge_code)
+                    VALUES (%s, %s)
+                    ON CONFLICT (user_id, badge_code) DO NOTHING
+                    """,
+                    (user_id, highest_attendance_badge),
+                )
         connection.commit()
 
 
+def load_badge_assets():
+    with get_db_connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT badge_code, path FROM badge_assets")
+            return {row[0]: row[1] for row in cursor.fetchall()}
+
+
 def load_profile(username: str):
     with get_db_connection() as connection:
         with connection.cursor() as cursor:
@@ -540,6 +593,7 @@ def load_profile(username: str):
         "display_name": row[2] or row[1],
         "avatar_path": row[3],
         "created_at": row[4].strftime("%d.%m.%Y"),
+        "registered_at": row[4],
         "earned_codes": earned_codes,
     }
 
@@ -1134,8 +1188,13 @@ def render_profile(request: Request, username: str):
         return HTMLResponse("<h1>Benutzer nicht gefunden</h1>", status_code=404)
 
     attended_count = attended_concert_count(profile["id"])
-    grant_earned_badges(profile["id"], attended_count)
+    grant_earned_badges(
+        profile["id"],
+        attended_count,
+        profile["registered_at"],
+    )
     profile = load_profile(username)
+    badge_assets = load_badge_assets()
     badges = [
         {
             "code": code,
@@ -1143,9 +1202,11 @@ def render_profile(request: Request, username: str):
             "icon": icon,
             "threshold": threshold,
             "description": description,
+            "category": category,
             "earned": code in profile["earned_codes"],
+            "image_path": badge_assets.get(code),
         }
-        for code, name, icon, threshold, description in BADGE_DEFINITIONS
+        for code, name, icon, threshold, description, category in BADGE_DEFINITIONS
     ]
 
     template = templates.get_template("profile.html")