import os
import uuid
import secrets
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
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import bcrypt
import httpx
import psycopg
try:
from PIL import Image, ImageOps, UnidentifiedImageError
except ImportError:
Image = ImageOps = UnidentifiedImageError = None
from fastapi import FastAPI, File, Form, Request, UploadFile
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse
from fastapi.encoders import jsonable_encoder
from fastapi.staticfiles import StaticFiles
from jinja2 import Environment, FileSystemLoader, select_autoescape
DATABASE_URL = os.environ["DATABASE_URL"]
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
UPLOAD_DIR = os.path.join(
BASE_DIR,
"static",
"uploads",
"flyers"
)
PHOTO_DIR = os.path.join(
BASE_DIR,
"static",
"uploads",
"photos"
)
AVATAR_DIR = os.path.join(
BASE_DIR,
"static",
"uploads",
"avatars"
)
PATCH_DIR = os.path.join(
BASE_DIR,
"static",
"uploads",
"patches"
)
DIARY_PHOTO_DIR = os.path.join(
os.environ.get("PRIVATE_UPLOAD_DIR", os.path.join(BASE_DIR, "private_uploads")),
"diary"
)
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)
os.makedirs(DIARY_PHOTO_DIR, exist_ok=True)
SESSION_COOKIE = "pingu_session"
SESSION_DAYS = 30
COOKIE_SECURE = os.environ.get("COOKIE_SECURE", "false").lower() in {"1", "true", "yes"}
ALLOWED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"}
MAX_IMAGE_BYTES = 10 * 1024 * 1024
MAX_IMAGE_PIXELS = 25_000_000
rate_limit_buckets = defaultdict(deque)
rate_limit_lock = threading.Lock()
INITIAL_ADMIN_USERNAME = os.environ.get("INITIAL_ADMIN_USERNAME")
INITIAL_ADMIN_PASSWORD = os.environ.get("INITIAL_ADMIN_PASSWORD")
INITIAL_ADMIN_EMAIL = os.environ.get("INITIAL_ADMIN_EMAIL")
BADGE_DEFINITIONS = (
("founder", "Gründer", "⚔️", None, "Von Anfang an dabei und MetalCircle aufgebaut", "special"),
("admin", "Admin", "🏴☠️", None, "Verantwortung für MetalCircle", "special"),
("captns_mate", "Captns Mate", "☠️", None, "Die treue Gefährtin des Captains", "special"),
("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 am selben Veranstaltungsort besucht", "venue"),
("ten_gigs", "10 Gigs", "🔥", 10, "10 besuchte Konzerte", "attendance"),
("tour_veteran", "25 Gigs", "⚡", 25, "25 besuchte Konzerte", "attendance"),
("fifty_gigs", "50 Gigs", "💀", 50, "50 besuchte Konzerte", "attendance"),
("hundred_gigs", "100 Gigs", "👑", 100, "100 besuchte Konzerte", "attendance"),
("border_breaker", "BORDER BREAKER", "🛂", 1, "Erstes Konzert im Ausland", "travel"),
("globe_banger", "GLOBE BANGER", "🌍", 5, "Konzerte in 5 Ländern", "travel"),
("double_trouble", "DOUBLE TROUBLE", "⚡", 2, "Zwei Konzerte an zwei aufeinanderfolgenden Tagen", "streak"),
("iron_weekend", "IRON WEEKEND", "🤘", 3, "Drei Konzerte innerhalb eines Wochenendes", "streak"),
)
ATTENDANCE_BADGE_CODES = tuple(
badge_code
for badge_code, _name, _icon, _threshold, _description, category in BADGE_DEFINITIONS
if category == "attendance"
)
VENUE_BADGE_CODES = tuple(
badge_code for badge_code, _name, _icon, _threshold, _description, category in BADGE_DEFINITIONS
if category == "venue"
)
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
}
BADGE_CATEGORY_LABELS = {
"special": "⚔️ Spezial-Patches",
"beta": "🧪 Community",
"attendance": "🎸 Konzert-Meilensteine",
"venue": "🤘 Stammorte",
"travel": "🌍 Unterwegs",
"streak": "⚡ Tourmodus",
}
def get_db_connection():
return psycopg.connect(DATABASE_URL)
def ensure_schema():
statements = [
"""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
username VARCHAR(50) NOT NULL UNIQUE,
email VARCHAR(255) NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
display_name VARCHAR(100),
is_admin BOOLEAN NOT NULL DEFAULT FALSE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
"""
ALTER TABLE users
ADD COLUMN IF NOT EXISTS is_admin BOOLEAN NOT NULL DEFAULT FALSE
""",
"""
UPDATE users
SET is_admin = TRUE
WHERE LOWER(username) = 'kai'
""",
"""
ALTER TABLE users
ADD COLUMN IF NOT EXISTS avatar_path TEXT
""",
"""
CREATE TABLE IF NOT EXISTS registration_invites (
id SERIAL PRIMARY KEY,
token_hash TEXT NOT NULL UNIQUE,
expires_at TIMESTAMP,
used_by INTEGER REFERENCES users(id),
used_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
"""
CREATE TABLE IF NOT EXISTS sessions (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
token_hash TEXT NOT NULL UNIQUE,
expires_at TIMESTAMP NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
"""
ALTER TABLE concerts
ADD COLUMN IF NOT EXISTS created_by INTEGER REFERENCES users(id) ON DELETE SET NULL
""",
"""
CREATE TABLE IF NOT EXISTS concert_comments (
id SERIAL PRIMARY KEY,
concert_id INTEGER NOT NULL REFERENCES concerts(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
body TEXT NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
"""
CREATE TABLE IF NOT EXISTS concert_photos (
id SERIAL PRIMARY KEY,
concert_id INTEGER NOT NULL REFERENCES concerts(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
path TEXT NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
"""
CREATE TABLE IF NOT EXISTS concert_attendance (
concert_id INTEGER NOT NULL REFERENCES concerts(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
status VARCHAR(20) NOT NULL CHECK (
status IN ('attending', 'maybe', 'ticket_search', 'ticket_offer')
),
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (concert_id, user_id)
)
""",
"""
DO $$
BEGIN
IF EXISTS (
SELECT 1 FROM pg_constraint
WHERE conname = 'concert_attendance_status_check'
AND NOT (
pg_get_constraintdef(oid) LIKE '%ticket_search%'
AND pg_get_constraintdef(oid) LIKE '%ticket_offer%'
)
) THEN
ALTER TABLE concert_attendance DROP CONSTRAINT concert_attendance_status_check;
ALTER TABLE concert_attendance ADD CONSTRAINT concert_attendance_status_check
CHECK (status IN ('attending', 'maybe', 'ticket_search', 'ticket_offer'));
END IF;
END $$
""",
"""
CREATE TABLE IF NOT EXISTS user_badges (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
badge_code VARCHAR(50) NOT NULL,
trigger_concert_id INTEGER REFERENCES concerts(id) ON DELETE SET NULL,
awarded_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, badge_code)
)
""",
"""
ALTER TABLE user_badges ADD COLUMN IF NOT EXISTS trigger_concert_id INTEGER REFERENCES concerts(id) ON DELETE SET NULL
""",
"""
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
)
""",
"""
ALTER TABLE venues
ADD COLUMN IF NOT EXISTS is_verified BOOLEAN NOT NULL DEFAULT FALSE
""",
"""
CREATE TABLE IF NOT EXISTS venue_aliases (
id SERIAL PRIMARY KEY,
venue_id INTEGER NOT NULL REFERENCES venues(id) ON DELETE CASCADE,
alias VARCHAR(255) NOT NULL,
UNIQUE (venue_id, alias)
)
""",
"""
CREATE INDEX IF NOT EXISTS idx_venue_aliases_alias
ON venue_aliases (LOWER(alias))
""",
"""
ALTER TABLE concerts
ADD COLUMN IF NOT EXISTS event_type VARCHAR(20) NOT NULL DEFAULT 'concert'
CHECK (event_type IN ('concert', 'festival', 'other'))
""",
"""
ALTER TABLE concerts
ADD COLUMN IF NOT EXISTS parent_event_id INTEGER
REFERENCES concerts(id) ON DELETE SET NULL
""",
"""
CREATE INDEX IF NOT EXISTS idx_concerts_parent_event_id
ON concerts (parent_event_id)
""",
"""
ALTER TABLE users
ADD COLUMN IF NOT EXISTS instagram_url VARCHAR(500)
""",
"""
ALTER TABLE users DROP COLUMN IF EXISTS email_verified
""",
"""
CREATE TABLE IF NOT EXISTS account_tokens (
id SERIAL PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
purpose VARCHAR(30) NOT NULL CHECK (purpose = 'password_reset'),
token_hash TEXT NOT NULL UNIQUE,
expires_at TIMESTAMP NOT NULL,
used_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""",
"""
DO $$
BEGIN
IF EXISTS (
SELECT 1 FROM pg_constraint
WHERE conname = 'account_tokens_purpose_check'
AND pg_get_constraintdef(oid) LIKE '%verify_email%'
) THEN
DELETE FROM account_tokens WHERE purpose = 'verify_email';
ALTER TABLE account_tokens DROP CONSTRAINT account_tokens_purpose_check;
ALTER TABLE account_tokens ADD CONSTRAINT account_tokens_purpose_check
CHECK (purpose = 'password_reset');
END IF;
END $$
""",
"""
CREATE INDEX IF NOT EXISTS idx_account_tokens_lookup
ON account_tokens (token_hash, purpose, expires_at)
""",
"""
ALTER TABLE users
ADD COLUMN IF NOT EXISTS profile_visibility VARCHAR(20) NOT NULL DEFAULT 'friends'
CHECK (profile_visibility IN ('public', 'friends', 'nobody'))
""",
"""
ALTER TABLE users ALTER COLUMN profile_visibility SET DEFAULT 'friends'
""",
"""
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint
WHERE conname = 'users_profile_visibility_check'
AND pg_get_constraintdef(oid) LIKE '%nobody%'
) THEN
ALTER TABLE users DROP CONSTRAINT IF EXISTS users_profile_visibility_check;
ALTER TABLE users ADD CONSTRAINT users_profile_visibility_check
CHECK (profile_visibility IN ('public', 'friends', 'nobody'));
END IF;
END $$
""",
"""
ALTER TABLE concerts ADD COLUMN IF NOT EXISTS flyer_url TEXT
""",
"""
ALTER TABLE concerts ADD COLUMN IF NOT EXISTS visibility VARCHAR(20) NOT NULL DEFAULT 'public'
CHECK (visibility IN ('public', 'friends', 'private'))
""",
"""
CREATE TABLE IF NOT EXISTS event_invitations (
concert_id INTEGER NOT NULL REFERENCES concerts(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
invited_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
viewed_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (concert_id, user_id)
)
""",
"""
CREATE INDEX IF NOT EXISTS idx_event_invitations_user
ON event_invitations (user_id, viewed_at)
""",
"""
CREATE TABLE IF NOT EXISTS friendships (
id SERIAL PRIMARY KEY,
requester_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
addressee_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
status VARCHAR(20) NOT NULL DEFAULT 'pending'
CHECK (status IN ('pending', 'accepted')),
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
CHECK (requester_id <> addressee_id),
UNIQUE (requester_id, addressee_id)
)
""",
"""
CREATE UNIQUE INDEX IF NOT EXISTS idx_friendships_pair
ON friendships (LEAST(requester_id, addressee_id), GREATEST(requester_id, addressee_id))
""",
"""
CREATE TABLE IF NOT EXISTS user_blocks (
blocker_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
blocked_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (blocker_id, blocked_id),
CHECK (blocker_id <> blocked_id)
)
""",
"""
CREATE INDEX IF NOT EXISTS idx_user_blocks_blocked
ON user_blocks (blocked_id, blocker_id)
""",
"""
CREATE TABLE IF NOT EXISTS direct_messages (
id SERIAL PRIMARY KEY,
sender_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
recipient_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
body TEXT NOT NULL,
read_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
CHECK (sender_id <> recipient_id)
)
""",
"""
CREATE INDEX IF NOT EXISTS idx_direct_messages_conversation
ON direct_messages (sender_id, recipient_id, created_at)
""",
"""
CREATE INDEX IF NOT EXISTS idx_direct_messages_unread
ON direct_messages (recipient_id, read_at)
""",
"""
CREATE TABLE IF NOT EXISTS followed_bands (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
band_key VARCHAR(255) NOT NULL,
display_name VARCHAR(255) NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, band_key)
)
""",
"""
CREATE TABLE IF NOT EXISTS followed_venues (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
venue_id INTEGER NOT NULL REFERENCES venues(id) ON DELETE CASCADE,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, venue_id)
)
""",
"""
CREATE TABLE IF NOT EXISTS concert_diary (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
concert_id INTEGER NOT NULL REFERENCES concerts(id) ON DELETE CASCADE,
rating SMALLINT NOT NULL CHECK (rating BETWEEN 1 AND 5),
favorite_song VARCHAR(255),
notes TEXT,
photo_path TEXT,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, concert_id)
)
""",
"""
ALTER TABLE concert_diary ADD COLUMN IF NOT EXISTS photo_path TEXT
""",
"""
CREATE TABLE IF NOT EXISTS concert_bands (
concert_id INTEGER NOT NULL REFERENCES concerts(id) ON DELETE CASCADE,
band_key VARCHAR(255) NOT NULL,
display_name VARCHAR(255) NOT NULL,
position SMALLINT NOT NULL DEFAULT 0,
PRIMARY KEY (concert_id, band_key)
)
""",
]
with get_db_connection() as connection:
with connection.cursor() as cursor:
for statement in statements:
cursor.execute(statement)
if INITIAL_ADMIN_USERNAME and INITIAL_ADMIN_PASSWORD:
cursor.execute(
"SELECT EXISTS (SELECT 1 FROM users)",
)
users_exist = cursor.fetchone()[0]
if not users_exist:
cursor.execute(
"""
INSERT INTO users (
username,
email,
password_hash,
display_name,
is_admin
)
VALUES (%s, %s, %s, %s, TRUE)
""",
(
INITIAL_ADMIN_USERNAME,
INITIAL_ADMIN_EMAIL
or f"{INITIAL_ADMIN_USERNAME}@local.invalid",
bcrypt.hashpw(
INITIAL_ADMIN_PASSWORD.encode("utf-8"),
bcrypt.gensalt(),
).decode("utf-8"),
INITIAL_ADMIN_USERNAME,
),
)
connection.commit()
@asynccontextmanager
async def lifespan(_app: FastAPI):
ensure_schema()
yield
app = FastAPI(title="MetalCircle", lifespan=lifespan)
@app.middleware("http")
async def security_controls(request: Request, call_next):
if COOKIE_SECURE and request.url.path not in {"/impressum", "/datenschutz"}:
forwarded_proto = request.headers.get("x-forwarded-proto", request.url.scheme).split(",", 1)[0].strip()
if forwarded_proto != "https":
target = str(request.url).replace("http://", "https://", 1)
return RedirectResponse(target, status_code=308)
if request.method in {"POST", "PUT", "PATCH", "DELETE"}:
origin = request.headers.get("origin")
fetch_site = request.headers.get("sec-fetch-site")
origin_host = urlparse(origin).netloc if origin else None
expected_host = request.headers.get("host", request.url.netloc)
referer = request.headers.get("referer")
referer_host = urlparse(referer).netloc if referer else None
if fetch_site == "cross-site" or (origin_host and origin_host != expected_host) or (referer_host and referer_host != expected_host):
return HTMLResponse("Anfrage aus fremder Quelle abgelehnt.", status_code=403)
if request.url.path not in {"/login", "/register"} and request.url.path.startswith("/password-reset") is False and not origin and not referer:
return HTMLResponse("CSRF-Prüfung fehlgeschlagen.", status_code=403)
path = request.url.path
if path == "/login":
bucket_name, limit, window = "login", 10, 15 * 60
elif path == "/register" or path.startswith("/password-reset"):
bucket_name, limit, window = "account", 10, 60 * 60
elif path.startswith("/messages/"):
bucket_name, limit, window = "messages", 30, 60
elif any(part in path for part in ("/photos", "/patches")) or path in {"/profile", "/concerts"}:
bucket_name, limit, window = "uploads", 20, 60 * 60
else:
bucket_name, limit, window = "writes", 120, 60
client_host = request.client.host if request.client else "unknown"
key = (client_host, bucket_name)
now = time.monotonic()
with rate_limit_lock:
bucket = rate_limit_buckets[key]
while bucket and bucket[0] <= now - window:
bucket.popleft()
if len(bucket) >= limit:
return HTMLResponse("Zu viele Anfragen. Bitte später erneut versuchen.", status_code=429,
headers={"Retry-After": str(window)})
bucket.append(now)
response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
if COOKIE_SECURE:
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
return response
@app.middleware("http")
async def require_login(request: Request, call_next):
if (
request.url.path == "/login"
or request.url.path == "/register"
or request.url.path.startswith("/register/")
or request.url.path.startswith("/password-reset")
or request.url.path in {"/impressum", "/datenschutz"}
or request.url.path == "/profile/export"
or request.url.path.startswith("/static/")
):
return await call_next(request)
if get_current_user(request):
return await call_next(request)
return login_redirect("/")
# ============================================================
# Templates
# ============================================================
templates = Environment(
loader=FileSystemLoader(
os.path.join(BASE_DIR, "templates")
),
autoescape=select_autoescape(["html"])
)
# ============================================================
# Static files
# ============================================================
app.mount(
"/static",
StaticFiles(
directory=os.path.join(BASE_DIR, "static")
),
name="static"
)
# ============================================================
# Templates
# ============================================================
def hash_token(token: str) -> str:
return hashlib.sha256(
token.encode("utf-8")
).hexdigest()
def generate_token() -> str:
return secrets.token_urlsafe(32)
def create_account_token(user_id: int, purpose: str, hours: int = 24) -> str:
token = generate_token()
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"DELETE FROM account_tokens WHERE user_id = %s AND purpose = %s AND used_at IS NULL",
(user_id, purpose),
)
cursor.execute(
"""
INSERT INTO account_tokens (user_id, purpose, token_hash, expires_at)
VALUES (%s, %s, %s, %s)
""",
(user_id, purpose, hash_token(token), datetime.now() + timedelta(hours=hours)),
)
connection.commit()
return token
def user_from_row(row):
if not row:
return None
return {
"id": row[0],
"username": row[1],
"display_name": row[2] or row[1],
"is_admin": bool(row[3]),
"pending_friend_count": row[4],
"unread_message_count": row[5],
"event_invitation_count": row[6],
"notification_count": row[4] + row[5] + row[6],
}
def get_current_user(request: Request):
token = request.cookies.get(SESSION_COOKIE)
if not token:
return None
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT
users.id,
users.username,
users.display_name,
users.is_admin,
(SELECT COUNT(*) FROM friendships
WHERE addressee_id = users.id AND status = 'pending'),
(SELECT COUNT(*) FROM direct_messages
WHERE recipient_id = users.id AND read_at IS NULL),
(SELECT COUNT(*) FROM event_invitations
WHERE user_id = users.id AND viewed_at IS NULL)
FROM sessions
JOIN users
ON users.id = sessions.user_id
WHERE sessions.token_hash = %s
AND sessions.expires_at > CURRENT_TIMESTAMP
""",
(hash_token(token),),
)
row = cursor.fetchone()
return user_from_row(row)
def create_session(user_id: int) -> str:
token = generate_token()
expires_at = datetime.now() + timedelta(days=SESSION_DAYS)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
INSERT INTO sessions (
user_id,
token_hash,
expires_at
)
VALUES (%s, %s, %s)
""",
(user_id, hash_token(token), expires_at),
)
connection.commit()
return token
def attach_session(response: RedirectResponse, token: str) -> RedirectResponse:
response.set_cookie(
SESSION_COOKIE,
token,
max_age=SESSION_DAYS * 24 * 60 * 60,
httponly=True,
samesite="lax",
secure=COOKIE_SECURE,
path="/",
)
return response
def login_redirect(next_path: str = "/"):
return RedirectResponse(
f"/login?next={next_path}",
status_code=303,
)
def concert_is_past(start_datetime, end_datetime) -> bool:
end = end_datetime or start_datetime
return end < datetime.now()
EVENT_TYPES = {
"concert": "Konzert",
"festival": "Festival",
"other": "Sonstiges",
}
def get_linkable_events(exclude_id: int | None = None):
with get_db_connection() as connection:
with connection.cursor() as cursor:
exclude_clause = "AND id <> %s" if exclude_id is not None else ""
parameters = (exclude_id,) if exclude_id is not None else ()
cursor.execute(
f"""
SELECT id, artist, start_datetime, event_type
FROM concerts
WHERE event_type IN ('concert', 'festival')
AND DATE(
CASE
WHEN event_type = 'festival'
THEN COALESCE(end_datetime, start_datetime)
ELSE start_datetime
END
) >= CURRENT_DATE
{exclude_clause}
ORDER BY start_datetime DESC
""",
parameters,
)
rows = cursor.fetchall()
return [
{
"id": row[0],
"title": row[1],
"date": row[2].strftime("%d.%m.%Y"),
"event_type": row[3],
"event_type_label": EVENT_TYPES[row[3]],
}
for row in rows
]
def resolve_event_relationship(cursor, event_type: str, parent_event_id: str, event_id=None):
if event_type not in EVENT_TYPES:
raise ValueError("Ungültige Veranstaltungskategorie.")
if event_type != "other" or not parent_event_id:
return None
try:
parent_id = int(parent_event_id)
except ValueError as error:
raise ValueError("Ungültige Hauptveranstaltung.") from error
if event_id is not None and parent_id == event_id:
raise ValueError("Eine Veranstaltung kann nicht mit sich selbst verknüpft werden.")
cursor.execute(
"""
SELECT
event_type,
DATE(
CASE
WHEN event_type = 'festival'
THEN COALESCE(end_datetime, start_datetime)
ELSE start_datetime
END
) >= CURRENT_DATE AS is_linkable
FROM concerts
WHERE id = %s
""",
(parent_id,),
)
parent = cursor.fetchone()
if not parent or parent[0] not in {"concert", "festival"}:
raise ValueError("Die Hauptveranstaltung muss ein Konzert oder Festival sein.")
if not parent[1]:
raise ValueError("Die Hauptveranstaltung ist bereits beendet und kann nicht mehr verknüpft werden.")
return parent_id
def load_concert(concert_id: int):
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT
concerts.id,
concerts.artist,
concerts.start_datetime,
concerts.end_datetime,
concerts.description,
concerts.ticket_url,
concerts.ticket_price,
concerts.flyer_path,
concerts.created_by,
venues.id,
venues.name,
venues.street,
venues.postal_code,
venues.city,
venues.country,
venues.latitude,
venues.longitude,
concerts.event_type,
concerts.parent_event_id,
parent_event.artist,
parent_event.event_type,
concerts.flyer_url,
concerts.visibility
FROM concerts
LEFT JOIN venues
ON concerts.venue_id = venues.id
LEFT JOIN concerts AS parent_event
ON concerts.parent_event_id = parent_event.id
WHERE concerts.id = %s
""",
(concert_id,),
)
row = cursor.fetchone()
if not row:
return None
start_datetime = row[2]
end_datetime = row[3]
is_past = concert_is_past(start_datetime, end_datetime)
return {
"id": row[0],
"artist": row[1],
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"description": row[4],
"ticket_url": row[5],
"ticket_price": row[6],
"flyer_path": row[7],
"flyer_url": row[21],
"visibility": row[22],
"created_by": row[8],
"is_past": is_past,
"date": start_datetime.strftime("%d.%m.%Y"),
"time": start_datetime.strftime("%H:%M"),
"start_local": start_datetime.strftime("%Y-%m-%dT%H:%M"),
"end_date": end_datetime.strftime("%d.%m.%Y") if end_datetime else None,
"end_time": end_datetime.strftime("%H:%M") if end_datetime else None,
"end_local": end_datetime.strftime("%Y-%m-%dT%H:%M") if end_datetime else "",
"event_type": row[17],
"event_type_label": EVENT_TYPES[row[17]],
"parent_event": {
"id": row[18],
"title": row[19],
"event_type": row[20],
"event_type_label": EVENT_TYPES.get(row[20]),
} if row[18] else None,
"venue": {
"id": row[9],
"name": row[10] or "Veranstaltungsort unbekannt",
"street": row[11],
"postal_code": row[12],
"city": row[13],
"country": row[14],
"latitude": row[15],
"longitude": row[16],
},
}
def can_edit_title(user, concert) -> bool:
if not user:
return False
if concert["is_past"]:
return user["is_admin"]
return user["is_admin"] or user["id"] == concert["created_by"]
def can_edit_details(user, concert) -> bool:
if not user:
return False
if concert["is_past"]:
return user["is_admin"]
return True
def can_edit_concert(user, concert) -> bool:
return can_edit_title(user, concert) or can_edit_details(user, concert)
def can_delete_concert(user, concert) -> bool:
if not user:
return False
if concert["is_past"]:
return user["is_admin"]
return user["is_admin"] or user["id"] == concert["created_by"]
def can_manage_event_access(user, concert) -> bool:
return bool(user and (user["is_admin"] or user["id"] == concert["created_by"]))
def serialize_concert_card(row):
concert_id, artist, start_datetime, end_datetime, venue, city, country, event_type, parent_event_id, visibility, is_invited = row
venue_text = venue or "Veranstaltungsort unbekannt"
if city:
venue_text += f", {city}"
return {
"id": concert_id,
"artist": artist,
"date": start_datetime.strftime("%d.%m.%Y"),
"time": start_datetime.strftime("%H:%M"),
"end_date": end_datetime.strftime("%d.%m.%Y") if end_datetime else None,
"venue": venue_text,
"city": city or "",
"country": country or "",
"event_type": event_type,
"event_type_label": EVENT_TYPES[event_type],
"parent_event_id": parent_event_id,
"visibility": visibility,
"is_invited": bool(is_invited),
"children": [],
"_start_datetime": start_datetime,
"_end_datetime": end_datetime,
}
def build_event_overview(rows):
cards = {row[0]: serialize_concert_card(row) for row in rows}
roots = []
for card in cards.values():
parent = cards.get(card["parent_event_id"])
if card["event_type"] == "other" and parent:
parent["children"].append(card)
else:
roots.append(card)
for card in roots:
card["children"].sort(key=lambda child: child["_start_datetime"])
upcoming = [card for card in roots if not concert_is_past(
card["_start_datetime"], card["_end_datetime"]
)]
past = [card for card in roots if card not in upcoming]
upcoming.sort(key=lambda card: card["_start_datetime"])
past.sort(key=lambda card: card["_start_datetime"], reverse=True)
return upcoming, past
def load_event_overview(user, search_query: str = "", date_from: str = "", date_to: str = "", venue_filter: str = "", country_filter: str = "", category_filter: str = ""):
query = """
SELECT concerts.id, concerts.artist, concerts.start_datetime,
concerts.end_datetime, venues.name, venues.city, venues.country,
concerts.event_type, concerts.parent_event_id,
concerts.visibility,
EXISTS (SELECT 1 FROM event_invitations ei
WHERE ei.concert_id = concerts.id AND ei.user_id = %s)
FROM concerts
LEFT JOIN venues ON concerts.venue_id = venues.id
WHERE %s
OR concerts.visibility = 'public'
OR concerts.created_by = %s
OR (
concerts.visibility = 'friends' AND EXISTS (
SELECT 1 FROM friendships f
WHERE f.status = 'accepted'
AND ((f.requester_id = concerts.created_by AND f.addressee_id = %s)
OR (f.addressee_id = concerts.created_by AND f.requester_id = %s))
)
)
OR (
concerts.visibility = 'private' AND EXISTS (
SELECT 1 FROM event_invitations ei
WHERE ei.concert_id = concerts.id AND ei.user_id = %s
)
)
ORDER BY concerts.start_datetime ASC
"""
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(query, (user["id"], user["is_admin"], user["id"], user["id"], user["id"], user["id"]))
upcoming, past = build_event_overview(cursor.fetchall())
term = search_query.strip().casefold()
if not any((term, date_from.strip(), date_to.strip(), venue_filter.strip(), country_filter.strip(), category_filter.strip())):
return upcoming, past
def matches(card):
text = f"{card['artist']} {card['venue']} {card['event_type_label']} {card['country']}".casefold()
own = (not term or term in text)
own = own and (not venue_filter or venue_filter.casefold() in card['venue'].casefold())
own = own and (not country_filter or country_filter.casefold() in card['country'].casefold())
own = own and (not category_filter or card['event_type'] == category_filter)
event_date = card['_start_datetime'].date().isoformat()
if date_from and date_to:
own = own and date_from <= event_date <= date_to
elif date_from:
own = own and event_date == date_from
elif date_to:
own = own and event_date == date_to
return own or any(matches(child) for child in card["children"])
return (
[card for card in upcoming if matches(card)],
[card for card in past if matches(card)],
)
def can_view_event(user, concert) -> bool:
if user["is_admin"] or concert["visibility"] == "public" or concert["created_by"] == user["id"]:
return True
with get_db_connection() as connection:
with connection.cursor() as cursor:
if concert["visibility"] == "private":
cursor.execute(
"SELECT 1 FROM event_invitations WHERE concert_id = %s AND user_id = %s",
(concert["id"], user["id"]),
)
else:
cursor.execute(
"""
SELECT 1 FROM friendships WHERE status = 'accepted'
AND ((requester_id = %s AND addressee_id = %s)
OR (addressee_id = %s AND requester_id = %s))
""",
(concert["created_by"], user["id"], concert["created_by"], user["id"]),
)
return cursor.fetchone() is not None
def get_invitable_users(exclude_user_id: int):
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT id, username, COALESCE(display_name, username)
FROM users WHERE id <> %s
ORDER BY COALESCE(display_name, username), username
""",
(exclude_user_id,),
)
return [{"id": row[0], "username": row[1], "display_name": row[2]} for row in cursor.fetchall()]
def get_event_invitee_ids(concert_id: int):
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT user_id FROM event_invitations WHERE concert_id = %s", (concert_id,))
return {row[0] for row in cursor.fetchall()}
def search_users(search_query: str):
term = search_query.strip()
if not term:
return []
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT username, COALESCE(display_name, username), avatar_path
FROM users
WHERE username ILIKE %s OR COALESCE(display_name, '') ILIKE %s
ORDER BY COALESCE(display_name, username), username
LIMIT 30
""",
(f"%{term}%", f"%{term}%"),
)
return [
{"username": row[0], "display_name": row[1], "avatar_path": row[2]}
for row in cursor.fetchall()
]
def normalize_instagram_url(value: str):
value = value.strip()
if not value:
return None
if value.startswith("@"):
value = value[1:]
if re.fullmatch(r"[A-Za-z0-9._]{1,30}", value):
return f"https://www.instagram.com/{value}/"
candidate = value if "://" in value else f"https://{value}"
parsed = urlparse(candidate)
if (parsed.hostname or "").lower() not in {"instagram.com", "www.instagram.com"}:
raise ValueError("Bitte einen gültigen Instagram-Profillink eingeben.")
username = parsed.path.strip("/").split("/", 1)[0]
if not re.fullmatch(r"[A-Za-z0-9._]{1,30}", username):
raise ValueError("Bitte einen gültigen Instagram-Profillink eingeben.")
return f"https://www.instagram.com/{username}/"
def normalize_external_url(value: str, field_name: str):
value = value.strip()
if not value:
return None
parsed = urlparse(value)
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
raise ValueError(f"Bitte für {field_name} eine vollständige HTTP- oder HTTPS-Adresse eingeben.")
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 inferred_band_names(title: str, event_type: str = "concert") -> list[str]:
if event_type != "concert":
return []
primary = re.split(r"\s+(?:-|–|—)\s+|:\s+", title.strip(), maxsplit=1)[0]
return [part.strip() for part in re.split(r"\s+(?:\+|/|&|and|und)\s+", primary) if part.strip()]
def parse_band_names(value: str, fallback_title: str = "", event_type: str = "concert") -> list[dict]:
names = [line.strip() for line in (value or "").splitlines() if line.strip()]
if not names:
names = inferred_band_names(fallback_title, event_type)
bands = []
seen = set()
for name in names[:30]:
key = normalize_artist_name(name)[:255]
if key and key not in seen:
seen.add(key)
bands.append({"key": key, "name": name[:255]})
return bands
def load_concert_bands(concert_id: int, title: str = "", event_type: str = "concert") -> list[dict]:
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"SELECT band_key, display_name FROM concert_bands WHERE concert_id = %s ORDER BY position, display_name",
(concert_id,),
)
bands = [{"key": row[0], "name": row[1]} for row in cursor.fetchall()]
return bands or parse_band_names("", title, event_type)
def replace_concert_bands(cursor, concert_id: int, bands: list[dict]):
cursor.execute("DELETE FROM concert_bands WHERE concert_id = %s", (concert_id,))
for position, band in enumerate(bands):
cursor.execute(
"INSERT INTO concert_bands (concert_id, band_key, display_name, position) VALUES (%s, %s, %s, %s)",
(concert_id, band["key"], band["name"], position),
)
def find_duplicate_concerts(user, artist: str, start_date: str, band_names: 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()
row_ids = [row[0] for row in rows]
cursor.execute(
"SELECT concert_id, band_key, display_name FROM concert_bands WHERE concert_id = ANY(%s) ORDER BY position",
(row_ids or [0],),
)
stored_bands = {}
for concert_id, band_key, display_name in cursor.fetchall():
stored_bands.setdefault(concert_id, []).append({"key": band_key, "name": display_name})
submitted_bands = parse_band_names(band_names, artist, "concert")
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 any(
artist_names_similar(submitted["name"], existing["name"])
for submitted in submitted_bands
for existing in (stored_bands.get(row[0]) or parse_band_names("", row[1], row[3]))
) or 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
extension = os.path.splitext(upload.filename)[1].lower()
if extension not in ALLOWED_IMAGE_EXTENSIONS:
return None, HTMLResponse(
"Ungültiges Bildformat. Erlaubt sind JPG, JPEG, PNG und WEBP.",
status_code=400,
)
contents = upload.file.read()
if len(contents) > MAX_IMAGE_BYTES:
return None, HTMLResponse(
"Die Datei darf maximal 10 MB groß sein.",
status_code=400,
)
if Image is None:
return None, HTMLResponse(
"Die sichere Bildprüfung ist noch nicht installiert. Bitte den Web-Container neu bauen.",
status_code=503,
)
try:
Image.MAX_IMAGE_PIXELS = MAX_IMAGE_PIXELS
with Image.open(BytesIO(contents)) as candidate:
candidate.verify()
with Image.open(BytesIO(contents)) as candidate:
image = ImageOps.exif_transpose(candidate)
if image.width * image.height > MAX_IMAGE_PIXELS:
raise ValueError("Bildauflösung zu groß")
image.thumbnail((2400, 2400))
has_alpha = image.mode in {"RGBA", "LA"} or (
image.mode == "P" and "transparency" in image.info
)
if image.mode not in {"RGB", "RGBA"}:
image = image.convert("RGBA" if has_alpha else "RGB")
output = BytesIO()
# Nicht jedes Browser-/Proxy-Setup verarbeitet serverseitig
# erzeugte WebP-Dateien zuverlässig. Nach der Validierung bleiben
# transparente Bilder PNG, normale Bilder werden als JPEG gespeichert.
output_format = "PNG" if image.mode == "RGBA" else "JPEG"
if output_format == "PNG":
image.save(output, format=output_format, optimize=True)
else:
image.save(output, format=output_format, quality=88, optimize=True)
safe_contents = output.getvalue()
except (ValueError, OSError, UnidentifiedImageError, Image.DecompressionBombError):
return None, HTMLResponse(
"Die Datei ist kein gültiges oder unterstütztes Bild.", status_code=400
)
filename = str(uuid.uuid4()) + (".png" if output_format == "PNG" else ".jpg")
destination = os.path.join(destination_dir, filename)
with open(destination, "wb") as file:
file.write(safe_contents)
return f"{url_prefix}{filename}", None
def remove_uploaded_file(path: str | None, destination_dir: str, url_prefix: str):
if not path or not path.startswith(url_prefix):
return False
relative_name = path.removeprefix(url_prefix)
if not relative_name or os.path.basename(relative_name) != relative_name:
return False
file_path = os.path.join(destination_dir, relative_name)
if not os.path.isfile(file_path):
return False
os.remove(file_path)
return True
def normalize_country(value: str) -> str:
country = (value or "").strip().casefold()
aliases = {"de": "deutschland", "deu": "deutschland", "germany": "deutschland"}
return aliases.get(country, country)
def attended_concert_stats(user_id: int) -> dict:
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT concerts.id, concerts.venue_id, concerts.event_type,
concerts.start_datetime, COALESCE(venues.country, '')
FROM concert_attendance
JOIN concerts ON concerts.id = concert_attendance.concert_id
LEFT JOIN venues ON venues.id = concerts.venue_id
WHERE concert_attendance.user_id = %s
AND concert_attendance.status = 'attending'
AND COALESCE(concerts.end_datetime, concerts.start_datetime) < CURRENT_TIMESTAMP
ORDER BY concerts.start_datetime, concerts.id
""",
(user_id,),
)
rows = cursor.fetchall()
venue_counts = {}
countries = set()
dates = []
weekend_counts = {}
badge_triggers = {}
seen_countries = set()
venue_event_ids = {}
weekend_event_ids = {}
previous_date = None
for concert_id, venue_id, event_type, concert_datetime, country_value in rows:
concert_date = concert_datetime.date()
if venue_id is not None and event_type != "other":
venue_counts[venue_id] = venue_counts.get(venue_id, 0) + 1
venue_event_ids.setdefault(venue_id, []).append(concert_id)
if venue_counts[venue_id] == 5:
badge_triggers.setdefault("regular", concert_id)
country = normalize_country(country_value)
if country:
countries.add(country)
if country not in seen_countries:
seen_countries.add(country)
if country != "deutschland":
badge_triggers.setdefault("border_breaker", concert_id)
if len(seen_countries) == 5:
badge_triggers.setdefault("globe_banger", concert_id)
dates.append(concert_date)
if previous_date and concert_date - previous_date == timedelta(days=1):
badge_triggers.setdefault("double_trouble", concert_id)
previous_date = concert_date
if concert_date.isoweekday() >= 5:
weekend_key = concert_date.isocalendar()[:2]
weekend_counts[weekend_key] = weekend_counts.get(weekend_key, 0) + 1
weekend_event_ids.setdefault(weekend_key, []).append(concert_id)
if weekend_counts[weekend_key] == 3:
badge_triggers.setdefault("iron_weekend", concert_id)
attendance_number = len(dates)
for code, _name, _icon, threshold, _description, category in BADGE_DEFINITIONS:
if category == "attendance" and threshold == attendance_number:
badge_triggers[code] = concert_id
distinct_dates = sorted(set(dates))
return {
"total": len(rows),
"max_same_venue": max(venue_counts.values(), default=0),
"country_count": len(countries),
"has_foreign_concert": any(country != "deutschland" for country in countries),
"has_consecutive_days": any(
later - earlier == timedelta(days=1)
for earlier, later in zip(distinct_dates, distinct_dates[1:])
),
"has_iron_weekend": max(weekend_counts.values(), default=0) >= 3,
"badge_triggers": badge_triggers,
}
def grant_earned_badges(user_id: int, stats: dict, registered_at):
highest_attendance_badge = None
venue_badge = "regular" if stats["max_same_venue"] >= 5 else None
for badge_code, _name, _icon, threshold, _description, category in BADGE_DEFINITIONS:
qualifies = category == "attendance" and threshold is not None and stats["total"] >= threshold
if qualifies:
highest_attendance_badge = badge_code
with get_db_connection() as connection:
with connection.cursor() as cursor:
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,),
)
for badge_code, _name, _icon, threshold, _description, category in BADGE_DEFINITIONS:
if category == "attendance" and threshold is not None and stats["total"] >= threshold:
cursor.execute(
"""
INSERT INTO user_badges (user_id, badge_code, trigger_concert_id)
VALUES (%s, %s, %s) ON CONFLICT (user_id, badge_code) DO UPDATE SET
trigger_concert_id = COALESCE(user_badges.trigger_concert_id, EXCLUDED.trigger_concert_id)
""",
(user_id, badge_code, stats["badge_triggers"].get(badge_code)),
)
if highest_attendance_badge:
cursor.execute(
"UPDATE user_badges SET trigger_concert_id = COALESCE(trigger_concert_id, %s) WHERE user_id = %s AND badge_code = %s",
(stats["badge_triggers"].get(highest_attendance_badge), user_id, highest_attendance_badge),
)
if venue_badge:
cursor.execute(
"INSERT INTO user_badges (user_id, badge_code, trigger_concert_id) VALUES (%s, %s, %s) ON CONFLICT (user_id, badge_code) DO UPDATE SET trigger_concert_id = COALESCE(user_badges.trigger_concert_id, EXCLUDED.trigger_concert_id)",
(user_id, venue_badge, stats["badge_triggers"].get(venue_badge)),
)
achievement_codes = []
if stats["has_foreign_concert"]:
achievement_codes.append("border_breaker")
if stats["country_count"] >= 5:
achievement_codes.append("globe_banger")
if stats["has_consecutive_days"]:
achievement_codes.append("double_trouble")
if stats["has_iron_weekend"]:
achievement_codes.append("iron_weekend")
for badge_code in achievement_codes:
cursor.execute(
"INSERT INTO user_badges (user_id, badge_code, trigger_concert_id) VALUES (%s, %s, %s) ON CONFLICT (user_id, badge_code) DO UPDATE SET trigger_concert_id = COALESCE(user_badges.trigger_concert_id, EXCLUDED.trigger_concert_id)",
(user_id, badge_code, stats["badge_triggers"].get(badge_code)),
)
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:
cursor.execute(
"""
SELECT id, username, display_name, avatar_path, created_at, instagram_url,
profile_visibility, is_admin
FROM users
WHERE LOWER(username) = LOWER(%s)
""",
(username,),
)
row = cursor.fetchone()
if not row:
return None
cursor.execute(
"SELECT badge_code, awarded_at, trigger_concert_id FROM user_badges WHERE user_id = %s",
(row[0],),
)
badge_rows = cursor.fetchall()
earned_codes = {badge_row[0] for badge_row in badge_rows}
badge_awarded_at = {badge_row[0]: badge_row[1] for badge_row in badge_rows}
badge_triggers = {badge_row[0]: badge_row[2] for badge_row in badge_rows}
is_founder = row[1].casefold() == "kai"
if is_founder:
earned_codes.add("founder")
if row[7]:
earned_codes.add("admin")
if row[1].casefold() == "bianca":
earned_codes.add("captns_mate")
return {
"id": row[0],
"username": row[1],
"display_name": row[2] or row[1],
"avatar_path": row[3],
"created_at": row[4].strftime("%d.%m.%Y"),
"registered_at": row[4],
"instagram_url": row[5],
"instagram_handle": row[5].rstrip("/").rsplit("/", 1)[-1] if row[5] else None,
"profile_visibility": row[6],
"is_admin": bool(row[7]),
"is_founder": is_founder,
"earned_codes": earned_codes,
"badge_awarded_at": badge_awarded_at,
"badge_triggers": badge_triggers,
}
def resolve_venue(
cursor,
venue_id: str,
venue_name: str,
city: str,
street: str,
postal_code: str,
country: str,
latitude: str,
longitude: str,
):
selected_venue_id = None
venue_name = venue_name.strip()
city = city.strip()
street = street.strip()
postal_code = postal_code.strip()
country = country.strip()
if venue_id:
if venue_id.startswith("nominatim:"):
external_id = venue_id.split(":", 1)[1]
cursor.execute(
"""
SELECT id
FROM venues
WHERE external_id IN (%s, %s)
AND source = 'nominatim'
LIMIT 1
""",
(external_id, external_id.rsplit(":", 1)[-1]),
)
existing_venue = cursor.fetchone()
if existing_venue:
selected_venue_id = existing_venue[0]
else:
cursor.execute(
"""
INSERT INTO venues (
name,
street,
postal_code,
city,
country,
latitude,
longitude,
external_id,
source
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
RETURNING id
""",
(
venue_name,
street or None,
postal_code or None,
city or None,
country or None,
float(latitude) if latitude else None,
float(longitude) if longitude else None,
external_id,
"nominatim",
),
)
selected_venue_id = cursor.fetchone()[0]
else:
try:
selected_venue_id = int(venue_id)
except ValueError:
selected_venue_id = None
if not selected_venue_id and venue_name:
cursor.execute(
"""
SELECT id
FROM venues
WHERE LOWER(name) = LOWER(%s)
AND LOWER(COALESCE(city, '')) = LOWER(%s)
LIMIT 1
""",
(venue_name, city),
)
existing_venue = cursor.fetchone()
if existing_venue:
selected_venue_id = existing_venue[0]
else:
cursor.execute(
"""
INSERT INTO venues (
name,
street,
postal_code,
city,
country,
latitude,
longitude
)
VALUES (%s, %s, %s, %s, %s, %s, %s)
RETURNING id
""",
(
venue_name,
street or None,
postal_code or None,
city or None,
country or None,
float(latitude) if latitude else None,
float(longitude) if longitude else None,
),
)
selected_venue_id = cursor.fetchone()[0]
return selected_venue_id
def require_admin(request: Request):
user = get_current_user(request)
if not user or not user["is_admin"]:
return None
return user
def get_admin_venues():
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT
venues.id,
venues.name,
venues.street,
venues.postal_code,
venues.city,
venues.country,
venues.source,
venues.is_verified,
COALESCE(string_agg(venue_aliases.alias, ', ' ORDER BY venue_aliases.alias), ''),
COUNT(DISTINCT concerts.id)
FROM venues
LEFT JOIN venue_aliases ON venue_aliases.venue_id = venues.id
LEFT JOIN concerts ON concerts.venue_id = venues.id
GROUP BY venues.id
ORDER BY venues.is_verified ASC, venues.name ASC, venues.city ASC
"""
)
rows = cursor.fetchall()
return [
{
"id": row[0],
"name": row[1],
"street": row[2] or "",
"postal_code": row[3] or "",
"city": row[4] or "",
"country": row[5] or "",
"source": row[6] or "manuell",
"is_verified": bool(row[7]),
"aliases": row[8],
"concert_count": row[9],
}
for row in rows
]
@app.get("/admin")
def admin_page(request: Request):
if not require_admin(request):
return HTMLResponse("Nicht erlaubt
", status_code=403)
return RedirectResponse("/admin/users", status_code=303)
@app.get("/admin/users", response_class=HTMLResponse)
def admin_users_page(request: Request):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT users.id, users.username, users.email, users.display_name,
users.is_admin, users.created_at,
EXISTS (
SELECT 1 FROM user_badges
WHERE user_badges.user_id = users.id
AND user_badges.badge_code = 'founder'
)
FROM users
ORDER BY is_admin DESC, username ASC
"""
)
rows = cursor.fetchall()
users = [
{
"id": row[0],
"username": row[1],
"email": row[2],
"display_name": row[3] or row[1],
"is_admin": bool(row[4]),
"created_at": row[5].strftime("%d.%m.%Y"),
"has_founder_badge": bool(row[6]),
}
for row in rows
]
template = templates.get_template("admin_users.html")
return template.render(
user=user,
users=users,
invite_url=None,
)
@app.get("/admin/venues", response_class=HTMLResponse)
def admin_venues_page(request: Request):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
template = templates.get_template("admin_venues.html")
return template.render(user=user, venues=get_admin_venues())
@app.get("/admin/patches", response_class=HTMLResponse)
def admin_patches_page(request: Request):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
assets = load_badge_assets()
patches = [
{
"code": code,
"name": name,
"icon": icon,
"description": description,
"image_path": assets.get(code),
"category": category,
}
for code, name, icon, _threshold, description, category in BADGE_DEFINITIONS
]
patch_groups = [
{
"code": category,
"label": label,
"patches": [patch for patch in patches if patch["category"] == category],
}
for category, label in BADGE_CATEGORY_LABELS.items()
if any(patch["category"] == category for patch in patches)
]
template = templates.get_template("admin_patches.html")
return template.render(user=user, patch_groups=patch_groups)
@app.get("/admin/statistics", response_class=HTMLResponse)
def admin_statistics_page(request: Request):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
with get_db_connection() as connection:
with connection.cursor() as cursor:
queries = {
"users": "SELECT COUNT(*) FROM users",
"events": "SELECT COUNT(*) FROM concerts",
"upcoming": "SELECT COUNT(*) FROM concerts WHERE start_datetime >= NOW()",
"comments": "SELECT COUNT(*) FROM concert_comments",
"attendance": "SELECT COUNT(*) FROM concert_attendance",
"friendships": "SELECT COUNT(*) FROM friendships WHERE status = 'accepted'",
"messages": "SELECT COUNT(*) FROM direct_messages",
}
stats = {}
for key, query in queries.items():
try:
cursor.execute(query)
stats[key] = cursor.fetchone()[0]
except Exception:
connection.rollback()
stats[key] = 0
template = templates.get_template("admin_statistics.html")
return template.render(user=user, stats=stats)
@app.post("/admin/invites", response_class=HTMLResponse)
def create_invite(request: Request):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
token = secrets.token_urlsafe(32)
token_hash = hash_token(token)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("""
INSERT INTO registration_invites (
token_hash
)
VALUES (%s)
RETURNING id
""", (
token_hash,
))
invite_id = cursor.fetchone()[0]
connection.commit()
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT users.id, users.username, users.email, users.display_name,
users.is_admin, users.created_at,
EXISTS (
SELECT 1 FROM user_badges
WHERE user_badges.user_id = users.id
AND user_badges.badge_code = 'founder'
)
FROM users
ORDER BY is_admin DESC, username ASC
"""
)
rows = cursor.fetchall()
users = [
{
"id": row[0],
"username": row[1],
"email": row[2],
"display_name": row[3] or row[1],
"is_admin": bool(row[4]),
"created_at": row[5].strftime("%d.%m.%Y"),
"has_founder_badge": bool(row[6]),
}
for row in rows
]
template = templates.get_template("admin_users.html")
return template.render(
user=user,
users=users,
invite_url=f"{str(request.base_url).rstrip('/')}/register/{token}",
invite_id=invite_id,
)
@app.post("/admin/venues/{venue_id}")
def update_venue(
request: Request,
venue_id: int,
name: str = Form(...),
street: str = Form(""),
postal_code: str = Form(""),
city: str = Form(""),
country: str = Form(""),
aliases: str = Form(""),
is_verified: str = Form(""),
):
if not require_admin(request):
return HTMLResponse("Nicht erlaubt
", status_code=403)
name = name.strip()
if not name:
return HTMLResponse("Der Name darf nicht leer sein.
", status_code=400)
normalized_aliases = sorted({
alias.strip()
for alias in aliases.split(",")
if alias.strip() and alias.strip().casefold() != name.casefold()
})
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
UPDATE venues
SET name = %s, street = %s, postal_code = %s, city = %s,
country = %s, is_verified = %s
WHERE id = %s
""",
(
name,
street.strip() or None,
postal_code.strip() or None,
city.strip() or None,
country.strip() or None,
is_verified == "on",
venue_id,
),
)
cursor.execute("DELETE FROM venue_aliases WHERE venue_id = %s", (venue_id,))
cursor.executemany(
"INSERT INTO venue_aliases (venue_id, alias) VALUES (%s, %s)",
[(venue_id, alias) for alias in normalized_aliases],
)
connection.commit()
return RedirectResponse("/admin/venues", status_code=303)
@app.post("/admin/venues/{venue_id}/merge")
def merge_venue(
request: Request,
venue_id: int,
target_venue_id: int = Form(...),
):
if not require_admin(request):
return HTMLResponse("Nicht erlaubt
", status_code=403)
if venue_id == target_venue_id:
return HTMLResponse("Ein Ort kann nicht mit sich selbst zusammengeführt werden.
", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT id FROM venues WHERE id IN (%s, %s)", (venue_id, target_venue_id))
if len(cursor.fetchall()) != 2:
return HTMLResponse("Veranstaltungsort nicht gefunden.
", status_code=404)
cursor.execute(
"""
INSERT INTO venue_aliases (venue_id, alias)
SELECT %s, name FROM venues WHERE id = %s
ON CONFLICT DO NOTHING
""",
(target_venue_id, venue_id),
)
cursor.execute(
"""
INSERT INTO venue_aliases (venue_id, alias)
SELECT %s, alias FROM venue_aliases WHERE venue_id = %s
ON CONFLICT DO NOTHING
""",
(target_venue_id, venue_id),
)
cursor.execute(
"UPDATE concerts SET venue_id = %s WHERE venue_id = %s",
(target_venue_id, venue_id),
)
cursor.execute("DELETE FROM venues WHERE id = %s", (venue_id,))
connection.commit()
return RedirectResponse("/admin/venues", status_code=303)
@app.post("/admin/venues/{venue_id}/delete")
def delete_venue(request: Request, venue_id: int):
if not require_admin(request):
return HTMLResponse("Nicht erlaubt
", status_code=403)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("DELETE FROM venues WHERE id = %s", (venue_id,))
connection.commit()
return RedirectResponse("/admin/venues", status_code=303)
def remove_patch_file(path: str | None):
remove_uploaded_file(path, PATCH_DIR, "/static/uploads/patches/")
@app.post("/admin/patches/{badge_code}")
async def update_patch_image(
request: Request,
badge_code: str,
image: UploadFile | None = File(None),
):
user = require_admin(request)
valid_codes = {definition[0] for definition in BADGE_DEFINITIONS}
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
if badge_code not in valid_codes:
return HTMLResponse("Patch nicht gefunden
", status_code=404)
image_path, error = save_image(image, PATCH_DIR, "/static/uploads/patches/")
if error:
return error
if not image_path:
return HTMLResponse("Bitte ein Bild auswählen.
", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT path FROM badge_assets WHERE badge_code = %s", (badge_code,))
old_asset = cursor.fetchone()
cursor.execute(
"""
INSERT INTO badge_assets (badge_code, path, updated_by, updated_at)
VALUES (%s, %s, %s, CURRENT_TIMESTAMP)
ON CONFLICT (badge_code) DO UPDATE
SET path = EXCLUDED.path, updated_by = EXCLUDED.updated_by,
updated_at = CURRENT_TIMESTAMP
""",
(badge_code, image_path, user["id"]),
)
connection.commit()
if old_asset:
remove_patch_file(old_asset[0])
return RedirectResponse("/admin/patches", status_code=303)
@app.post("/admin/patches/{badge_code}/delete")
def delete_patch_image(request: Request, badge_code: str):
if not require_admin(request):
return HTMLResponse("Nicht erlaubt
", status_code=403)
if badge_code not in {definition[0] for definition in BADGE_DEFINITIONS}:
return HTMLResponse("Patch nicht gefunden
", status_code=404)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("DELETE FROM badge_assets WHERE badge_code = %s RETURNING path", (badge_code,))
deleted_asset = cursor.fetchone()
connection.commit()
if deleted_asset:
remove_patch_file(deleted_asset[0])
return RedirectResponse("/admin/patches", status_code=303)
@app.post("/admin/users/{user_id}")
def update_user_role(
request: Request,
user_id: int,
role: str = Form(...),
):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
if role not in {"admin", "user"}:
return HTMLResponse("Ungültige Rolle
", status_code=400)
if user_id == user["id"] and role != "admin":
return HTMLResponse(
"Die eigenen Adminrechte können nicht entfernt werden.
",
status_code=400,
)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT username FROM users WHERE id = %s", (user_id,))
target_user = cursor.fetchone()
if target_user and (target_user[0] or "").casefold() == "kai" and role != "admin":
return HTMLResponse(
"Der Gründer Kai muss Admin bleiben.
",
status_code=400,
)
cursor.execute(
"UPDATE users SET is_admin = %s WHERE id = %s",
(role == "admin", user_id),
)
connection.commit()
return RedirectResponse("/admin/users", status_code=303)
@app.post("/admin/users/{user_id}/delete")
def delete_user(request: Request, user_id: int):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
if user_id == user["id"]:
return HTMLResponse(
"Der eigene Account kann nicht gelöscht werden.
",
status_code=400,
)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT username, avatar_path FROM users WHERE id = %s", (user_id,))
target_user = cursor.fetchone()
if target_user and (target_user[0] or "").casefold() == "kai":
return HTMLResponse(
"Der Gründer Kai kann nicht gelöscht werden.
",
status_code=400,
)
cursor.execute("SELECT path FROM concert_photos WHERE user_id = %s", (user_id,))
photo_paths = [row[0] for row in cursor.fetchall()]
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
connection.commit()
if target_user:
remove_uploaded_file(target_user[1], AVATAR_DIR, "/static/uploads/avatars/")
for photo_path in photo_paths:
remove_uploaded_file(photo_path, PHOTO_DIR, "/static/uploads/photos/")
return RedirectResponse("/admin/users", status_code=303)
@app.post("/admin/users/{user_id}/account-link", response_class=HTMLResponse)
def create_user_account_link(request: Request, user_id: int, purpose: str = Form(...)):
user = require_admin(request)
if not user:
return HTMLResponse("Nicht erlaubt
", status_code=403)
if purpose != "password_reset":
return HTMLResponse("Ungültiger Linktyp
", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT username, email FROM users WHERE id = %s", (user_id,))
target = cursor.fetchone()
if not target:
return HTMLResponse("Benutzer nicht gefunden
", status_code=404)
token = create_account_token(user_id, purpose, 2)
path = f"/password-reset/{token}"
template = templates.get_template("account_link.html")
return template.render(
user=user, target_username=target[0], target_email=target[1],
purpose=purpose, account_url=f"{str(request.base_url).rstrip('/')}{path}",
)
@app.get("/password-reset/{token}", response_class=HTMLResponse)
def password_reset_page(token: str):
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT 1 FROM account_tokens WHERE token_hash = %s
AND purpose = 'password_reset' AND used_at IS NULL
AND expires_at > CURRENT_TIMESTAMP
""",
(hash_token(token),),
)
valid = cursor.fetchone()
if not valid:
return HTMLResponse("Reset-Link ungültig oder abgelaufen.
", status_code=410)
return templates.get_template("password_reset.html").render(token=token, error=None)
@app.post("/password-reset/{token}", response_class=HTMLResponse)
def password_reset(token: str, password: str = Form(...), password_repeat: str = Form(...)):
if len(password) < 10 or password != password_repeat:
return templates.get_template("password_reset.html").render(
token=token, error="Passwörter müssen übereinstimmen und mindestens 10 Zeichen lang sein."
)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT id, user_id FROM account_tokens WHERE token_hash = %s
AND purpose = 'password_reset' AND used_at IS NULL
AND expires_at > CURRENT_TIMESTAMP FOR UPDATE
""",
(hash_token(token),),
)
account_token = cursor.fetchone()
if not account_token:
return HTMLResponse("Reset-Link ungültig oder abgelaufen.
", status_code=410)
password_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
cursor.execute("UPDATE users SET password_hash = %s WHERE id = %s", (password_hash, account_token[1]))
cursor.execute("UPDATE account_tokens SET used_at = CURRENT_TIMESTAMP WHERE id = %s", (account_token[0],))
cursor.execute("DELETE FROM sessions WHERE user_id = %s", (account_token[1],))
connection.commit()
return RedirectResponse("/login?reset=1", status_code=303)
@app.post("/register")
def register_user(
token: str = Form(...),
username: str = Form(...),
display_name: str = Form(""),
email: str = Form(...),
password: str = Form(...)
):
username = username.strip()
display_name = display_name.strip()
email = email.strip().lower()
if len(username) < 3:
return HTMLResponse(
"Fehler
Der Benutzername muss mindestens 3 Zeichen lang sein.
",
status_code=400
)
if len(password) < 10:
return HTMLResponse(
"Fehler
Das Passwort muss mindestens 10 Zeichen lang sein.
",
status_code=400
)
with get_db_connection() as connection:
with connection.cursor() as cursor:
# Einladung prüfen
cursor.execute("""
SELECT
id,
expires_at,
used_at
FROM registration_invites
WHERE token_hash = %s
""", (
hash_token(token),
))
invite = cursor.fetchone()
if not invite:
return HTMLResponse(
"Ungültige Einladung
",
status_code=404
)
invite_id, expires_at, used_at = invite
if used_at:
return HTMLResponse(
"Diese Einladung wurde bereits verwendet.
",
status_code=410
)
if expires_at and datetime.now() > expires_at:
return HTMLResponse(
"Diese Einladung ist abgelaufen.
",
status_code=410
)
# Prüfen ob Username bereits existiert
cursor.execute("""
SELECT id
FROM users
WHERE LOWER(username) = LOWER(%s)
""", (
username,
))
if cursor.fetchone():
return HTMLResponse(
"Fehler
Dieser Benutzername ist bereits vergeben.
",
status_code=400
)
# Prüfen ob E-Mail bereits existiert
cursor.execute("""
SELECT id
FROM users
WHERE LOWER(email) = LOWER(%s)
""", (
email,
))
if cursor.fetchone():
return HTMLResponse(
"Fehler
Diese E-Mail-Adresse ist bereits registriert.
",
status_code=400
)
password_hash = bcrypt.hashpw(
password.encode("utf-8"),
bcrypt.gensalt()
).decode("utf-8")
cursor.execute("SELECT COUNT(*) FROM users")
is_first_user = cursor.fetchone()[0] == 0
cursor.execute("""
INSERT INTO users (
username,
email,
password_hash,
display_name,
is_admin
)
VALUES (
%s,
%s,
%s,
%s,
%s
)
RETURNING id
""", (
username,
email,
password_hash,
display_name or username,
is_first_user
))
user_id = cursor.fetchone()[0]
# Einladung verbrauchen
cursor.execute("""
UPDATE registration_invites
SET
used_by = %s,
used_at = CURRENT_TIMESTAMP
WHERE id = %s
""", (
user_id,
invite_id
))
connection.commit()
response = RedirectResponse("/", status_code=303)
return attach_session(response, create_session(user_id))
# ============================================================
# Registration
# ============================================================
@app.get(
"/register/{token}",
response_class=HTMLResponse
)
def register_page(token: str):
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("""
SELECT
id,
expires_at,
used_at
FROM registration_invites
WHERE token_hash = %s
""", (
hash_token(token),
))
invite = cursor.fetchone()
if not invite:
return HTMLResponse(
"Ungültige Einladung
",
status_code=404
)
invite_id, expires_at, used_at = invite
if used_at:
return HTMLResponse(
"Diese Einladung wurde bereits verwendet.
",
status_code=410
)
if expires_at:
from datetime import datetime
if datetime.now() > expires_at:
return HTMLResponse(
"Diese Einladung ist abgelaufen.
",
status_code=410
)
template = templates.get_template(
"register.html"
)
return template.render(
token=token
)
# ============================================================
# Rechtliche Hinweise
@app.get("/impressum", response_class=HTMLResponse)
def impressum_page():
return templates.get_template("impressum.html").render()
@app.get("/datenschutz", response_class=HTMLResponse)
def privacy_page():
return templates.get_template("datenschutz.html").render()
# Login
# ============================================================
@app.get("/login", response_class=HTMLResponse)
def login_page(request: Request, next: str = "/"):
if get_current_user(request):
return RedirectResponse(next or "/", status_code=303)
template = templates.get_template("login.html")
return template.render(next_path=next or "/", error=None)
@app.post("/login")
def login(
username: str = Form(...),
password: str = Form(...),
next: str = Form("/"),
):
username = username.strip()
next_path = next if next.startswith("/") else "/"
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT id, password_hash
FROM users
WHERE LOWER(username) = LOWER(%s)
OR LOWER(email) = LOWER(%s)
""",
(username, username),
)
row = cursor.fetchone()
if not row or not bcrypt.checkpw(
password.encode("utf-8"),
row[1].encode("utf-8"),
):
template = templates.get_template("login.html")
return HTMLResponse(
template.render(
next_path=next_path,
error="Benutzername oder Passwort ist falsch.",
),
status_code=401,
)
response = RedirectResponse(next_path, status_code=303)
return attach_session(response, create_session(row[0]))
@app.post("/logout")
def logout(request: Request):
token = request.cookies.get(SESSION_COOKIE)
if token:
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"DELETE FROM sessions WHERE token_hash = %s",
(hash_token(token),),
)
connection.commit()
response = RedirectResponse("/", status_code=303)
response.delete_cookie(SESSION_COOKIE, path="/", secure=COOKIE_SECURE, samesite="lax")
return response
# ============================================================
# Home
# ============================================================
@app.get("/", response_class=HTMLResponse)
def home(request: Request, q: str = "", date_from: str = "", date_to: str = "", venue: str = "", country: str = "", category: str = ""):
user = get_current_user(request)
upcoming_concerts, _past_concerts = load_event_overview(user, q, date_from, date_to, venue, country, category)
template = templates.get_template("index.html")
return template.render(
user=user,
upcoming_concerts=upcoming_concerts,
past_concerts=[],
archive=False,
search_query=q.strip(),
user_results=search_users(q),
filters={"date_from": date_from, "date_to": date_to, "venue": venue, "country": country, "category": category},
)
@app.get("/events/past", response_class=HTMLResponse)
def past_events(request: Request, q: str = "", date_from: str = "", date_to: str = "", venue: str = "", country: str = "", category: str = ""):
user = get_current_user(request)
_upcoming_concerts, past_concerts = load_event_overview(user, q, date_from, date_to, venue, country, category)
template = templates.get_template("index.html")
return template.render(
user=user,
upcoming_concerts=[],
past_concerts=past_concerts,
archive=True,
search_query=q.strip(),
user_results=search_users(q),
filters={"date_from": date_from, "date_to": date_to, "venue": venue, "country": country, "category": category},
)
# ============================================================
# Profiles
# ============================================================
def render_profile(
request: Request,
username: str,
force_own: bool = False,
form_error: str | None = None,
form_success: str | None = None,
instagram_input: str | None = None,
status_code: int = 200,
):
viewer = get_current_user(request)
profile = load_profile(username)
if not profile:
return HTMLResponse("Benutzer nicht gefunden
", status_code=404)
is_own_profile = bool(viewer and viewer["id"] == profile["id"])
friendship = None
block_status = {"blocked_by_viewer": False, "blocked_viewer": False}
if viewer and not is_own_profile:
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT id, requester_id, addressee_id, status
FROM friendships
WHERE (requester_id = %s AND addressee_id = %s)
OR (requester_id = %s AND addressee_id = %s)
""",
(viewer["id"], profile["id"], profile["id"], viewer["id"]),
)
row = cursor.fetchone()
if row:
friendship = {
"id": row[0], "requester_id": row[1],
"addressee_id": row[2], "status": row[3],
}
cursor.execute(
"""
SELECT blocker_id, blocked_id FROM user_blocks
WHERE (blocker_id = %s AND blocked_id = %s)
OR (blocker_id = %s AND blocked_id = %s)
""",
(viewer["id"], profile["id"], profile["id"], viewer["id"]),
)
for blocker_id, _blocked_id in cursor.fetchall():
if blocker_id == viewer["id"]:
block_status["blocked_by_viewer"] = True
else:
block_status["blocked_viewer"] = True
can_view_details = (
profile["profile_visibility"] == "public"
or is_own_profile
or bool(viewer and viewer["is_admin"])
or bool(friendship and friendship["status"] == "accepted")
)
connections = {"incoming": [], "outgoing": [], "friends": [], "blocked": []}
if is_own_profile:
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT f.id, f.requester_id, f.addressee_id, f.status,
other.username, COALESCE(other.display_name, other.username),
other.avatar_path
FROM friendships f
JOIN users other ON other.id = CASE
WHEN f.requester_id = %s THEN f.addressee_id
ELSE f.requester_id
END
WHERE f.requester_id = %s OR f.addressee_id = %s
ORDER BY f.updated_at DESC
""",
(viewer["id"], viewer["id"], viewer["id"]),
)
for row in cursor.fetchall():
item = {
"id": row[0], "requester_id": row[1], "addressee_id": row[2],
"status": row[3], "username": row[4],
"display_name": row[5], "avatar_path": row[6],
}
if row[3] == "accepted":
connections["friends"].append(item)
elif row[2] == viewer["id"]:
connections["incoming"].append(item)
else:
connections["outgoing"].append(item)
cursor.execute(
"""
SELECT u.username, COALESCE(u.display_name, u.username), u.avatar_path
FROM user_blocks b
JOIN users u ON u.id = b.blocked_id
WHERE b.blocker_id = %s
ORDER BY COALESCE(u.display_name, u.username), u.username
""",
(viewer["id"],),
)
connections["blocked"] = [
{"username": row[0], "display_name": row[1], "avatar_path": row[2]}
for row in cursor.fetchall()
]
attendance_stats = attended_concert_stats(profile["id"])
grant_earned_badges(
profile["id"],
attendance_stats,
profile["registered_at"],
)
profile = load_profile(username)
badge_assets = load_badge_assets()
highest_visible_attendance = next(
(code for code, _name, _icon, threshold, _description, category in reversed(BADGE_DEFINITIONS)
if category == "attendance" and threshold is not None and attendance_stats["total"] >= threshold),
None,
)
badges = [
{
"code": code,
"name": name,
"icon": icon,
"threshold": threshold,
"description": description,
"category": category,
"earned": code in profile["earned_codes"] and (category != "attendance" or code == highest_visible_attendance),
"image_path": badge_assets.get(code),
"awarded_at": (profile["badge_awarded_at"].get(code) or profile["registered_at"]).strftime("%d.%m.%Y"),
"trigger_concert": None,
"sort_key": (0, datetime.min) if code == "founder" else (1, datetime.min) if code == "admin" else (2, profile["badge_awarded_at"].get(code) or datetime.max),
}
for code, name, icon, threshold, description, category in BADGE_DEFINITIONS
]
for badge in badges:
trigger_id = profile["badge_triggers"].get(badge["code"])
if trigger_id:
trigger_concert = load_concert(trigger_id)
if trigger_concert and can_view_event(viewer, trigger_concert):
badge["trigger_concert"] = {
"id": trigger_id, "artist": trigger_concert["artist"],
"date": trigger_concert["start_datetime"].strftime("%d.%m.%Y"),
}
badges.sort(key=lambda badge: badge["sort_key"])
template = templates.get_template("profile.html")
return HTMLResponse(
template.render(
user=viewer,
profile=profile,
attended_count=attendance_stats["total"],
badges=badges,
is_own_profile=force_own or is_own_profile,
can_view_details=can_view_details,
friendship=friendship,
block_status=block_status,
connections=connections,
form_error=form_error,
form_success=form_success,
instagram_input=instagram_input,
),
status_code=status_code,
)
@app.get("/profile", response_class=HTMLResponse)
def own_profile(request: Request, saved: str = ""):
user = get_current_user(request)
saved_items = set(saved.split(","))
messages = []
if "profile" in saved_items:
messages.append("Profil gespeichert")
if "avatar" in saved_items:
messages.append("Profilbild aktualisiert")
if "instagram" in saved_items:
messages.append("Instagram verknüpft")
return render_profile(
request,
user["username"],
force_own=True,
form_success=" · ".join(messages) if messages else None,
)
@app.get("/profile/export")
def export_profile_data(request: Request):
"""Download the authenticated user's application data as JSON."""
user = get_current_user(request)
if not user:
return login_redirect("/profile/export")
user_id = user["id"]
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT id, username, email, display_name, avatar_path,
instagram_url, profile_visibility, is_admin, created_at
FROM users WHERE id = %s
""",
(user_id,),
)
account = cursor.fetchone()
cursor.execute(
"""
SELECT c.id, c.artist, c.event_type, c.start_datetime, c.end_datetime,
c.venue_id, c.description, c.ticket_url, c.ticket_price,
c.flyer_path, c.flyer_url, c.visibility, c.created_at
FROM concerts c WHERE c.created_by = %s ORDER BY c.start_datetime
""",
(user_id,),
)
created_events = cursor.fetchall()
cursor.execute(
"""
SELECT concert_id, status, updated_at
FROM concert_attendance WHERE user_id = %s ORDER BY updated_at
""",
(user_id,),
)
attendance = cursor.fetchall()
cursor.execute(
"""
SELECT id, requester_id, addressee_id, status, created_at, updated_at
FROM friendships WHERE requester_id = %s OR addressee_id = %s
ORDER BY created_at
""",
(user_id, user_id),
)
friendships = cursor.fetchall()
cursor.execute(
"""
SELECT blocker_id, blocked_id, created_at
FROM user_blocks WHERE blocker_id = %s OR blocked_id = %s
ORDER BY created_at
""",
(user_id, user_id),
)
blocks = cursor.fetchall()
cursor.execute(
"""
SELECT id, sender_id, recipient_id, body, read_at, created_at
FROM direct_messages WHERE sender_id = %s OR recipient_id = %s
ORDER BY created_at
""",
(user_id, user_id),
)
messages = cursor.fetchall()
cursor.execute(
"""
SELECT concert_id, invited_by, viewed_at, created_at
FROM event_invitations WHERE user_id = %s ORDER BY created_at
""",
(user_id,),
)
invitations = cursor.fetchall()
cursor.execute(
"""
SELECT id, concert_id, body, created_at
FROM concert_comments WHERE user_id = %s ORDER BY created_at
""",
(user_id,),
)
comments = cursor.fetchall()
cursor.execute(
"""
SELECT id, concert_id, path, created_at
FROM concert_photos WHERE user_id = %s ORDER BY created_at
""",
(user_id,),
)
photos = cursor.fetchall()
cursor.execute(
"SELECT band_key, display_name, created_at FROM followed_bands WHERE user_id = %s ORDER BY created_at",
(user_id,),
)
followed_bands = cursor.fetchall()
cursor.execute(
"SELECT venue_id, created_at FROM followed_venues WHERE user_id = %s ORDER BY created_at",
(user_id,),
)
followed_venues = cursor.fetchall()
cursor.execute(
"SELECT concert_id, rating, favorite_song, notes, photo_path, created_at, updated_at FROM concert_diary WHERE user_id = %s ORDER BY updated_at",
(user_id,),
)
diary_entries = cursor.fetchall()
cursor.execute(
"""
SELECT badge_code, awarded_at, trigger_concert_id FROM user_badges
WHERE user_id = %s ORDER BY awarded_at
""",
(user_id,),
)
badges = cursor.fetchall()
def rows_to_dicts(rows, keys):
return [dict(zip(keys, row)) for row in rows]
data = {
"export_version": 1,
"exported_at": datetime.now(),
"account": dict(zip(
("id", "username", "email", "display_name", "avatar_path",
"instagram_url", "profile_visibility", "is_admin", "created_at"),
account,
)) if account else None,
"created_events": rows_to_dicts(
created_events,
("id", "artist", "event_type", "start_datetime", "end_datetime", "venue_id",
"description", "ticket_url", "ticket_price", "flyer_path", "flyer_url",
"visibility", "created_at"),
),
"attendance": rows_to_dicts(attendance, ("concert_id", "status", "updated_at")),
"friendships": rows_to_dicts(friendships, ("id", "requester_id", "addressee_id", "status", "created_at", "updated_at")),
"blocks": rows_to_dicts(blocks, ("blocker_id", "blocked_id", "created_at")),
"messages": rows_to_dicts(messages, ("id", "sender_id", "recipient_id", "body", "read_at", "created_at")),
"event_invitations": rows_to_dicts(invitations, ("concert_id", "invited_by", "viewed_at", "created_at")),
"comments": rows_to_dicts(comments, ("id", "concert_id", "body", "created_at")),
"photos": rows_to_dicts(photos, ("id", "concert_id", "path", "created_at")),
"followed_bands": rows_to_dicts(followed_bands, ("band_key", "display_name", "created_at")),
"followed_venues": rows_to_dicts(followed_venues, ("venue_id", "created_at")),
"concert_diary": rows_to_dicts(diary_entries, ("concert_id", "rating", "favorite_song", "notes", "photo_path", "created_at", "updated_at")),
"badges": rows_to_dicts(badges, ("badge_code", "awarded_at", "trigger_concert_id")),
}
filename = re.sub(r"[^A-Za-z0-9_-]", "_", user["username"])
return JSONResponse(
content=jsonable_encoder(data),
headers={"Content-Disposition": f'attachment; filename="pingu-concerts-{filename}-daten.json"'},
)
@app.post("/profile/delete")
def delete_own_account(request: Request):
"""Permanently remove the authenticated user's account and personal data."""
user = get_current_user(request)
if not user:
return login_redirect("/profile")
user_id = user["id"]
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT avatar_path FROM users WHERE id = %s", (user_id,))
avatar_path = cursor.fetchone()[0]
cursor.execute("SELECT path FROM concert_photos WHERE user_id = %s", (user_id,))
photo_paths = [row[0] for row in cursor.fetchall()]
cursor.execute("SELECT flyer_path FROM concerts WHERE created_by = %s AND flyer_path IS NOT NULL", (user_id,))
flyer_paths = [row[0] for row in cursor.fetchall()]
cursor.execute("SELECT photo_path FROM concert_diary WHERE user_id = %s AND photo_path IS NOT NULL", (user_id,))
diary_photo_paths = [row[0] for row in cursor.fetchall()]
cursor.execute("UPDATE registration_invites SET used_by = NULL WHERE used_by = %s", (user_id,))
cursor.execute("UPDATE concerts SET flyer_path = NULL WHERE created_by = %s", (user_id,))
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
connection.commit()
for path, directory, prefix in [(avatar_path, AVATAR_DIR, "/static/uploads/avatars/")] + [(p, PHOTO_DIR, "/static/uploads/photos/") for p in photo_paths] + [(p, UPLOAD_DIR, "/static/uploads/flyers/") for p in flyer_paths] + [(p, DIARY_PHOTO_DIR, "/diary/photo/") for p in diary_photo_paths]:
remove_uploaded_file(path, directory, prefix)
response = RedirectResponse("/login", status_code=303)
response.delete_cookie(SESSION_COOKIE, path="/", secure=COOKIE_SECURE, samesite="lax")
return response
@app.get("/users/{username}", response_class=HTMLResponse)
def user_profile(request: Request, username: str):
return render_profile(request, username)
@app.post("/profile")
async def update_profile(
request: Request,
display_name: str = Form(""),
instagram_url: str = Form(""),
profile_visibility: str = Form("friends"),
avatar: UploadFile | None = File(None),
):
user = get_current_user(request)
try:
normalized_instagram_url = normalize_instagram_url(instagram_url)
except ValueError as error:
return render_profile(
request,
user["username"],
force_own=True,
form_error=str(error),
instagram_input=instagram_url,
status_code=400,
)
if profile_visibility not in {"public", "friends", "nobody"}:
return HTMLResponse("Ungültige Profilsichtbarkeit.", status_code=400)
avatar_path, error = save_image(
avatar,
AVATAR_DIR,
"/static/uploads/avatars/",
)
if error:
return error
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT avatar_path FROM users WHERE id = %s", (user["id"],))
previous_profile = cursor.fetchone()
cursor.execute(
"""
UPDATE users
SET
display_name = COALESCE(NULLIF(%s, ''), display_name),
avatar_path = COALESCE(%s, avatar_path),
instagram_url = %s,
profile_visibility = %s
WHERE id = %s
RETURNING avatar_path, instagram_url
""",
(display_name.strip(), avatar_path, normalized_instagram_url, profile_visibility, user["id"]),
)
saved_profile = cursor.fetchone()
connection.commit()
if not saved_profile:
remove_uploaded_file(avatar_path, AVATAR_DIR, "/static/uploads/avatars/")
return HTMLResponse("Profil konnte nicht gespeichert werden.
", status_code=500)
if avatar_path and previous_profile and previous_profile[0] != avatar_path:
remove_uploaded_file(previous_profile[0], AVATAR_DIR, "/static/uploads/avatars/")
saved_parts = ["profile"]
if avatar_path and saved_profile[0] == avatar_path:
saved_parts.append("avatar")
if normalized_instagram_url and saved_profile[1] == normalized_instagram_url:
saved_parts.append("instagram")
return RedirectResponse(
f"/profile?saved={','.join(saved_parts)}",
status_code=303,
)
@app.post("/users/{username}/friend-request")
def send_friend_request(request: Request, username: str):
user = get_current_user(request)
profile = load_profile(username)
if not profile:
return HTMLResponse("Benutzer nicht gefunden.", status_code=404)
if profile["id"] == user["id"]:
return HTMLResponse("Du kannst dir nicht selbst eine Anfrage schicken.", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT 1 FROM user_blocks
WHERE (blocker_id = %s AND blocked_id = %s)
OR (blocker_id = %s AND blocked_id = %s)
""",
(user["id"], profile["id"], profile["id"], user["id"]),
)
if cursor.fetchone():
return HTMLResponse("Freundschaftsanfrage wegen einer Blockierung nicht möglich.", status_code=403)
cursor.execute(
"""
INSERT INTO friendships (requester_id, addressee_id)
VALUES (%s, %s)
ON CONFLICT DO NOTHING
""",
(user["id"], profile["id"]),
)
connection.commit()
return RedirectResponse(f"/users/{profile['username']}", status_code=303)
@app.post("/users/{username}/block")
def block_user(request: Request, username: str):
user = get_current_user(request)
profile = load_profile(username)
if not profile:
return HTMLResponse("Benutzer nicht gefunden.", status_code=404)
if profile["id"] == user["id"]:
return HTMLResponse("Du kannst dich nicht selbst blockieren.", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
INSERT INTO user_blocks (blocker_id, blocked_id)
VALUES (%s, %s) ON CONFLICT DO NOTHING
""",
(user["id"], profile["id"]),
)
cursor.execute(
"""
DELETE FROM friendships
WHERE (requester_id = %s AND addressee_id = %s)
OR (requester_id = %s AND addressee_id = %s)
""",
(user["id"], profile["id"], profile["id"], user["id"]),
)
connection.commit()
return RedirectResponse(f"/users/{profile['username']}", status_code=303)
@app.post("/users/{username}/unblock")
def unblock_user(request: Request, username: str):
user = get_current_user(request)
profile = load_profile(username)
if not profile:
return HTMLResponse("Benutzer nicht gefunden.", status_code=404)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"DELETE FROM user_blocks WHERE blocker_id = %s AND blocked_id = %s",
(user["id"], profile["id"]),
)
connection.commit()
return RedirectResponse(f"/users/{profile['username']}", status_code=303)
@app.post("/friendships/{friendship_id}/{action}")
def manage_friendship(request: Request, friendship_id: int, action: str, return_to: str = Form("")):
user = get_current_user(request)
if action not in {"accept", "decline", "remove"}:
return HTMLResponse("Ungültige Aktion.", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"SELECT requester_id, addressee_id, status FROM friendships WHERE id = %s",
(friendship_id,),
)
friendship = cursor.fetchone()
if not friendship or user["id"] not in friendship[:2]:
return HTMLResponse("Freundschaft nicht gefunden.", status_code=404)
if action == "accept":
if friendship[1] != user["id"] or friendship[2] != "pending":
return HTMLResponse("Diese Anfrage kann nicht angenommen werden.", status_code=403)
cursor.execute(
"UPDATE friendships SET status = 'accepted', updated_at = CURRENT_TIMESTAMP WHERE id = %s",
(friendship_id,),
)
else:
if action == "decline" and (friendship[1] != user["id"] or friendship[2] != "pending"):
return HTMLResponse("Diese Anfrage kann nicht abgelehnt werden.", status_code=403)
cursor.execute("DELETE FROM friendships WHERE id = %s", (friendship_id,))
connection.commit()
other_id = friendship[1] if friendship[0] == user["id"] else friendship[0]
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT username FROM users WHERE id = %s", (other_id,))
other = cursor.fetchone()
if return_to == "/messages":
return RedirectResponse("/messages", status_code=303)
return RedirectResponse(f"/users/{other[0]}" if other else "/", status_code=303)
# ============================================================
# Direct messages
# ============================================================
def load_chat_partner(cursor, user, username: str):
cursor.execute(
"""
SELECT u.id, u.username, COALESCE(u.display_name, u.username), u.avatar_path,
u.is_admin
FROM users u
WHERE LOWER(u.username) = LOWER(%s)
AND u.id <> %s
AND NOT EXISTS (
SELECT 1 FROM user_blocks b
WHERE (b.blocker_id = %s AND b.blocked_id = u.id)
OR (b.blocker_id = u.id AND b.blocked_id = %s)
)
AND (%s OR u.is_admin OR EXISTS (
SELECT 1 FROM friendships f
WHERE f.status = 'accepted'
AND ((f.requester_id = %s AND f.addressee_id = u.id)
OR (f.addressee_id = %s AND f.requester_id = u.id))
))
""",
(username, user["id"], user["id"], user["id"], user["is_admin"], user["id"], user["id"]),
)
row = cursor.fetchone()
if not row:
return None
return {"id": row[0], "username": row[1], "display_name": row[2],
"avatar_path": row[3], "is_admin": bool(row[4])}
@app.get("/messages", response_class=HTMLResponse)
def message_inbox(request: Request):
user = get_current_user(request)
conversations = []
friend_requests = []
event_invitations = []
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT f.id, u.username, COALESCE(u.display_name, u.username), f.created_at
FROM friendships f JOIN users u ON u.id = f.requester_id
WHERE f.addressee_id = %s AND f.status = 'pending'
ORDER BY f.created_at DESC
""",
(user["id"],),
)
friend_requests = [
{"id": row[0], "username": row[1], "display_name": row[2],
"created_at": row[3].strftime("%d.%m.%Y %H:%M")}
for row in cursor.fetchall()
]
cursor.execute(
"""
SELECT ei.concert_id, c.artist, c.start_datetime,
COALESCE(u.display_name, u.username)
FROM event_invitations ei
JOIN concerts c ON c.id = ei.concert_id
LEFT JOIN users u ON u.id = ei.invited_by
WHERE ei.user_id = %s AND ei.viewed_at IS NULL
ORDER BY ei.created_at DESC
""",
(user["id"],),
)
event_invitations = [
{"concert_id": row[0], "artist": row[1],
"date": row[2].strftime("%d.%m.%Y %H:%M"), "invited_by": row[3] or "Ein Mitglied"}
for row in cursor.fetchall()
]
cursor.execute(
"""
SELECT u.id, u.username, COALESCE(u.display_name, u.username), u.avatar_path
FROM users u
WHERE u.id <> %s
AND NOT EXISTS (
SELECT 1 FROM user_blocks b
WHERE (b.blocker_id = %s AND b.blocked_id = u.id)
OR (b.blocker_id = u.id AND b.blocked_id = %s)
)
AND (%s OR u.is_admin OR EXISTS (
SELECT 1 FROM friendships f
WHERE f.status = 'accepted'
AND ((f.requester_id = %s AND f.addressee_id = u.id)
OR (f.addressee_id = %s AND f.requester_id = u.id))
))
ORDER BY COALESCE(u.display_name, u.username)
""",
(user["id"], user["id"], user["id"], user["is_admin"], user["id"], user["id"]),
)
for row in cursor.fetchall():
cursor.execute(
"""
SELECT body, created_at, sender_id,
(SELECT COUNT(*) FROM direct_messages
WHERE sender_id = %s AND recipient_id = %s AND read_at IS NULL)
FROM direct_messages
WHERE (sender_id = %s AND recipient_id = %s)
OR (sender_id = %s AND recipient_id = %s)
ORDER BY created_at DESC LIMIT 1
""",
(row[0], user["id"], user["id"], row[0], row[0], user["id"]),
)
latest = cursor.fetchone()
conversations.append({
"id": row[0], "username": row[1], "display_name": row[2],
"avatar_path": row[3], "last_message": latest[0] if latest else None,
"last_at": latest[1].strftime("%d.%m.%Y %H:%M") if latest else None,
"last_at_raw": latest[1] if latest else None,
"last_from_me": bool(latest and latest[2] == user["id"]),
"unread_count": latest[3] if latest else 0,
})
conversations.sort(key=lambda item: (item["unread_count"] > 0, item["last_at_raw"] or datetime.min), reverse=True)
template = templates.get_template("messages.html")
return template.render(user=user, conversations=conversations, friend_requests=friend_requests,
event_invitations=event_invitations, partner=None, messages=[])
@app.get("/messages/{username}", response_class=HTMLResponse)
def message_conversation(request: Request, username: str):
user = get_current_user(request)
with get_db_connection() as connection:
with connection.cursor() as cursor:
partner = load_chat_partner(cursor, user, username)
if not partner:
return HTMLResponse("Chat nicht erlaubt. Er ist für Freunde und Unterhaltungen mit Admins verfügbar.", status_code=403)
cursor.execute(
"""
UPDATE direct_messages SET read_at = CURRENT_TIMESTAMP
WHERE sender_id = %s AND recipient_id = %s AND read_at IS NULL
""",
(partner["id"], user["id"]),
)
cursor.execute(
"""
SELECT sender_id, body, created_at
FROM direct_messages
WHERE (sender_id = %s AND recipient_id = %s)
OR (sender_id = %s AND recipient_id = %s)
ORDER BY created_at ASC, id ASC
LIMIT 500
""",
(user["id"], partner["id"], partner["id"], user["id"]),
)
messages = [
{"from_me": row[0] == user["id"], "body": row[1],
"created_at": row[2].strftime("%d.%m.%Y %H:%M")}
for row in cursor.fetchall()
]
connection.commit()
template = templates.get_template("messages.html")
return template.render(user=get_current_user(request), conversations=[], partner=partner, messages=messages)
@app.post("/messages/{username}")
def send_message(request: Request, username: str, body: str = Form(...)):
user = get_current_user(request)
body = body.strip()
if not body or len(body) > 2000:
return HTMLResponse("Eine Nachricht muss zwischen 1 und 2000 Zeichen lang sein.", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
partner = load_chat_partner(cursor, user, username)
if not partner:
return HTMLResponse("Chat nicht erlaubt. Er ist für Freunde und Unterhaltungen mit Admins verfügbar.", status_code=403)
cursor.execute(
"INSERT INTO direct_messages (sender_id, recipient_id, body) VALUES (%s, %s, %s)",
(user["id"], partner["id"], body),
)
connection.commit()
return RedirectResponse(f"/messages/{partner['username']}#latest", status_code=303)
# ============================================================
# New concert
# ============================================================
@app.get(
"/concerts/new",
response_class=HTMLResponse
)
def new_concert(request: Request):
user = get_current_user(request)
if not user:
return login_redirect("/concerts/new")
template = templates.get_template("new_concert.html")
return template.render(user=user, linkable_events=get_linkable_events(),
invitable_users=get_invitable_users(user["id"]))
@app.get("/api/concerts/duplicates")
def duplicate_concerts(request: Request, artist: str = "", start_date: str = "", band_names: 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, band_names)})
@app.get("/following", response_class=HTMLResponse)
def following_page(request: Request):
user = get_current_user(request)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"SELECT band_key, display_name FROM followed_bands WHERE user_id = %s ORDER BY display_name",
(user["id"],),
)
followed_bands = [{"key": row[0], "name": row[1]} for row in cursor.fetchall()]
cursor.execute(
"""
SELECT v.id, v.name, COALESCE(v.city, '') FROM followed_venues fv
JOIN venues v ON v.id = fv.venue_id
WHERE fv.user_id = %s ORDER BY v.name, v.city
""",
(user["id"],),
)
followed_venues = [{"id": row[0], "name": row[1], "city": row[2]} for row in cursor.fetchall()]
cursor.execute(
"""
SELECT c.id, c.artist, c.start_datetime, c.venue_id,
COALESCE(v.name, ''), COALESCE(v.city, ''), c.visibility, c.created_by,
c.event_type
FROM concerts c LEFT JOIN venues v ON v.id = c.venue_id
WHERE COALESCE(c.end_datetime, c.start_datetime) >= CURRENT_TIMESTAMP
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
""",
(user["id"], user["is_admin"], user["id"], user["id"], user["id"]),
)
candidates = cursor.fetchall()
candidate_ids = [row[0] for row in candidates]
cursor.execute(
"SELECT concert_id, band_key, display_name FROM concert_bands WHERE concert_id = ANY(%s) ORDER BY position",
(candidate_ids or [0],),
)
candidate_bands = {}
for concert_id, band_key, display_name in cursor.fetchall():
candidate_bands.setdefault(concert_id, []).append({"key": band_key, "name": display_name})
band_names = [band["name"] for band in followed_bands]
venue_ids = {venue["id"] for venue in followed_venues}
events = [
{
"id": row[0], "artist": row[1], "date": row[2].strftime("%d.%m.%Y"),
"time": row[2].strftime("%H:%M"), "venue": ", ".join(filter(None, (row[4], row[5]))),
"matched_band": any(
artist_names_similar(event_band["name"], followed_band)
for event_band in (candidate_bands.get(row[0]) or parse_band_names("", row[1], row[8]))
for followed_band in band_names
),
"matched_venue": row[3] in venue_ids,
}
for row in candidates
if row[3] in venue_ids or any(
artist_names_similar(event_band["name"], followed_band)
for event_band in (candidate_bands.get(row[0]) or parse_band_names("", row[1], row[8]))
for followed_band in band_names
)
]
return templates.get_template("following.html").render(
user=user, followed_bands=followed_bands, followed_venues=followed_venues, events=events
)
@app.post("/following/bands/remove")
def remove_followed_band(request: Request, band_key: str = Form(...)):
user = get_current_user(request)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("DELETE FROM followed_bands WHERE user_id = %s AND band_key = %s", (user["id"], band_key[:255]))
connection.commit()
return RedirectResponse("/following", status_code=303)
@app.post("/following/venues/remove")
def remove_followed_venue(request: Request, venue_id: int = Form(...)):
user = get_current_user(request)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("DELETE FROM followed_venues WHERE user_id = %s AND venue_id = %s", (user["id"], venue_id))
connection.commit()
return RedirectResponse("/following", status_code=303)
@app.post("/concerts/{concert_id}/follow-band")
def follow_band(request: Request, concert_id: int, band_key: str = Form(...), action: str = Form("follow")):
user = get_current_user(request)
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse("Veranstaltung nicht gefunden.", status_code=404)
bands = load_concert_bands(concert_id, concert["artist"], concert["event_type"])
selected_band = next((band for band in bands if band["key"] == band_key), None)
if not selected_band:
return HTMLResponse("Band nicht gefunden.", status_code=404)
with get_db_connection() as connection:
with connection.cursor() as cursor:
if action == "unfollow":
cursor.execute("DELETE FROM followed_bands WHERE user_id = %s AND band_key = %s", (user["id"], band_key))
elif action == "follow":
cursor.execute(
"INSERT INTO followed_bands (user_id, band_key, display_name) VALUES (%s, %s, %s) ON CONFLICT (user_id, band_key) DO UPDATE SET display_name = EXCLUDED.display_name",
(user["id"], band_key, selected_band["name"]),
)
else:
return HTMLResponse("Ungültige Aktion.", status_code=400)
connection.commit()
return RedirectResponse(f"/concerts/{concert_id}#following", status_code=303)
@app.post("/concerts/{concert_id}/follow-venue")
def follow_venue(request: Request, concert_id: int, action: str = Form("follow")):
user = get_current_user(request)
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse("Veranstaltung nicht gefunden.", status_code=404)
venue_id = concert["venue"]["id"]
if not venue_id:
return HTMLResponse("Diese Veranstaltung hat keine zugeordnete Location.", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
if action == "unfollow":
cursor.execute("DELETE FROM followed_venues WHERE user_id = %s AND venue_id = %s", (user["id"], venue_id))
elif action == "follow":
cursor.execute("INSERT INTO followed_venues (user_id, venue_id) VALUES (%s, %s) ON CONFLICT DO NOTHING", (user["id"], venue_id))
else:
return HTMLResponse("Ungültige Aktion.", status_code=400)
connection.commit()
return RedirectResponse(f"/concerts/{concert_id}#following", status_code=303)
@app.get("/diary", response_class=HTMLResponse)
def diary_page(request: Request):
user = get_current_user(request)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT d.concert_id, c.artist, c.start_datetime, d.rating,
COALESCE(d.favorite_song, ''), COALESCE(d.notes, ''),
COALESCE(v.name, ''), COALESCE(v.city, ''), d.photo_path
FROM concert_diary d JOIN concerts c ON c.id = d.concert_id
LEFT JOIN venues v ON v.id = c.venue_id
WHERE d.user_id = %s ORDER BY c.start_datetime DESC
""",
(user["id"],),
)
entries = [
{"concert_id": row[0], "artist": row[1], "date": row[2].strftime("%d.%m.%Y"),
"rating": row[3], "favorite_song": row[4], "notes": row[5],
"venue": ", ".join(filter(None, (row[6], row[7]))), "photo_path": row[8]}
for row in cursor.fetchall()
]
cursor.execute(
"""
SELECT ub.badge_code, ub.trigger_concert_id, ub.awarded_at, ba.path
FROM user_badges ub LEFT JOIN badge_assets ba ON ba.badge_code = ub.badge_code
WHERE ub.user_id = %s AND ub.trigger_concert_id IS NOT NULL
ORDER BY ub.awarded_at
""",
(user["id"],),
)
diary_badges = {}
for badge_code, concert_id, awarded_at, image_path in cursor.fetchall():
definition = BADGE_BY_CODE.get(badge_code)
if definition:
name, icon, _threshold, description, _category = definition
diary_badges.setdefault(concert_id, []).append({
"name": name, "icon": icon, "description": description,
"image_path": image_path, "awarded_at": awarded_at.strftime("%d.%m.%Y"),
})
for entry in entries:
entry["badges"] = diary_badges.get(entry["concert_id"], [])
return templates.get_template("diary.html").render(user=user, entries=entries)
@app.get("/diary/photo/{filename}")
def diary_photo(request: Request, filename: str):
user = get_current_user(request)
if not filename or os.path.basename(filename) != filename:
return HTMLResponse("Bild nicht gefunden.", status_code=404)
photo_path = f"/diary/photo/{filename}"
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"SELECT 1 FROM concert_diary WHERE user_id = %s AND photo_path = %s",
(user["id"], photo_path),
)
if not cursor.fetchone():
return HTMLResponse("Bild nicht gefunden.", status_code=404)
file_path = os.path.join(DIARY_PHOTO_DIR, filename)
if not os.path.isfile(file_path):
return HTMLResponse("Bild nicht gefunden.", status_code=404)
return FileResponse(file_path)
@app.post("/concerts/{concert_id}/diary")
def save_diary_entry(
request: Request, concert_id: int, rating: int = Form(...),
favorite_song: str = Form(""), notes: str = Form(""),
photo: UploadFile | None = File(None), remove_photo: str = Form("")
):
user = get_current_user(request)
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse("Veranstaltung nicht gefunden.", status_code=404)
if not concert["is_past"] or rating not in range(1, 6):
return HTMLResponse("Das Tagebuch ist nur für vergangene Konzerte mit einer Bewertung von 1 bis 5 verfügbar.", status_code=400)
favorite_song = favorite_song.strip()
notes = notes.strip()
if len(favorite_song) > 255 or len(notes) > 5000:
return HTMLResponse("Tagebucheintrag ist zu lang.", status_code=400)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"SELECT 1 FROM concert_attendance WHERE concert_id = %s AND user_id = %s AND status = 'attending'",
(concert_id, user["id"]),
)
if not cursor.fetchone():
return HTMLResponse("Ein Tagebucheintrag ist nur für besuchte Konzerte möglich.", status_code=403)
cursor.execute(
"SELECT photo_path FROM concert_diary WHERE user_id = %s AND concert_id = %s",
(user["id"], concert_id),
)
previous_row = cursor.fetchone()
previous_photo = previous_row[0] if previous_row else None
new_photo, error = save_image(photo, DIARY_PHOTO_DIR, "/diary/photo/")
if error:
return error
photo_path = new_photo or (None if remove_photo else previous_photo)
cursor.execute(
"""
INSERT INTO concert_diary (user_id, concert_id, rating, favorite_song, notes, photo_path)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (user_id, concert_id) DO UPDATE SET
rating = EXCLUDED.rating, favorite_song = EXCLUDED.favorite_song,
notes = EXCLUDED.notes, photo_path = EXCLUDED.photo_path,
updated_at = CURRENT_TIMESTAMP
""",
(user["id"], concert_id, rating, favorite_song or None, notes or None, photo_path),
)
connection.commit()
if previous_photo and previous_photo != photo_path:
remove_uploaded_file(previous_photo, DIARY_PHOTO_DIR, "/diary/photo/")
return RedirectResponse(f"/concerts/{concert_id}#diary", status_code=303)
@app.post("/concerts/{concert_id}/diary/delete")
def delete_diary_entry(request: Request, concert_id: int):
user = get_current_user(request)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT photo_path FROM concert_diary WHERE user_id = %s AND concert_id = %s", (user["id"], concert_id))
row = cursor.fetchone()
cursor.execute("DELETE FROM concert_diary WHERE user_id = %s AND concert_id = %s", (user["id"], concert_id))
connection.commit()
if row and row[0]:
remove_uploaded_file(row[0], DIARY_PHOTO_DIR, "/diary/photo/")
return RedirectResponse(f"/concerts/{concert_id}#diary", status_code=303)
# ============================================================
# Concert detail
# ============================================================
def _ics_escape(value: str) -> str:
return str(value or "").replace("\\", "\\\\").replace(";", "\\;").replace(",", "\\,").replace("\n", "\\n")
@app.get("/concerts/{concert_id}.ics")
def concert_ics(request: Request, concert_id: int):
concert = load_concert(concert_id)
if not concert:
return PlainTextResponse("Veranstaltung nicht gefunden", status_code=404)
# Kalenderdaten unterliegen denselben Sichtbarkeitsregeln wie die Detailseite.
# Ohne Sitzung bzw. ohne Berechtigung keine Informationen preisgeben.
user = get_current_user(request)
if not user or not can_view_event(user, concert):
return PlainTextResponse("Veranstaltung nicht gefunden", status_code=404)
local_zone = ZoneInfo("Europe/Berlin")
start = concert["start_datetime"].replace(tzinfo=local_zone)
end_value = concert.get("end_datetime")
end = end_value.replace(tzinfo=local_zone) if end_value else start + timedelta(hours=2)
if end <= start:
end = start + timedelta(hours=2)
venue = concert.get("venue", {}) or {}
location = ", ".join(filter(None, [venue.get("name"), venue.get("city"), venue.get("country")]))
stamp = datetime.now(ZoneInfo("UTC")).strftime("%Y%m%dT%H%M%SZ")
body = "\r\n".join([
"BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//MetalCircle//Concerts//DE", "CALSCALE:GREGORIAN",
"BEGIN:VEVENT", f"UID:metalcircle-{concert_id}@konzerte.pinguholic.de", f"DTSTAMP:{stamp}",
f"DTSTART;TZID=Europe/Berlin:{start.strftime('%Y%m%dT%H%M%S')}",
f"DTEND;TZID=Europe/Berlin:{end.strftime('%Y%m%dT%H%M%S')}",
f"SUMMARY:{_ics_escape(concert['artist'])}", f"LOCATION:{_ics_escape(location)}",
f"URL:https://konzerte.pinguholic.de/concerts/{concert_id}", "END:VEVENT", "END:VCALENDAR", ""
])
return PlainTextResponse(body, media_type="text/calendar; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="metalcircle-{concert_id}.ics"'})
@app.get(
"/concerts/{concert_id}",
response_class=HTMLResponse
)
def concert_detail(request: Request, concert_id: int):
concert = load_concert(concert_id)
user = get_current_user(request)
if not concert or not can_view_event(user, concert):
return HTMLResponse(
"Veranstaltung nicht gefunden
",
status_code=404
)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"UPDATE event_invitations SET viewed_at = CURRENT_TIMESTAMP WHERE concert_id = %s AND user_id = %s AND viewed_at IS NULL",
(concert_id, user["id"]),
)
connection.commit()
user = get_current_user(request)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT
concert_comments.id,
concert_comments.body,
concert_comments.created_at,
users.display_name,
users.username
FROM concert_comments
JOIN users
ON users.id = concert_comments.user_id
WHERE concert_comments.concert_id = %s
ORDER BY concert_comments.created_at ASC
""",
(concert_id,),
)
comment_rows = cursor.fetchall()
cursor.execute(
"""
SELECT
concert_photos.id,
concert_photos.path,
concert_photos.created_at,
users.display_name,
users.username
FROM concert_photos
JOIN users
ON users.id = concert_photos.user_id
WHERE concert_photos.concert_id = %s
ORDER BY concert_photos.created_at DESC
""",
(concert_id,),
)
photo_rows = cursor.fetchall()
cursor.execute(
"""
SELECT
concert_attendance.user_id,
concert_attendance.status,
users.display_name,
users.username
FROM concert_attendance
JOIN users ON users.id = concert_attendance.user_id
WHERE concert_attendance.concert_id = %s
AND NOT EXISTS (
SELECT 1 FROM user_blocks b
WHERE (b.blocker_id = %s AND b.blocked_id = concert_attendance.user_id)
OR (b.blocker_id = concert_attendance.user_id AND b.blocked_id = %s)
)
ORDER BY users.display_name NULLS LAST, users.username
""",
(concert_id, user["id"], user["id"]),
)
attendance_rows = cursor.fetchall()
cursor.execute(
"SELECT band_key FROM followed_bands WHERE user_id = %s",
(user["id"],),
)
followed_band_keys = {row[0] for row in cursor.fetchall()}
venue_id = concert["venue"]["id"]
if venue_id:
cursor.execute(
"SELECT EXISTS (SELECT 1 FROM followed_venues WHERE user_id = %s AND venue_id = %s)",
(user["id"], venue_id),
)
follows_venue = cursor.fetchone()[0]
else:
follows_venue = False
cursor.execute(
"SELECT rating, COALESCE(favorite_song, ''), COALESCE(notes, ''), photo_path FROM concert_diary WHERE user_id = %s AND concert_id = %s",
(user["id"], concert_id),
)
diary_row = cursor.fetchone()
comments = [
{
"id": row[0],
"body": row[1],
"created_at": row[2].strftime("%d.%m.%Y %H:%M"),
"author": row[3] or row[4],
"username": row[4],
}
for row in comment_rows
]
photos = [
{
"id": row[0],
"path": row[1],
"created_at": row[2].strftime("%d.%m.%Y %H:%M"),
"author": row[3] or row[4],
}
for row in photo_rows
]
attendance = [
{
"user_id": row[0],
"status": row[1],
"name": row[2] or row[3],
"username": row[3],
}
for row in attendance_rows
]
attending_users = [item for item in attendance if item["status"] == "attending"]
ticket_seekers = [item for item in attendance if item["status"] == "ticket_search"]
ticket_offers = [item for item in attendance if item["status"] == "ticket_offer"]
maybe_users = [item for item in attendance if item["status"] == "maybe"]
current_attendance = next(
(item["status"] for item in attendance if item["user_id"] == user["id"]),
None,
)
diary_entry = {
"rating": diary_row[0], "favorite_song": diary_row[1], "notes": diary_row[2],
"photo_path": diary_row[3]
} if diary_row else None
concert_bands = load_concert_bands(concert_id, concert["artist"], concert["event_type"])
for band in concert_bands:
band["is_following"] = band["key"] in followed_band_keys
template = templates.get_template("concert_detail.html")
return template.render(
user=user,
concert=concert,
comments=comments,
photos=photos,
can_edit=can_edit_concert(user, concert),
can_delete=can_delete_concert(user, concert),
current_attendance=current_attendance,
attending_users=attending_users,
attending_count=len(attending_users),
ticket_seekers=ticket_seekers,
ticket_seeker_count=len(ticket_seekers),
ticket_offers=ticket_offers,
ticket_offer_count=len(ticket_offers),
maybe_users=maybe_users,
maybe_count=len(maybe_users),
concert_bands=concert_bands,
follows_venue=follows_venue,
diary_entry=diary_entry,
can_write_diary=concert["is_past"] and current_attendance == "attending",
)
# ============================================================
# Create concert
# ============================================================
@app.post("/concerts")
async def create_concert(
request: Request,
artist: str = Form(...),
band_names: str = Form(""),
event_type: str = Form("concert"),
parent_event_id: str = Form(""),
visibility: str = Form("public"),
invited_user_ids: list[int] = Form(default=[]),
venue_id: str = Form(""),
venue_name: str = Form(""),
city: str = Form(""),
street: str = Form(""),
postal_code: str = Form(""),
country: str = Form(""),
latitude: str = Form(""),
longitude: str = Form(""),
start_datetime: str = Form(...),
end_datetime: str = Form(""),
description: str = Form(""),
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)
if not user:
return login_redirect("/concerts/new")
artist = artist.strip()
if len(artist) < 2 or len(artist) > 255:
return HTMLResponse("Der Titel oder Künstlername muss zwischen 2 und 255 Zeichen lang sein.
", status_code=400)
if event_type not in EVENT_TYPES:
return HTMLResponse("Ungültige Veranstaltungskategorie.
", status_code=400)
parsed_bands = parse_band_names(band_names, artist, event_type)
if event_type == "festival" and not end_datetime:
return HTMLResponse("Bei Festivals ist ein Enddatum erforderlich.
", status_code=400)
if event_type != "festival":
end_datetime = ""
if visibility not in {"public", "friends", "private"}:
return HTMLResponse("Ungültige Sichtbarkeit.
", status_code=400)
if event_type != "other":
visibility = "public"
duplicate_matches = find_duplicate_concerts(user, artist, start_datetime, band_names)
if duplicate_matches and not duplicate_confirmed:
match_items = "".join(
f'{escape(match["artist"])} · {match["date"]} {match["time"]}'
for match in duplicate_matches
)
return HTMLResponse(
"Mögliche doppelte Veranstaltung
"
"Am selben Tag existiert bereits eine Veranstaltung mit einem sehr ähnlichen Künstlernamen.
"
f""
"Bitte gehe zurück, prüfe den Treffer und bestätige den Hinweis im Formular, wenn du trotzdem speichern möchtest.
",
status_code=409,
)
if end_datetime and datetime.fromisoformat(end_datetime) < datetime.fromisoformat(start_datetime):
return HTMLResponse("Das Enddatum darf nicht vor dem Beginn liegen.
", status_code=400)
try:
normalized_flyer_url = normalize_external_url(flyer_url, "den Flyer")
except ValueError as error:
return HTMLResponse(f"{error}
", status_code=400)
flyer_path, error = save_image(
flyer,
UPLOAD_DIR,
"/static/uploads/flyers/",
)
if error:
return error
with get_db_connection() as connection:
with connection.cursor() as cursor:
try:
selected_parent_event_id = resolve_event_relationship(
cursor, event_type, parent_event_id
)
except ValueError as error:
return HTMLResponse(f"{error}
", status_code=400)
selected_venue_id = resolve_venue(
cursor,
venue_id,
venue_name,
city,
street,
postal_code,
country,
latitude,
longitude,
)
cursor.execute(
"""
INSERT INTO concerts (
artist,
event_type,
parent_event_id,
venue_id,
start_datetime,
end_datetime,
description,
ticket_url,
ticket_price,
flyer_path,
flyer_url,
visibility,
created_by
)
VALUES (
%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
)
RETURNING id
""",
(
artist,
event_type,
selected_parent_event_id,
selected_venue_id,
start_datetime,
end_datetime or None,
description or None,
ticket_url or None,
ticket_price or None,
flyer_path,
normalized_flyer_url,
visibility,
user["id"],
),
)
concert_id = cursor.fetchone()[0]
replace_concert_bands(cursor, concert_id, parsed_bands)
if visibility == "private" and invited_user_ids:
cursor.execute(
"""
INSERT INTO event_invitations (concert_id, user_id, invited_by)
SELECT %s, id, %s FROM users
WHERE id = ANY(%s) AND id <> %s
ON CONFLICT (concert_id, user_id) DO NOTHING
""",
(concert_id, user["id"], invited_user_ids, user["id"]),
)
connection.commit()
return RedirectResponse(
f"/concerts/{concert_id}",
status_code=303
)
# ============================================================
# Edit concert
# ============================================================
@app.get(
"/concerts/{concert_id}/edit",
response_class=HTMLResponse
)
def edit_concert_page(request: Request, concert_id: int):
user = get_current_user(request)
if not user:
return login_redirect(f"/concerts/{concert_id}/edit")
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse(
"Veranstaltung nicht gefunden
",
status_code=404
)
if not can_edit_concert(user, concert):
return HTMLResponse(
"Vergangene Veranstaltungen dürfen nur Admins bearbeiten.
",
status_code=403
)
template = templates.get_template("edit_concert.html")
return template.render(
user=user,
concert=concert,
linkable_events=get_linkable_events(exclude_id=concert_id),
can_edit_title=can_edit_title(user, concert),
can_edit_details=can_edit_details(user, concert),
can_delete=can_delete_concert(user, concert),
invitable_users=get_invitable_users(user["id"]),
invited_user_ids=get_event_invitee_ids(concert_id),
can_manage_access=can_manage_event_access(user, concert),
band_names="\n".join(band["name"] for band in load_concert_bands(concert_id, concert["artist"], concert["event_type"])),
)
@app.post("/concerts/{concert_id}/edit")
async def edit_concert(
request: Request,
concert_id: int,
artist: str = Form(""),
band_names: str = Form(""),
event_type: str = Form("concert"),
parent_event_id: str = Form(""),
visibility: str = Form("public"),
invited_user_ids: list[int] = Form(default=[]),
venue_id: str = Form(""),
venue_name: str = Form(""),
city: str = Form(""),
street: str = Form(""),
postal_code: str = Form(""),
country: str = Form(""),
latitude: str = Form(""),
longitude: str = Form(""),
start_datetime: str = Form(""),
end_datetime: str = Form(""),
description: str = Form(""),
ticket_url: str = Form(""),
ticket_price: str = Form(""),
flyer_url: str = Form(""),
flyer: UploadFile | None = File(None)
):
user = get_current_user(request)
if not user:
return login_redirect(f"/concerts/{concert_id}/edit")
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse(
"Veranstaltung nicht gefunden
",
status_code=404
)
if not can_edit_concert(user, concert):
return HTMLResponse(
"Nicht erlaubt
",
status_code=403
)
next_artist = concert["artist"]
if can_edit_title(user, concert) and artist.strip():
next_artist = artist.strip()
next_start = concert["start_datetime"]
next_end = concert["end_datetime"]
next_description = concert["description"]
next_ticket_url = concert["ticket_url"]
next_ticket_price = concert["ticket_price"]
next_flyer = concert["flyer_path"]
next_flyer_url = concert["flyer_url"]
next_venue_id = concert["venue"]["id"]
next_event_type = concert["event_type"]
next_visibility = concert["visibility"]
next_parent_event_id = concert["parent_event"]["id"] if concert["parent_event"] else None
if can_edit_details(user, concert):
if not can_manage_event_access(user, concert):
event_type = concert["event_type"]
if event_type not in EVENT_TYPES:
return HTMLResponse("Ungültige Veranstaltungskategorie.
", status_code=400)
if event_type == "festival" and not end_datetime:
return HTMLResponse("Bei Festivals ist ein Enddatum erforderlich.
", status_code=400)
if event_type != "festival":
end_datetime = ""
if can_manage_event_access(user, concert):
if visibility not in {"public", "friends", "private"}:
return HTMLResponse("Ungültige Sichtbarkeit.
", status_code=400)
if event_type != "other":
visibility = "public"
else:
visibility = concert["visibility"]
effective_start = start_datetime or concert["start_local"]
if end_datetime and datetime.fromisoformat(end_datetime) < datetime.fromisoformat(effective_start):
return HTMLResponse("Das Enddatum darf nicht vor dem Beginn liegen.
", status_code=400)
try:
next_flyer_url = normalize_external_url(flyer_url, "den Flyer")
except ValueError as error:
return HTMLResponse(f"{error}
", status_code=400)
flyer_path, error = save_image(
flyer,
UPLOAD_DIR,
"/static/uploads/flyers/",
)
if error:
return error
with get_db_connection() as connection:
with connection.cursor() as cursor:
try:
next_parent_event_id = resolve_event_relationship(
cursor, event_type, parent_event_id, concert_id
)
except ValueError as error:
return HTMLResponse(f"{error}
", status_code=400)
resolved_venue_id = resolve_venue(
cursor,
venue_id,
venue_name or (
concert["venue"]["name"]
if concert["venue"]["id"] else ""
),
city,
street,
postal_code,
country,
latitude,
longitude,
)
connection.commit()
next_start = start_datetime or concert["start_datetime"]
next_end = end_datetime or None
next_event_type = event_type
next_visibility = visibility
next_description = description or None
next_ticket_url = ticket_url or None
next_ticket_price = ticket_price or None
next_flyer = flyer_path or concert["flyer_path"]
next_venue_id = resolved_venue_id or concert["venue"]["id"]
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
UPDATE concerts
SET
artist = %s,
event_type = %s,
parent_event_id = %s,
venue_id = %s,
start_datetime = %s,
end_datetime = %s,
description = %s,
ticket_url = %s,
ticket_price = %s,
flyer_path = %s,
flyer_url = %s
, visibility = %s
WHERE id = %s
""",
(
next_artist,
next_event_type,
next_parent_event_id,
next_venue_id,
next_start,
next_end or None,
next_description,
next_ticket_url,
next_ticket_price,
next_flyer,
next_flyer_url,
next_visibility,
concert_id,
),
)
if can_manage_event_access(user, concert) and next_visibility == "private":
cursor.execute(
"DELETE FROM event_invitations WHERE concert_id = %s AND NOT (user_id = ANY(%s))",
(concert_id, invited_user_ids or [0]),
)
if invited_user_ids:
cursor.execute(
"""
INSERT INTO event_invitations (concert_id, user_id, invited_by)
SELECT %s, id, %s FROM users WHERE id = ANY(%s) AND id <> %s
ON CONFLICT (concert_id, user_id) DO NOTHING
""",
(concert_id, user["id"], invited_user_ids, user["id"]),
)
elif can_manage_event_access(user, concert):
cursor.execute("DELETE FROM event_invitations WHERE concert_id = %s", (concert_id,))
if can_edit_title(user, concert):
replace_concert_bands(
cursor, concert_id, parse_band_names(band_names, next_artist, next_event_type)
)
connection.commit()
return RedirectResponse(
f"/concerts/{concert_id}",
status_code=303
)
@app.post("/concerts/{concert_id}/delete")
def delete_concert(request: Request, concert_id: int):
user = get_current_user(request)
if not user:
return login_redirect(f"/concerts/{concert_id}")
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse(
"Veranstaltung nicht gefunden
",
status_code=404
)
if not can_delete_concert(user, concert):
return HTMLResponse(
"Nicht erlaubt
",
status_code=403
)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT path FROM concert_photos WHERE concert_id = %s", (concert_id,))
photo_paths = [row[0] for row in cursor.fetchall()]
cursor.execute(
"DELETE FROM concerts WHERE id = %s",
(concert_id,),
)
connection.commit()
remove_uploaded_file(concert["flyer_path"], UPLOAD_DIR, "/static/uploads/flyers/")
for photo_path in photo_paths:
remove_uploaded_file(photo_path, PHOTO_DIR, "/static/uploads/photos/")
return RedirectResponse("/", status_code=303)
@app.post("/concerts/{concert_id}/attendance")
def set_attendance(
request: Request,
concert_id: int,
status: str = Form(...),
):
user = get_current_user(request)
if not user:
return login_redirect(f"/concerts/{concert_id}")
if status not in {"attending", "maybe", "ticket_search", "ticket_offer"}:
return HTMLResponse("Ungültige Auswahl
", status_code=400)
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse("Veranstaltung nicht gefunden
", status_code=404)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
INSERT INTO concert_attendance (concert_id, user_id, status)
VALUES (%s, %s, %s)
ON CONFLICT (concert_id, user_id)
DO UPDATE SET
status = EXCLUDED.status,
updated_at = CURRENT_TIMESTAMP
""",
(concert_id, user["id"], status),
)
connection.commit()
return RedirectResponse(f"/concerts/{concert_id}#attendance", status_code=303)
@app.post("/concerts/{concert_id}/comments")
def add_comment(
request: Request,
concert_id: int,
body: str = Form(...),
):
user = get_current_user(request)
if not user:
return login_redirect(f"/concerts/{concert_id}")
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse(
"Veranstaltung nicht gefunden
",
status_code=404
)
text = body.strip()
if not text:
return RedirectResponse(
f"/concerts/{concert_id}",
status_code=303
)
if len(text) > 2000:
return HTMLResponse(
"Kommentar ist zu lang (maximal 2000 Zeichen).
",
status_code=400
)
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
INSERT INTO concert_comments (
concert_id,
user_id,
body
)
VALUES (%s, %s, %s)
""",
(concert_id, user["id"], text),
)
connection.commit()
return RedirectResponse(
f"/concerts/{concert_id}#comments",
status_code=303
)
@app.post("/concerts/{concert_id}/photos")
async def add_photo(
request: Request,
concert_id: int,
photo: UploadFile | None = File(None),
):
user = get_current_user(request)
if not user:
return login_redirect(f"/concerts/{concert_id}")
concert = load_concert(concert_id)
if not concert or not can_view_event(user, concert):
return HTMLResponse(
"Veranstaltung nicht gefunden
",
status_code=404
)
if not photo or not photo.filename:
return RedirectResponse(
f"/concerts/{concert_id}",
status_code=303
)
path, error = save_image(
photo,
PHOTO_DIR,
"/static/uploads/photos/",
)
if error:
return error
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
INSERT INTO concert_photos (
concert_id,
user_id,
path
)
VALUES (%s, %s, %s)
""",
(concert_id, user["id"], path),
)
connection.commit()
return RedirectResponse(
f"/concerts/{concert_id}#photos",
status_code=303
)
# ============================================================
# Venue search
# ============================================================
def legacy_search_venues(q: str):
q = q.strip()
if len(q) < 2:
return []
results = []
query_tokens = [
token
for token in re.findall(r"[a-z0-9]+", q.lower())
if len(token) >= 3
]
token_match = " AND ".join("name ILIKE %s" for _ in query_tokens) or "FALSE"
# ========================================================
# Local database
# ========================================================
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(f"""
SELECT
id,
name,
street,
postal_code,
city,
country,
latitude,
longitude,
external_id,
source
FROM venues
WHERE
name ILIKE %s
OR city ILIKE %s
OR street ILIKE %s
OR ({token_match})
ORDER BY
CASE
WHEN LOWER(name) = LOWER(%s)
THEN 0
WHEN LOWER(name) LIKE LOWER(%s)
THEN 1
WHEN LOWER(name) LIKE LOWER(%s)
THEN 2
ELSE 3
END,
name
LIMIT 10
""", (
f"%{q}%",
f"%{q}%",
f"%{q}%",
*[f"%{token}%" for token in query_tokens],
q,
f"{q}%",
f"%{q}%",
))
rows = cursor.fetchall()
for row in rows:
(
venue_id,
name,
street,
postal_code,
city,
country,
latitude,
longitude,
external_id,
source
) = row
results.append({
"id": venue_id,
"name": name,
"street": street,
"postal_code": postal_code,
"city": city,
"country": country,
"latitude": latitude,
"longitude": longitude,
"external_id": external_id,
"source": source,
"local": True
})
# ========================================================
# Nominatim
# ========================================================
if not results:
headers = {
"User-Agent": "MetalCircle/1.0"
}
external_query = q
params = {
"q": external_query,
"format": "jsonv2",
"addressdetails": 1,
"limit": 20
}
try:
response = httpx.get(
"https://nominatim.openstreetmap.org/search",
params=params,
headers=headers,
timeout=8
)
response.raise_for_status()
data = response.json()
candidates = []
venue_types = {
"music_venue",
"concert_hall",
"stadium",
"sports_centre",
"theatre",
"arts_centre",
"exhibition_hall",
"conference_centre",
"events_venue",
"nightclub",
"community_centre",
"social_centre",
"festival",
"arena",
"auditorium",
"dance",
"cinema"
}
venue_keywords = [
"halle",
"arena",
"stadion",
"stadium",
"club",
"klub",
"theater",
"theatre",
"bühne",
"buehne",
"concert",
"konzert",
"music",
"musik",
"festival",
"event",
"venue",
"zentrum",
"center",
"centre",
"matrix",
"turbinenhalle",
"westfalenhalle"
]
excluded_types = {
"street",
"road",
"residential",
"postcode",
"house",
"railway",
"bus_stop",
"station",
"person"
}
for item in data:
address = item.get(
"address",
{}
)
name = (
item.get("name")
or ""
).strip()
display_name = (
item.get("display_name")
or ""
)
if not name:
name = display_name.split(
","
)[0].strip()
if not name:
continue
osm_type = (
item.get("type")
or ""
).lower()
osm_class = (
item.get("class")
or ""
).lower()
if osm_type in excluded_types:
continue
name_lower = name.lower()
query_lower = q.lower()
display_lower = display_name.lower()
score = 0
name_tokens = re.findall(r"[a-z0-9]+", name_lower)
matched_tokens = sum(
any(candidate.startswith(token) or token.startswith(candidate)
for candidate in name_tokens)
for token in query_tokens
)
score += matched_tokens * 30
if name_lower == query_lower:
score += 120
elif name_lower.startswith(
query_lower
):
score += 100
elif query_lower in name_lower:
score += 80
elif query_lower in display_lower:
score += 40
if osm_type in venue_types:
score += 70
for keyword in venue_keywords:
if keyword in name_lower:
score += 40
break
if osm_class in {
"amenity",
"leisure",
"tourism"
}:
score += 20
if osm_type in {
"street",
"road",
"residential",
"person",
"postcode",
"house"
}:
score -= 200
if score < 50:
continue
candidates.append({
"id": None,
"name": name,
"street":
address.get("road")
or address.get("pedestrian")
or address.get("footway"),
"postal_code":
address.get("postcode"),
"city":
address.get("city")
or address.get("town")
or address.get("village")
or address.get("municipality"),
"country":
address.get(
"country",
"Deutschland"
),
"latitude":
float(item["lat"])
if item.get("lat")
else None,
"longitude":
float(item["lon"])
if item.get("lon")
else None,
"external_id":
item.get("osm_id"),
"source":
"nominatim",
"local":
False,
"_score":
score
})
unique = {}
for candidate in candidates:
key = (
candidate["external_id"],
candidate["name"],
candidate["city"]
)
if key not in unique:
unique[key] = candidate
elif (
candidate["_score"]
>
unique[key]["_score"]
):
unique[key] = candidate
sorted_candidates = sorted(
unique.values(),
key=lambda item:
item["_score"],
reverse=True
)
for candidate in sorted_candidates[:10]:
candidate.pop(
"_score",
None
)
results.append(candidate)
except Exception as error:
print(
f"Nominatim search failed: {error}"
)
return results
def venue_search_tokens(value: str):
return [token for token in re.findall(r"[^\W_]+", value.casefold()) if len(token) >= 2]
def venue_result_key(venue):
return (
(venue.get("name") or "").strip().casefold(),
(venue.get("city") or "").strip().casefold(),
(venue.get("country") or "").strip().casefold(),
)
@app.get("/api/venues/search")
def search_venues(q: str):
query = q.strip()
if len(query) < 2:
return []
tokens = venue_search_tokens(query)
token_clauses = []
token_values = []
for token in tokens:
token_clauses.append(
"""(
venues.name ILIKE %s OR venues.city ILIKE %s OR venues.country ILIKE %s
OR EXISTS (
SELECT 1 FROM venue_aliases
WHERE venue_aliases.venue_id = venues.id
AND venue_aliases.alias ILIKE %s
)
)"""
)
token_values.extend([f"%{token}%"] * 4)
token_match = " AND ".join(token_clauses) or "FALSE"
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute(
f"""
SELECT venues.id, venues.name, venues.street, venues.postal_code,
venues.city, venues.country, venues.latitude, venues.longitude,
venues.external_id, venues.source, venues.is_verified
FROM venues
WHERE venues.name ILIKE %s OR venues.city ILIKE %s
OR venues.country ILIKE %s OR venues.street ILIKE %s
OR EXISTS (
SELECT 1 FROM venue_aliases
WHERE venue_aliases.venue_id = venues.id
AND venue_aliases.alias ILIKE %s
)
OR ({token_match})
ORDER BY venues.is_verified DESC,
CASE WHEN LOWER(venues.name) = LOWER(%s) THEN 0
WHEN venues.name ILIKE %s THEN 1 ELSE 2 END,
venues.name
LIMIT 15
""",
[f"%{query}%"] * 5
+ token_values
+ [query, f"{query}%"],
)
rows = cursor.fetchall()
ranked_results = []
for row in rows:
ranked_results.append({
"id": row[0],
"name": row[1],
"street": row[2],
"postal_code": row[3],
"city": row[4],
"country": row[5],
"latitude": row[6],
"longitude": row[7],
"external_id": row[8],
"source": row[9],
"local": True,
"verified": bool(row[10]),
"_score": 300 + (50 if row[10] else 0),
})
try:
response = httpx.get(
"https://nominatim.openstreetmap.org/search",
params={
"q": query,
"format": "jsonv2",
"addressdetails": 1,
"namedetails": 1,
"limit": 25,
},
headers={"User-Agent": "MetalCircle/1.0"},
timeout=8,
)
response.raise_for_status()
venue_types = {
"music_venue", "concert_hall", "stadium", "sports_centre", "theatre",
"arts_centre", "exhibition_hall", "conference_centre", "events_venue",
"nightclub", "community_centre", "festival", "arena", "auditorium",
"cinema", "event_venue", "recreation_ground",
}
venue_keywords = {
"arena", "club", "festival", "halle", "hall", "stadion", "stadium",
"theater", "theatre", "concert", "konzert", "music", "venue", "centre",
"center", "festivalgelände", "festivalterrein",
}
excluded_types = {
"street", "road", "residential", "postcode", "house", "bus_stop", "person",
}
for item in response.json():
address = item.get("address") or {}
display_name = (item.get("display_name") or "").strip()
name = (item.get("name") or display_name.split(",", 1)[0]).strip()
osm_type = (item.get("type") or "").casefold()
osm_class = (item.get("class") or item.get("category") or "").casefold()
if not name or osm_type in excluded_types:
continue
searchable = " ".join((name, display_name)).casefold()
matched_tokens = sum(token in searchable for token in tokens)
score = matched_tokens * 35
if name.casefold() == query.casefold():
score += 120
elif name.casefold().startswith(query.casefold()):
score += 90
elif query.casefold() in searchable:
score += 60
if osm_type in venue_types:
score += 80
if any(keyword in name.casefold() for keyword in venue_keywords):
score += 35
if osm_class in {"amenity", "leisure", "tourism"}:
score += 20
if score < 50:
continue
osm_id = item.get("osm_id")
osm_kind = (item.get("osm_type") or "object").casefold()
external_id = f"{osm_kind}:{osm_id}" if osm_id is not None else None
ranked_results.append({
"id": f"nominatim:{external_id}" if external_id else None,
"name": name,
"street": address.get("road") or address.get("pedestrian"),
"postal_code": address.get("postcode"),
"city": address.get("city") or address.get("town") or address.get("village")
or address.get("municipality") or address.get("county"),
"country": address.get("country"),
"latitude": float(item["lat"]) if item.get("lat") else None,
"longitude": float(item["lon"]) if item.get("lon") else None,
"external_id": external_id,
"source": "nominatim",
"local": False,
"verified": False,
"_score": score,
})
except (httpx.HTTPError, ValueError) as error:
print(f"Nominatim search failed: {error}")
unique_results = {}
for venue in sorted(ranked_results, key=lambda item: item["_score"], reverse=True):
key = venue_result_key(venue)
if key not in unique_results:
venue.pop("_score", None)
unique_results[key] = venue
return list(unique_results.values())[:10]