import os
import uuid
import secrets
import hashlib
import httpx
import psycopg
from datetime import datetime
from fastapi import FastAPI, Form, UploadFile, File
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from jinja2 import Environment, FileSystemLoader, select_autoescape
app = FastAPI(title="Pingu Concerts")
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"
)
os.makedirs(
UPLOAD_DIR,
exist_ok=True
)
# ============================================================
# 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"
)
# ============================================================
# Database
# ============================================================
def get_db_connection():
return psycopg.connect(DATABASE_URL)
# ============================================================
# Token helpers
# ============================================================
def hash_token(token: str) -> str:
return hashlib.sha256(
token.encode("utf-8")
).hexdigest()
def generate_token() -> str:
return secrets.token_urlsafe(32)
@app.get("/admin/invites")
def create_invite():
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()
return {
"invite_id": invite_id,
"invite_url": f"/register/{token}"
}
@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) < 8:
return HTMLResponse(
"Fehler
Das Passwort muss mindestens 8 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
)
# Passwort hashen
import bcrypt
password_hash = bcrypt.hashpw(
password.encode("utf-8"),
bcrypt.gensalt()
).decode("utf-8")
# Benutzer anlegen
cursor.execute("""
INSERT INTO users (
username,
email,
password_hash,
display_name
)
VALUES (
%s,
%s,
%s,
%s
)
RETURNING id
""", (
username,
email,
password_hash,
display_name or username
))
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()
return HTMLResponse(
f"""
Account erstellt 🎸
Willkommen bei Pingu Concerts, {display_name or username}!
Dein Account wurde erfolgreich erstellt.
Zu Pingu Concerts
"""
)
# ============================================================
# 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
)
# ============================================================
# Home
# ============================================================
@app.get("/", response_class=HTMLResponse)
def home():
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("""
SELECT
concerts.id,
concerts.artist,
concerts.start_datetime,
venues.name,
venues.city
FROM concerts
LEFT JOIN venues
ON concerts.venue_id = venues.id
ORDER BY concerts.start_datetime
""")
rows = cursor.fetchall()
concerts = []
for row in rows:
(
concert_id,
artist,
start_datetime,
venue,
city
) = row
venue_text = venue or "Veranstaltungsort unbekannt"
if city:
venue_text += f", {city}"
concerts.append({
"id": concert_id,
"artist": artist,
"date": start_datetime.strftime("%d.%m.%Y"),
"time": start_datetime.strftime("%H:%M"),
"venue": venue_text
})
template = templates.get_template(
"index.html"
)
return template.render(
concerts=concerts
)
# ============================================================
# New concert
# ============================================================
@app.get(
"/concerts/new",
response_class=HTMLResponse
)
def new_concert():
template = templates.get_template(
"new_concert.html"
)
return template.render()
# ============================================================
# Concert detail
# ============================================================
@app.get(
"/concerts/{concert_id}",
response_class=HTMLResponse
)
def concert_detail(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,
venues.name,
venues.street,
venues.postal_code,
venues.city,
venues.country
FROM concerts
LEFT JOIN venues
ON concerts.venue_id = venues.id
WHERE concerts.id = %s
""", (
concert_id,
))
row = cursor.fetchone()
if not row:
return HTMLResponse(
"Konzert nicht gefunden
",
status_code=404
)
(
concert_id,
artist,
start_datetime,
end_datetime,
description,
ticket_url,
ticket_price,
flyer_path,
venue_name,
venue_street,
venue_postal_code,
venue_city,
venue_country
) = row
concert = {
"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,
"end_time":
end_datetime.strftime(
"%H:%M"
)
if end_datetime
else None,
"description":
description,
"ticket_url":
ticket_url,
"ticket_price":
ticket_price,
"flyer_path":
flyer_path,
"venue": {
"name":
venue_name
or "Veranstaltungsort unbekannt",
"street":
venue_street,
"postal_code":
venue_postal_code,
"city":
venue_city,
"country":
venue_country
}
}
template = templates.get_template(
"concert_detail.html"
)
return template.render(
concert=concert
)
# ============================================================
# Create concert
# ============================================================
@app.post("/concerts")
async def create_concert(
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)
):
flyer_path = None
# ========================================================
# Flyer speichern
# ========================================================
if flyer and flyer.filename:
allowed_extensions = {
".jpg",
".jpeg",
".png",
".webp"
}
original_name = flyer.filename
extension = os.path.splitext(
original_name
)[1].lower()
if extension not in allowed_extensions:
return HTMLResponse(
"Ungültiges Flyer-Format. "
"Erlaubt sind JPG, JPEG, PNG und WEBP.",
status_code=400
)
filename = (
str(uuid.uuid4())
+ extension
)
destination = os.path.join(
UPLOAD_DIR,
filename
)
contents = await flyer.read()
# 10 MB Limit
if len(contents) > 10 * 1024 * 1024:
return HTMLResponse(
"Der Flyer darf maximal 10 MB groß sein.",
status_code=400
)
with open(
destination,
"wb"
) as file:
file.write(contents)
flyer_path = (
"/static/uploads/flyers/"
+ filename
)
# ========================================================
# Datenbank
# ========================================================
with get_db_connection() as connection:
with connection.cursor() as cursor:
selected_venue_id = None
# ==================================================
# Venue auswählen
# ==================================================
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
# ==================================================
# Manueller Venue-Fallback
# ==================================================
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]
)
# ==================================================
# Konzert speichern
# ==================================================
cursor.execute("""
INSERT INTO concerts (
artist,
venue_id,
start_datetime,
end_datetime,
description,
ticket_url,
ticket_price,
flyer_path
)
VALUES (
%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
))
concert_id = (
cursor.fetchone()[0]
)
connection.commit()
return RedirectResponse(
f"/concerts/{concert_id}",
status_code=303
)
# ============================================================
# Venue search
# ============================================================
@app.get("/api/venues/search")
def search_venues(q: str):
q = q.strip()
if len(q) < 2:
return []
results = []
# ========================================================
# Local database
# ========================================================
with get_db_connection() as connection:
with connection.cursor() as cursor:
cursor.execute("""
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
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}%",
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"
}
params = {
"q": q,
"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
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