2252 lines
60 KiB
Python
2252 lines
60 KiB
Python
import os
|
|
import uuid
|
|
import secrets
|
|
import hashlib
|
|
import re
|
|
from contextlib import asynccontextmanager
|
|
from datetime import datetime, timedelta
|
|
|
|
import bcrypt
|
|
import httpx
|
|
import psycopg
|
|
|
|
from fastapi import FastAPI, File, Form, Request, UploadFile
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
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"
|
|
)
|
|
|
|
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
|
os.makedirs(PHOTO_DIR, exist_ok=True)
|
|
os.makedirs(AVATAR_DIR, exist_ok=True)
|
|
|
|
SESSION_COOKIE = "pingu_session"
|
|
SESSION_DAYS = 30
|
|
ALLOWED_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp"}
|
|
MAX_IMAGE_BYTES = 10 * 1024 * 1024
|
|
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 = (
|
|
("first_gig", "Erster Gig", "🎸", 1, "Dein erstes besuchtes Konzert"),
|
|
("regular", "Stammgast", "🤘", 5, "5 Konzerte besucht"),
|
|
("ten_gigs", "Zehnerrunde", "🔥", 10, "10 Konzerte besucht"),
|
|
("tour_veteran", "Tourveteran", "⚡", 25, "25 Konzerte besucht"),
|
|
)
|
|
|
|
|
|
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
|
|
""",
|
|
"""
|
|
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')
|
|
),
|
|
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (concert_id, user_id)
|
|
)
|
|
""",
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS user_badges (
|
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
badge_code VARCHAR(50) NOT NULL,
|
|
awarded_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (user_id, badge_code)
|
|
)
|
|
""",
|
|
]
|
|
|
|
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 id FROM users WHERE username = %s",
|
|
(INITIAL_ADMIN_USERNAME,),
|
|
)
|
|
existing_user = cursor.fetchone()
|
|
|
|
if not existing_user:
|
|
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="Pingu Concerts", lifespan=lifespan)
|
|
|
|
|
|
@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("/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 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]),
|
|
}
|
|
|
|
|
|
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
|
|
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",
|
|
)
|
|
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()
|
|
|
|
|
|
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
|
|
FROM concerts
|
|
LEFT JOIN venues
|
|
ON concerts.venue_id = venues.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],
|
|
"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 "",
|
|
"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 serialize_concert_card(row):
|
|
concert_id, artist, start_datetime, venue, city = 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"),
|
|
"venue": venue_text,
|
|
}
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
filename = str(uuid.uuid4()) + extension
|
|
destination = os.path.join(destination_dir, filename)
|
|
contents = upload.file.read()
|
|
|
|
if len(contents) > MAX_IMAGE_BYTES:
|
|
return None, HTMLResponse(
|
|
"Die Datei darf maximal 10 MB groß sein.",
|
|
status_code=400,
|
|
)
|
|
|
|
with open(destination, "wb") as file:
|
|
file.write(contents)
|
|
|
|
return f"{url_prefix}{filename}", None
|
|
|
|
|
|
def attended_concert_count(user_id: int) -> int:
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM concert_attendance
|
|
JOIN concerts ON concerts.id = concert_attendance.concert_id
|
|
WHERE concert_attendance.user_id = %s
|
|
AND concert_attendance.status = 'attending'
|
|
AND COALESCE(
|
|
concerts.end_datetime,
|
|
concerts.start_datetime
|
|
) < CURRENT_TIMESTAMP
|
|
""",
|
|
(user_id,),
|
|
)
|
|
return cursor.fetchone()[0]
|
|
|
|
|
|
def grant_earned_badges(user_id: int, attended_count: int):
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
for badge_code, _name, _icon, threshold, _description in BADGE_DEFINITIONS:
|
|
if attended_count >= threshold:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO user_badges (user_id, badge_code)
|
|
VALUES (%s, %s)
|
|
ON CONFLICT (user_id, badge_code) DO NOTHING
|
|
""",
|
|
(user_id, badge_code),
|
|
)
|
|
connection.commit()
|
|
|
|
|
|
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
|
|
FROM users
|
|
WHERE LOWER(username) = LOWER(%s)
|
|
""",
|
|
(username,),
|
|
)
|
|
row = cursor.fetchone()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
cursor.execute(
|
|
"SELECT badge_code FROM user_badges WHERE user_id = %s",
|
|
(row[0],),
|
|
)
|
|
earned_codes = {badge_row[0] for badge_row in cursor.fetchall()}
|
|
|
|
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"),
|
|
"earned_codes": earned_codes,
|
|
}
|
|
|
|
|
|
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
|
|
|
|
if venue_id:
|
|
if venue_id.startswith("nominatim:"):
|
|
external_id = venue_id.split(":", 1)[1]
|
|
|
|
cursor.execute(
|
|
"""
|
|
SELECT id
|
|
FROM venues
|
|
WHERE external_id = %s
|
|
AND source = 'nominatim'
|
|
LIMIT 1
|
|
""",
|
|
(external_id,),
|
|
)
|
|
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 "Deutschland",
|
|
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 "Deutschland",
|
|
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
|
|
|
|
|
|
@app.get("/admin", response_class=HTMLResponse)
|
|
def admin_page(request: Request):
|
|
user = require_admin(request)
|
|
|
|
if not user:
|
|
return HTMLResponse("<h1>Nicht erlaubt</h1>", status_code=403)
|
|
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT id, username, email, display_name, is_admin, created_at
|
|
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"),
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
template = templates.get_template("admin.html")
|
|
return template.render(user=user, users=users, invite_url=None)
|
|
|
|
|
|
@app.post("/admin/invites", response_class=HTMLResponse)
|
|
def create_invite(request: Request):
|
|
user = require_admin(request)
|
|
|
|
if not user:
|
|
return HTMLResponse("<h1>Nicht erlaubt</h1>", 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 id, username, email, display_name, is_admin, created_at
|
|
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"),
|
|
}
|
|
for row in rows
|
|
]
|
|
template = templates.get_template("admin.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/users/{user_id}")
|
|
def update_user_role(
|
|
request: Request,
|
|
user_id: int,
|
|
role: str = Form(...),
|
|
):
|
|
user = require_admin(request)
|
|
|
|
if not user:
|
|
return HTMLResponse("<h1>Nicht erlaubt</h1>", status_code=403)
|
|
if role not in {"admin", "user"}:
|
|
return HTMLResponse("<h1>Ungültige Rolle</h1>", status_code=400)
|
|
if user_id == user["id"] and role != "admin":
|
|
return HTMLResponse(
|
|
"<h1>Die eigenen Adminrechte können nicht entfernt werden.</h1>",
|
|
status_code=400,
|
|
)
|
|
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"UPDATE users SET is_admin = %s WHERE id = %s",
|
|
(role == "admin", user_id),
|
|
)
|
|
connection.commit()
|
|
|
|
return RedirectResponse("/admin", 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("<h1>Nicht erlaubt</h1>", status_code=403)
|
|
if user_id == user["id"]:
|
|
return HTMLResponse(
|
|
"<h1>Der eigene Account kann nicht gelöscht werden.</h1>",
|
|
status_code=400,
|
|
)
|
|
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute("DELETE FROM users WHERE id = %s", (user_id,))
|
|
connection.commit()
|
|
|
|
return RedirectResponse("/admin", 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(
|
|
"<h1>Fehler</h1><p>Der Benutzername muss mindestens 3 Zeichen lang sein.</p>",
|
|
status_code=400
|
|
)
|
|
|
|
if len(password) < 8:
|
|
return HTMLResponse(
|
|
"<h1>Fehler</h1><p>Das Passwort muss mindestens 8 Zeichen lang sein.</p>",
|
|
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(
|
|
"<h1>Ungültige Einladung</h1>",
|
|
status_code=404
|
|
)
|
|
|
|
invite_id, expires_at, used_at = invite
|
|
|
|
if used_at:
|
|
return HTMLResponse(
|
|
"<h1>Diese Einladung wurde bereits verwendet.</h1>",
|
|
status_code=410
|
|
)
|
|
|
|
if expires_at and datetime.now() > expires_at:
|
|
return HTMLResponse(
|
|
"<h1>Diese Einladung ist abgelaufen.</h1>",
|
|
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(
|
|
"<h1>Fehler</h1><p>Dieser Benutzername ist bereits vergeben.</p>",
|
|
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(
|
|
"<h1>Fehler</h1><p>Diese E-Mail-Adresse ist bereits registriert.</p>",
|
|
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(
|
|
"<h1>Ungültige Einladung</h1>",
|
|
status_code=404
|
|
)
|
|
|
|
invite_id, expires_at, used_at = invite
|
|
|
|
if used_at:
|
|
return HTMLResponse(
|
|
"<h1>Diese Einladung wurde bereits verwendet.</h1>",
|
|
status_code=410
|
|
)
|
|
|
|
if expires_at:
|
|
from datetime import datetime
|
|
|
|
if datetime.now() > expires_at:
|
|
return HTMLResponse(
|
|
"<h1>Diese Einladung ist abgelaufen.</h1>",
|
|
status_code=410
|
|
)
|
|
|
|
template = templates.get_template(
|
|
"register.html"
|
|
)
|
|
|
|
return template.render(
|
|
token=token
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# 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)
|
|
return response
|
|
|
|
|
|
# ============================================================
|
|
# Home
|
|
# ============================================================
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def home(request: Request):
|
|
concert_query = """
|
|
SELECT
|
|
concerts.id,
|
|
concerts.artist,
|
|
concerts.start_datetime,
|
|
venues.name,
|
|
venues.city
|
|
FROM concerts
|
|
LEFT JOIN venues
|
|
ON concerts.venue_id = venues.id
|
|
WHERE COALESCE(concerts.end_datetime, concerts.start_datetime)
|
|
{operator} CURRENT_TIMESTAMP
|
|
ORDER BY concerts.start_datetime {direction}
|
|
"""
|
|
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
concert_query.format(operator=">=", direction="ASC")
|
|
)
|
|
upcoming_rows = cursor.fetchall()
|
|
|
|
cursor.execute(
|
|
concert_query.format(operator="<", direction="DESC")
|
|
)
|
|
past_rows = cursor.fetchall()
|
|
|
|
template = templates.get_template("index.html")
|
|
return template.render(
|
|
user=get_current_user(request),
|
|
upcoming_concerts=[serialize_concert_card(row) for row in upcoming_rows],
|
|
past_concerts=[serialize_concert_card(row) for row in past_rows],
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# Profiles
|
|
# ============================================================
|
|
|
|
def render_profile(request: Request, username: str):
|
|
viewer = get_current_user(request)
|
|
profile = load_profile(username)
|
|
|
|
if not profile:
|
|
return HTMLResponse("<h1>Benutzer nicht gefunden</h1>", status_code=404)
|
|
|
|
attended_count = attended_concert_count(profile["id"])
|
|
grant_earned_badges(profile["id"], attended_count)
|
|
profile = load_profile(username)
|
|
badges = [
|
|
{
|
|
"code": code,
|
|
"name": name,
|
|
"icon": icon,
|
|
"threshold": threshold,
|
|
"description": description,
|
|
"earned": code in profile["earned_codes"],
|
|
}
|
|
for code, name, icon, threshold, description in BADGE_DEFINITIONS
|
|
]
|
|
|
|
template = templates.get_template("profile.html")
|
|
return template.render(
|
|
user=viewer,
|
|
profile=profile,
|
|
attended_count=attended_count,
|
|
badges=badges,
|
|
is_own_profile=viewer["id"] == profile["id"],
|
|
)
|
|
|
|
|
|
@app.get("/profile", response_class=HTMLResponse)
|
|
def own_profile(request: Request):
|
|
user = get_current_user(request)
|
|
return render_profile(request, user["username"])
|
|
|
|
|
|
@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(""),
|
|
avatar: UploadFile | None = File(None),
|
|
):
|
|
user = get_current_user(request)
|
|
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(
|
|
"""
|
|
UPDATE users
|
|
SET
|
|
display_name = COALESCE(NULLIF(%s, ''), display_name),
|
|
avatar_path = COALESCE(%s, avatar_path)
|
|
WHERE id = %s
|
|
""",
|
|
(display_name.strip(), avatar_path, user["id"]),
|
|
)
|
|
connection.commit()
|
|
|
|
return RedirectResponse("/profile", 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)
|
|
|
|
|
|
# ============================================================
|
|
# Concert detail
|
|
# ============================================================
|
|
|
|
@app.get(
|
|
"/concerts/{concert_id}",
|
|
response_class=HTMLResponse
|
|
)
|
|
def concert_detail(request: Request, concert_id: int):
|
|
concert = load_concert(concert_id)
|
|
|
|
if not concert:
|
|
return HTMLResponse(
|
|
"<h1>Konzert nicht gefunden</h1>",
|
|
status_code=404
|
|
)
|
|
|
|
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
|
|
ORDER BY users.display_name NULLS LAST, users.username
|
|
""",
|
|
(concert_id,),
|
|
)
|
|
attendance_rows = cursor.fetchall()
|
|
|
|
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"]
|
|
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,
|
|
)
|
|
|
|
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),
|
|
maybe_users=maybe_users,
|
|
maybe_count=len(maybe_users),
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# Create concert
|
|
# ============================================================
|
|
|
|
@app.post("/concerts")
|
|
async def create_concert(
|
|
request: Request,
|
|
artist: str = Form(...),
|
|
venue_id: str = Form(""),
|
|
venue_name: str = Form(""),
|
|
city: str = Form(""),
|
|
street: str = Form(""),
|
|
postal_code: str = Form(""),
|
|
country: str = Form("Deutschland"),
|
|
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: UploadFile | None = File(None)
|
|
):
|
|
user = get_current_user(request)
|
|
|
|
if not user:
|
|
return login_redirect("/concerts/new")
|
|
|
|
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:
|
|
selected_venue_id = resolve_venue(
|
|
cursor,
|
|
venue_id,
|
|
venue_name,
|
|
city,
|
|
street,
|
|
postal_code,
|
|
country,
|
|
latitude,
|
|
longitude,
|
|
)
|
|
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO concerts (
|
|
artist,
|
|
venue_id,
|
|
start_datetime,
|
|
end_datetime,
|
|
description,
|
|
ticket_url,
|
|
ticket_price,
|
|
flyer_path,
|
|
created_by
|
|
)
|
|
VALUES (
|
|
%s, %s, %s, %s, %s, %s, %s, %s, %s
|
|
)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
artist,
|
|
selected_venue_id,
|
|
start_datetime,
|
|
end_datetime or None,
|
|
description or None,
|
|
ticket_url or None,
|
|
ticket_price or None,
|
|
flyer_path,
|
|
user["id"],
|
|
),
|
|
)
|
|
|
|
concert_id = cursor.fetchone()[0]
|
|
|
|
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:
|
|
return HTMLResponse(
|
|
"<h1>Konzert nicht gefunden</h1>",
|
|
status_code=404
|
|
)
|
|
|
|
if not can_edit_concert(user, concert):
|
|
return HTMLResponse(
|
|
"<h1>Vergangene Veranstaltungen dürfen nur Admins bearbeiten.</h1>",
|
|
status_code=403
|
|
)
|
|
|
|
template = templates.get_template("edit_concert.html")
|
|
return template.render(
|
|
user=user,
|
|
concert=concert,
|
|
can_edit_title=can_edit_title(user, concert),
|
|
can_edit_details=can_edit_details(user, concert),
|
|
can_delete=can_delete_concert(user, concert),
|
|
)
|
|
|
|
|
|
@app.post("/concerts/{concert_id}/edit")
|
|
async def edit_concert(
|
|
request: Request,
|
|
concert_id: int,
|
|
artist: str = Form(""),
|
|
venue_id: str = Form(""),
|
|
venue_name: str = Form(""),
|
|
city: str = Form(""),
|
|
street: str = Form(""),
|
|
postal_code: str = Form(""),
|
|
country: str = Form("Deutschland"),
|
|
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: 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:
|
|
return HTMLResponse(
|
|
"<h1>Konzert nicht gefunden</h1>",
|
|
status_code=404
|
|
)
|
|
|
|
if not can_edit_concert(user, concert):
|
|
return HTMLResponse(
|
|
"<h1>Nicht erlaubt</h1>",
|
|
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_venue_id = concert["venue"]["id"]
|
|
|
|
if can_edit_details(user, concert):
|
|
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:
|
|
resolved_venue_id = resolve_venue(
|
|
cursor,
|
|
venue_id,
|
|
venue_name or concert["venue"]["name"],
|
|
city,
|
|
street,
|
|
postal_code,
|
|
country,
|
|
latitude,
|
|
longitude,
|
|
)
|
|
connection.commit()
|
|
|
|
next_start = start_datetime or concert["start_datetime"]
|
|
next_end = end_datetime or None
|
|
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,
|
|
venue_id = %s,
|
|
start_datetime = %s,
|
|
end_datetime = %s,
|
|
description = %s,
|
|
ticket_url = %s,
|
|
ticket_price = %s,
|
|
flyer_path = %s
|
|
WHERE id = %s
|
|
""",
|
|
(
|
|
next_artist,
|
|
next_venue_id,
|
|
next_start,
|
|
next_end or None,
|
|
next_description,
|
|
next_ticket_url,
|
|
next_ticket_price,
|
|
next_flyer,
|
|
concert_id,
|
|
),
|
|
)
|
|
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:
|
|
return HTMLResponse(
|
|
"<h1>Konzert nicht gefunden</h1>",
|
|
status_code=404
|
|
)
|
|
|
|
if not can_delete_concert(user, concert):
|
|
return HTMLResponse(
|
|
"<h1>Nicht erlaubt</h1>",
|
|
status_code=403
|
|
)
|
|
|
|
with get_db_connection() as connection:
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"DELETE FROM concerts WHERE id = %s",
|
|
(concert_id,),
|
|
)
|
|
connection.commit()
|
|
|
|
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"}:
|
|
return HTMLResponse("<h1>Ungültige Auswahl</h1>", status_code=400)
|
|
if not load_concert(concert_id):
|
|
return HTMLResponse("<h1>Konzert nicht gefunden</h1>", 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:
|
|
return HTMLResponse(
|
|
"<h1>Konzert nicht gefunden</h1>",
|
|
status_code=404
|
|
)
|
|
|
|
text = body.strip()
|
|
|
|
if not text:
|
|
return RedirectResponse(
|
|
f"/concerts/{concert_id}",
|
|
status_code=303
|
|
)
|
|
|
|
if len(text) > 2000:
|
|
return HTMLResponse(
|
|
"<h1>Kommentar ist zu lang (maximal 2000 Zeichen).</h1>",
|
|
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:
|
|
return HTMLResponse(
|
|
"<h1>Konzert nicht gefunden</h1>",
|
|
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
|
|
# ============================================================
|
|
|
|
@app.get("/api/venues/search")
|
|
def 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": "PinguConcerts/1.0"
|
|
}
|
|
|
|
external_query = re.sub(r"\bhall\b", "halle", q, flags=re.IGNORECASE)
|
|
|
|
params = {
|
|
|
|
"q": external_query,
|
|
|
|
"format": "jsonv2",
|
|
|
|
"addressdetails": 1,
|
|
|
|
"limit": 20,
|
|
|
|
"countrycodes": "de"
|
|
|
|
}
|
|
|
|
|
|
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
|