feat: warn about duplicate events

app/main.py

@@ -5,7 +5,10 @@ import hashlib
 import re
 import time
 import threading
+import unicodedata
+from difflib import SequenceMatcher
 from collections import defaultdict, deque
+from html import escape
 from io import BytesIO
 from urllib.parse import urlparse
 from contextlib import asynccontextmanager
@@ -1065,6 +1068,72 @@ def normalize_external_url(value: str, field_name: str):
     return value
 
 
+def normalize_artist_name(value: str) -> str:
+    normalized = unicodedata.normalize("NFKD", value or "")
+    without_accents = "".join(character for character in normalized if not unicodedata.combining(character))
+    return " ".join(re.findall(r"[a-z0-9]+", without_accents.casefold()))
+
+
+def artist_names_similar(first: str, second: str) -> bool:
+    left = normalize_artist_name(first)
+    right = normalize_artist_name(second)
+    if not left or not right:
+        return False
+    if left == right:
+        return True
+    if min(len(left), len(right)) >= 4 and (left in right or right in left):
+        return True
+    if min(len(left), len(right)) < 5:
+        return False
+    return SequenceMatcher(None, left, right).ratio() >= 0.78
+
+
+def find_duplicate_concerts(user, artist: str, start_date: str):
+    try:
+        concert_date = datetime.strptime(start_date[:10], "%Y-%m-%d").date()
+    except (TypeError, ValueError):
+        return []
+    with get_db_connection() as connection:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT c.id, c.artist, c.start_datetime, c.event_type,
+                       COALESCE(v.name, ''), COALESCE(v.city, '')
+                FROM concerts c
+                LEFT JOIN venues v ON v.id = c.venue_id
+                WHERE c.start_datetime::date = %s
+                  AND (
+                      c.visibility = 'public' OR c.created_by = %s OR %s
+                      OR EXISTS (
+                          SELECT 1 FROM event_invitations ei
+                          WHERE ei.concert_id = c.id AND ei.user_id = %s
+                      )
+                      OR (c.visibility = 'friends' AND EXISTS (
+                          SELECT 1 FROM friendships f
+                          WHERE f.status = 'accepted'
+                            AND ((f.requester_id = c.created_by AND f.addressee_id = %s)
+                              OR (f.addressee_id = c.created_by AND f.requester_id = %s))
+                      ))
+                  )
+                ORDER BY c.start_datetime, c.id
+                """,
+                (concert_date, user["id"], user["is_admin"], user["id"], user["id"], user["id"]),
+            )
+            rows = cursor.fetchall()
+    return [
+        {
+            "id": row[0],
+            "artist": row[1],
+            "date": row[2].strftime("%d.%m.%Y"),
+            "time": row[2].strftime("%H:%M"),
+            "event_type": row[3],
+            "venue": ", ".join(part for part in (row[4], row[5]) if part),
+        }
+        for row in rows
+        if artist_names_similar(artist, row[1])
+    ]
+
+
 def save_image(upload: UploadFile, destination_dir: str, url_prefix: str):
     if not upload or not upload.filename:
         return None, None
@@ -3004,6 +3073,14 @@ def new_concert(request: Request):
                            invitable_users=get_invitable_users(user["id"]))
 
 
+@app.get("/api/concerts/duplicates")
+def duplicate_concerts(request: Request, artist: str = "", start_date: str = ""):
+    user = get_current_user(request)
+    if len(artist.strip()) < 2 or len(artist) > 300:
+        return JSONResponse({"matches": []})
+    return JSONResponse({"matches": find_duplicate_concerts(user, artist, start_date)})
+
+
 # ============================================================
 # Concert detail
 # ============================================================
@@ -3208,6 +3285,7 @@ async def create_concert(
     ticket_url: str = Form(""),
     ticket_price: str = Form(""),
     flyer_url: str = Form(""),
+    duplicate_confirmed: bool = Form(False),
     flyer: UploadFile | None = File(None)
 ):
     user = get_current_user(request)
@@ -3215,6 +3293,9 @@ async def create_concert(
     if not user:
         return login_redirect("/concerts/new")
 
+    artist = artist.strip()
+    if len(artist) < 2 or len(artist) > 255:
+        return HTMLResponse("<h1>Der Titel oder Künstlername muss zwischen 2 und 255 Zeichen lang sein.</h1>", status_code=400)
     if event_type not in EVENT_TYPES:
         return HTMLResponse("<h1>Ungültige Veranstaltungskategorie.</h1>", status_code=400)
     if event_type == "festival" and not end_datetime:
@@ -3225,6 +3306,19 @@ async def create_concert(
         return HTMLResponse("<h1>Ungültige Sichtbarkeit.</h1>", status_code=400)
     if event_type != "other":
         visibility = "public"
+    duplicate_matches = find_duplicate_concerts(user, artist, start_datetime)
+    if duplicate_matches and not duplicate_confirmed:
+        match_items = "".join(
+            f'<li><a href="/concerts/{match["id"]}">{escape(match["artist"])} · {match["date"]} {match["time"]}</a></li>'
+            for match in duplicate_matches
+        )
+        return HTMLResponse(
+            "<h1>Mögliche doppelte Veranstaltung</h1>"
+            "<p>Am selben Tag existiert bereits eine Veranstaltung mit einem sehr ähnlichen Künstlernamen.</p>"
+            f"<ul>{match_items}</ul>"
+            "<p>Bitte gehe zurück, prüfe den Treffer und bestätige den Hinweis im Formular, wenn du trotzdem speichern möchtest.</p>",
+            status_code=409,
+        )
     if end_datetime and datetime.fromisoformat(end_datetime) < datetime.fromisoformat(start_datetime):
         return HTMLResponse("<h1>Das Enddatum darf nicht vor dem Beginn liegen.</h1>", status_code=400)
     try: