"""Wildfleisch-Reservierungsserver, ausschließlich Standardbibliothek."""
import argparse
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
import getpass
import hashlib
import hmac
import html
import os
import re
import secrets
import sqlite3
import time
from http import cookies
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, quote
ROOT = Path(__file__).resolve().parent
DB = ROOT / 'data' / 'wildfleisch.sqlite3'
# Öffentlicher URL-Präfix hinter dem Reverse-Proxy, z. B. /fabian. Leer = Server-Wurzel.
BASE = os.environ.get('WILD_BASE', '').rstrip('/')
def u(path):
"""Pfad für Links, Formulare und Weiterleitungen, inklusive URL-Präfix."""
if not path.startswith('/'):
path = '/' + path
return BASE + path if BASE else path
EXAMPLES_DIR = ROOT / 'beispiele'
UPLOADS = ROOT / 'data' / 'images'
EXAMPLES = {
'ruecken.jpg': 'Hirsch · Rücken',
'schulter.jpg': 'Hirsch · Schulter',
'keule.jpg': 'Hirsch · Keule',
'ragout.jpg': 'Hirsch · Ragout',
'hack.jpg': 'Hirsch · Hack',
'hals.jpg': 'Hirsch · Hals',
'wildschwein-ruecken.jpg': 'Wildschwein · Rücken',
'wildschwein-schulter.jpg': 'Wildschwein · Schulter',
'reh-ruecken.jpg': 'Reh · Rücken',
'hase-keule.jpg': 'Hase · Keule',
}
MAX_IMAGE = 2_000_000
IMAGE_TYPES = {'jpg': 'image/jpeg', 'png': 'image/png', 'gif': 'image/gif', 'webp': 'image/webp'}
ANIMALS = ('Rotwild Hirsch', 'Rotwild Kalb', 'Wildschwein', 'Reh')
CUTS = ('Rücken', 'Schulter', 'Keule', 'Hals', 'Hack', 'Gulasch')
CUT_KEYS = (
('Gulasch', ('gulasch', 'ragout')),
('Hack', ('hack',)),
('Rücken', ('rücken', 'ruecken')),
('Schulter', ('schulter',)),
('Keule', ('keule',)),
('Hals', ('hals',)),
)
SESSIONS = {}
ATTEMPTS = {}
class Database(sqlite3.Connection):
def __exit__(self, *args):
try:
return super().__exit__(*args)
finally:
self.close()
def connect():
db = sqlite3.connect(DB, timeout=10, factory=Database)
db.row_factory = sqlite3.Row
db.execute('PRAGMA foreign_keys=ON')
return db
def initialize():
DB.parent.mkdir(parents=True, exist_ok=True)
with connect() as db:
db.executescript('''
CREATE TABLE IF NOT EXISTS admin (id INTEGER PRIMARY KEY CHECK(id=1), salt TEXT NOT NULL, hash TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS products (id INTEGER PRIMARY KEY, name TEXT NOT NULL, description TEXT NOT NULL, unit TEXT NOT NULL, stock INTEGER NOT NULL CHECK(stock>=0));
CREATE TABLE IF NOT EXISTS reservations (id INTEGER PRIMARY KEY, reference TEXT UNIQUE NOT NULL, name TEXT NOT NULL, contact TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'offen', created TEXT DEFAULT CURRENT_TIMESTAMP);
CREATE TABLE IF NOT EXISTS items (reservation_id INTEGER REFERENCES reservations(id), product_id INTEGER REFERENCES products(id), quantity INTEGER NOT NULL CHECK(quantity>0), PRIMARY KEY(reservation_id, product_id));
''')
columns = {row[1] for row in db.execute('PRAGMA table_info(products)')}
if 'image' not in columns:
db.execute('ALTER TABLE products ADD COLUMN image TEXT')
if 'price_cents' not in columns:
db.execute('ALTER TABLE products ADD COLUMN price_cents INTEGER')
if 'animal' not in columns:
db.execute('ALTER TABLE products ADD COLUMN animal TEXT')
if 'cut' not in columns:
db.execute('ALTER TABLE products ADD COLUMN cut TEXT')
for row in db.execute('SELECT id, name, animal, cut FROM products'):
animal = row['animal'] if row['animal'] in ANIMALS else guess_animal(row['name'])
cut = row['cut'] or guess_cut(row['name'])
if animal != row['animal'] or cut != row['cut']:
db.execute('UPDATE products SET animal=?, cut=? WHERE id=?', (animal, cut, row['id']))
DB.chmod(0o600)
def password_hash(password, salt):
return hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), n=16384, r=8, p=1).hex()
def set_password(password):
if len(password) < 12:
raise ValueError('Mindestens 12 Zeichen verwenden.')
salt = secrets.token_hex(16)
with connect() as db:
db.execute('INSERT OR REPLACE INTO admin VALUES (1,?,?)', (salt, password_hash(password, salt)))
SESSIONS.clear()
class StockConflict(ValueError):
"""Gewünschte Packung ist weg. Nichts wurde reserviert."""
def __init__(self, conflicts, kept):
super().__init__('Der gewünschte Bestand ist inzwischen nicht mehr verfügbar. Bitte neu auswählen.')
self.conflicts = conflicts
self.kept = kept
def closest_alternatives(db, product, blocked_ids, limit=6):
if not product or not product['animal'] or not product['cut']:
return []
wanted = weight_grams(product['unit'] or '')
found = []
if product['stock'] > 0 and product['id'] not in blocked_ids:
found.append(dict(product))
for candidate in db.execute('SELECT * FROM products WHERE stock>0 AND animal=? AND cut=? AND id!=?', (product['animal'], product['cut'], product['id'])):
item = dict(candidate)
if item['id'] not in blocked_ids:
found.append(item)
def distance(item):
grams = weight_grams(item['unit'] or '')
if wanted is None or grams is None:
return (1, Decimal('999999'), item['id'])
return (0 if item['id'] == product['id'] else 1, abs(grams - wanted), item['id'])
found.sort(key=distance)
chosen = found[:limit]
blocked_ids.update(item['id'] for item in chosen)
return chosen
def reserve(name, contact, quantities):
name, contact = name.strip(), contact.strip()
if not 1 <= len(name) <= 120 or not 3 <= len(contact) <= 200:
raise ValueError('Bitte Name und E-Mail oder Telefonnummer angeben.')
chosen = {int(k): int(v) for k, v in quantities.items() if int(v) != 0}
if not chosen or any(v < 1 or v > 9999 for v in chosen.values()):
raise ValueError('Bitte mindestens einen verfügbaren Artikel auswählen.')
with connect() as db:
db.execute('BEGIN IMMEDIATE')
kept, missing = [], []
for pid, qty in chosen.items():
row = db.execute('SELECT * FROM products WHERE id=?', (pid,)).fetchone()
if row and row['stock'] >= qty:
kept.append((dict(row), qty))
else:
missing.append((dict(row) if row else None, qty))
if missing:
blocked = {item[0]['id'] for item in kept}
conflicts = []
for product, qty in missing:
conflicts.append({
'product': product,
'qty': qty,
'alternatives': closest_alternatives(db, product, blocked),
})
raise StockConflict(conflicts, kept)
ref = secrets.token_hex(6).upper()
cur = db.execute('INSERT INTO reservations(reference,name,contact) VALUES (?,?,?)', (ref, name, contact))
for product, qty in kept:
updated = db.execute('UPDATE products SET stock=stock-? WHERE id=? AND stock>=?', (qty, product['id'], qty))
if updated.rowcount != 1:
raise ValueError('Der gewünschte Bestand ist inzwischen nicht mehr verfügbar. Bitte neu auswählen.')
db.execute('INSERT INTO items VALUES (?,?,?)', (cur.lastrowid, product['id'], qty))
return ref
def change_status(rid, status):
if status not in ('offen', 'bestätigt', 'abgeholt', 'storniert'):
raise ValueError('Ungültiger Status.')
with connect() as db:
db.execute('BEGIN IMMEDIATE')
old = db.execute('SELECT status FROM reservations WHERE id=?', (rid,)).fetchone()
if not old:
raise ValueError('Reservierung nicht gefunden.')
if old['status'] == status:
return
if old['status'] in ('abgeholt', 'storniert'):
raise ValueError('Abgeschlossene Reservierungen können nicht erneut geöffnet werden.')
if status == 'storniert':
for item in db.execute('SELECT * FROM items WHERE reservation_id=?', (rid,)):
db.execute('UPDATE products SET stock=stock+? WHERE id=?', (item['quantity'], item['product_id']))
db.execute('UPDATE reservations SET status=? WHERE id=?', (status, rid))
def delete_product(pid):
with connect() as db:
db.execute('BEGIN IMMEDIATE')
row = db.execute('SELECT image FROM products WHERE id=?', (pid,)).fetchone()
if not row:
raise ValueError('Artikel nicht gefunden.')
used = db.execute('SELECT count(*) FROM items WHERE product_id=?', (pid,)).fetchone()[0]
if used:
raise ValueError('Dieser Artikel steckt noch in Reservierungen. Zuerst die Anfragen löschen.')
db.execute('DELETE FROM products WHERE id=?', (pid,))
return row['image']
def delete_reservation(rid):
with connect() as db:
db.execute('BEGIN IMMEDIATE')
old = db.execute('SELECT status FROM reservations WHERE id=?', (rid,)).fetchone()
if not old:
raise ValueError('Reservierung nicht gefunden.')
if old['status'] in ('offen', 'bestätigt'):
for item in db.execute('SELECT * FROM items WHERE reservation_id=?', (rid,)):
db.execute('UPDATE products SET stock=stock+? WHERE id=?', (item['quantity'], item['product_id']))
db.execute('DELETE FROM items WHERE reservation_id=?', (rid,))
db.execute('DELETE FROM reservations WHERE id=?', (rid,))
def esc(value):
return html.escape(str(value), quote=True)
def guess_animal(name):
folded = name.casefold().replace('rowtwild', 'rotwild')
if 'kalb' in folded:
return 'Rotwild Kalb'
if 'wildschwein' in folded:
return 'Wildschwein'
if re.search(r'\breh\b', folded):
return 'Reh'
if 'hirsch' in folded or 'rotwild' in folded:
return 'Rotwild Hirsch'
return None
def guess_cut(name):
folded = name.casefold()
for cut, keys in CUT_KEYS:
if any(key in folded for key in keys):
return cut
return None
def catalog_selection(raw_path):
query = parse_qs(raw_path.split('?', 1)[1] if '?' in raw_path else '', max_num_fields=10)
tier = query.get('tier', [''])[0]
cut = query.get('cut', [''])[0]
return tier if tier in ANIMALS else '', cut if cut in CUTS else ''
def catalog_href(tier='', cut=''):
params = []
if tier:
params.append('tier=' + quote(tier))
if cut:
params.append('cut=' + quote(cut))
return u('/') + (('?' + '&'.join(params)) if params else '')
def choice_chip(label, href, selected, count):
mark = f'{count}' if count else 'nichts vorrätig'
if not count:
return f'{esc(label)}{mark}'
cls = 'chip on' if selected else 'chip'
return f'{esc(label)}{mark}'
def parse_price(value):
text = value.strip().replace('€', '').replace(' ', '')
if text.lower().endswith('/kg'):
text = text[:-3].strip()
text = text.replace(',', '.')
try:
amount = Decimal(text)
except InvalidOperation:
raise ValueError('Preis pro kg als Betrag angeben, zum Beispiel 18,50.')
if amount < 0 or amount > Decimal('9999.99') or amount != amount.quantize(Decimal('0.01')):
raise ValueError('Preis pro kg mit höchstens zwei Nachkommastellen angeben, zum Beispiel 18,50.')
return int((amount * 100).to_integral_value(ROUND_HALF_UP))
def format_price(cents):
if cents is None:
return ''
euros, rest = divmod(int(cents), 100)
return f'{euros},{rest:02d} € / kg'
WEIGHT = re.compile(r'(\d+(?:[.,]\d+)?)\s*(kg|g)\b', re.IGNORECASE)
def weight_grams(text):
if not text:
return None
match = WEIGHT.search(text)
if not match:
return None
amount = Decimal(match.group(1).replace(',', '.'))
grams = amount * 1000 if match.group(2).lower() == 'kg' else amount
if grams <= 0 or grams > Decimal('100000'):
return None
return grams
def piece_name(text):
head = re.split(r'\s+[aà]\s+', text.strip(), maxsplit=1, flags=re.IGNORECASE)[0]
head = WEIGHT.sub('', head).strip(' ,.-')
return head or 'Einheit'
DEFAULT_EXAMPLES = {
'Rücken': 'ruecken.jpg',
'Schulter': 'schulter.jpg',
'Keule': 'keule.jpg',
'Hals': 'hals.jpg',
'Hack': 'hack.jpg',
'Gulasch': 'ragout.jpg',
}
ANIMAL_EXAMPLES = {
('Wildschwein', 'Rücken'): 'wildschwein-ruecken.jpg',
('Wildschwein', 'Schulter'): 'wildschwein-schulter.jpg',
('Reh', 'Rücken'): 'reh-ruecken.jpg',
}
def default_example(animal, cut):
return ANIMAL_EXAMPLES.get((animal, cut)) or DEFAULT_EXAMPLES.get(cut)
def parse_pack_weights(text):
"""Eine ganze Grammzahl pro Zeile, wie die vorhandenen Einzelpackungen."""
packs = []
for raw in text.splitlines():
line = raw.strip()
if not line:
continue
if re.fullmatch(r'\d{2,5}', line):
grams = int(line)
else:
parsed = weight_grams(line)
if parsed is None or parsed != parsed.to_integral_value():
raise ValueError(f'Gewicht nicht erkannt: „{line}“. Eine ganze Grammzahl pro Zeile, zum Beispiel 740.')
grams = int(parsed)
if not 50 <= grams <= 20000:
raise ValueError(f'{grams} g liegt außerhalb von 50 g bis 20 kg.')
packs.append(grams)
if not packs:
raise ValueError('Mindestens ein Gewicht angeben, eine Zeile pro Packung.')
if len(packs) > 40:
raise ValueError('Höchstens 40 Packungen auf einmal.')
return packs
def piece_price(price_cents, unit, description=''):
"""Stückpreis aus Kilopreis und dem Gewicht in Einheit oder Beschreibung."""
grams = weight_grams(unit)
if grams is None:
grams = weight_grams(description)
if grams is None or price_cents is None:
return None
cents = int((Decimal(int(price_cents)) * grams / Decimal(1000)).quantize(Decimal('1'), rounding=ROUND_HALF_UP))
euros, rest = divmod(cents, 100)
return f'{euros},{rest:02d} € je {piece_name(unit)}'
def image_kind(data):
if data.startswith(b'\xff\xd8\xff'):
return 'jpg'
if data.startswith(b'\x89PNG\r\n\x1a\n'):
return 'png'
if data.startswith((b'GIF87a', b'GIF89a')):
return 'gif'
if len(data) >= 12 and data.startswith(b'RIFF') and data[8:12] == b'WEBP':
return 'webp'
return None
def drop_upload(key):
if not key or str(key).startswith('beispiel:'):
return
if not re.fullmatch(r'[0-9a-f]{32}\.(jpg|png|gif|webp)', key):
return
path = (UPLOADS / key).resolve()
if path.parent == UPLOADS.resolve() and path.is_file():
path.unlink()
def store_image(previous, choice, upload):
"""Neues Bild speichern. Gibt (schlüssel, zu löschendes altes Upload) zurück."""
if upload:
kind = image_kind(upload)
if not kind or not 32 <= len(upload) <= MAX_IMAGE:
raise ValueError('Bitte ein Bild als JPEG, PNG, WEBP oder GIF bis 2 MB verwenden.')
UPLOADS.mkdir(parents=True, exist_ok=True)
name = secrets.token_hex(16) + '.' + kind
path = UPLOADS / name
path.write_bytes(upload)
path.chmod(0o600)
return name, previous
if choice == 'clear':
return None, previous
if choice.startswith('beispiel:'):
fname = choice.split(':', 1)[1]
if fname not in EXAMPLES or not (EXAMPLES_DIR / fname).is_file():
raise ValueError('Beispielbild nicht gefunden.')
return 'beispiel:' + fname, previous
return previous, None
def image_path(key):
if not key:
return None
if str(key).startswith('beispiel:'):
name = key.split(':', 1)[1]
if name not in EXAMPLES:
return None
path = (EXAMPLES_DIR / name).resolve()
if path.parent != EXAMPLES_DIR.resolve() or not path.is_file():
return None
return path
if not re.fullmatch(r'[0-9a-f]{32}\.(jpg|png|gif|webp)', key):
return None
path = (UPLOADS / key).resolve()
if path.parent != UPLOADS.resolve() or not path.is_file():
return None
return path
def parse_multipart(body, content_type):
match = re.search(r'boundary=(?:"([^"]+)"|([^;\s]+))', content_type)
if not match:
raise ValueError('Ungültige Anfrage.')
marker = b'--' + (match.group(1) or match.group(2)).encode('ascii', 'strict')
fields, upload, pos, parts = {}, None, body.find(marker), 0
if pos < 0:
raise ValueError('Ungültige Anfrage.')
pos += len(marker)
while parts < 30:
if body[pos:pos + 2] == b'--':
break
if body[pos:pos + 2] != b'\r\n':
raise ValueError('Ungültige Anfrage.')
pos += 2
end = body.find(b'\r\n\r\n', pos)
nxt = body.find(b'\r\n' + marker, end + 4) if end >= 0 else -1
if end < 0 or nxt < 0:
raise ValueError('Ungültige Anfrage.')
header = body[pos:end].decode('latin-1', 'replace')
content = body[end + 4:nxt]
pos = nxt + 2 + len(marker)
parts += 1
name = re.search(r'name="([^"]+)"', header)
if not name or len(name.group(1)) > 40:
continue
filename = re.search(r'filename="([^"]*)"', header)
if filename:
if filename.group(1) and content:
upload = content
else:
fields[name.group(1)] = content.decode('utf-8')
return fields, upload
STYLE = '''*{box-sizing:border-box}body{margin:0;background:#f5f3eb;color:#21382a;font:17px system-ui,sans-serif}header,main,footer{max-width:1050px;margin:auto;padding:24px}header{display:flex;justify-content:space-between;align-items:center}a{color:#385c3b}h1{font: bold clamp(34px,6vw,62px) Georgia,serif;line-height:1.1;margin:20px 0}h2{font-family:Georgia,serif}p{line-height:1.6}.hero{padding:28px 0 40px;max-width:720px}.eyebrow{letter-spacing:.15em;text-transform:uppercase;font-size:13px}.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(260px,1fr));gap:18px}.card{background:white;padding:24px;border:1px solid #d5ddcf;border-radius:14px;margin-bottom:18px}label{display:block;margin:12px 0}input,textarea,select,button{font:inherit;padding:11px;border:1px solid #bcc8b7;border-radius:7px;width:100%}input[type=radio],input[type=checkbox]{width:auto}input[type=number]{max-width:160px}button{background:#294e35;color:white;cursor:pointer;margin-top:12px}.danger{background:#8a3b2a}nav{display:flex;gap:18px}.muted,small{color:#566555}.notice{padding:18px;background:#e4ecd9;border-radius:10px}.error{background:#f9dfd5}.row{display:flex;gap:20px;flex-wrap:wrap}.row>*{flex:1;min-width:200px}footer{font-size:14px;border-top:1px solid #d5ddcf}.cut{width:100%;height:190px;object-fit:cover;border-radius:10px;display:block;margin-bottom:12px}.picks{display:flex;flex-wrap:wrap;gap:12px}.picks label{width:148px;font-size:14px}.picks img{width:100%;height:96px;object-fit:cover;border-radius:8px}fieldset{border:0;padding:0;margin:12px 0}.filters{display:flex;flex-wrap:wrap;gap:10px;margin:8px 0 28px}.chip{display:inline-flex;align-items:center;gap:8px;padding:12px 16px;border:1px solid #d5ddcf;border-radius:999px;background:white;text-decoration:none;color:#21382a}.chip.on{background:#294e35;color:white}.chip small{color:#566555;font-size:14px}.chip.on small{color:#d5ddcf}.chip.off{opacity:.55}.crumbs{display:flex;gap:8px;align-items:center;margin-bottom:8px}.weight{font-size:28px;font-weight:700;margin:8px 0}.grid{grid-template-columns:repeat(auto-fill,minmax(260px,1fr))}.pack{display:flex;flex-direction:column}.pack button,.pack label{margin-top:auto}summary{cursor:pointer;font-weight:600}.grouphead{display:flex;gap:14px;align-items:center}.grouphead img{width:84px;height:84px;object-fit:cover;border-radius:10px;margin:0}.stockline{display:flex;gap:8px;align-items:center;flex-wrap:wrap;padding:8px 0;border-top:1px solid #e4eadc}.stockline form{display:flex;gap:8px;align-items:center;margin:0;flex-wrap:wrap}.stockline input,.stockline button,.stockline a,.stockline label{width:auto;margin:0}.stockline label{display:flex;align-items:center;gap:6px}.stockline input[name=unit]{width:220px}.stockline input[name=price],.stockline input[name=stock]{width:110px}textarea{min-height:140px}'''
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass # Keine Kontaktdaten oder Formulare protokollieren.
def session(self):
jar = cookies.SimpleCookie()
try:
jar.load(self.headers.get('Cookie', ''))
sid = jar['session'].value if 'session' in jar else ''
except cookies.CookieError:
sid = ''
now = time.time()
for key in list(SESSIONS):
if SESSIONS[key]['expires'] < now:
SESSIONS.pop(key, None)
session = SESSIONS.get(sid)
if session is None:
sid = secrets.token_urlsafe(32)
session = {'csrf': secrets.token_urlsafe(32), 'admin': False, 'expires': now + 3600}
SESSIONS[sid] = session
self.sid, self.sess = sid, session
def cookie(self):
forwarded = self.headers.get('X-Forwarded-Proto', '').split(',')[0].strip().lower()
secure = '; Secure' if forwarded == 'https' else ''
return f'session={self.sid}; HttpOnly; SameSite=Strict; Path={BASE or "/"}; Max-Age=3600{secure}'
def route(self):
path = self.path.split('?', 1)[0]
if not BASE:
return path
if path == BASE or path == BASE + '/':
return '/'
if path.startswith(BASE + '/'):
return path[len(BASE):] or '/'
return None
def send_page(self, content, status=200):
body = (f'''Wildfleisch · Bestand & ReservierungWALD & WILD''' + content + '''''').encode()
self.send_response(status)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Content-Length', str(len(body)))
self.send_header('Set-Cookie', self.cookie())
self.send_header('Cache-Control', 'no-store')
self.send_header('X-Content-Type-Options', 'nosniff')
self.send_header('X-Frame-Options', 'DENY')
self.send_header('Content-Security-Policy', "default-src 'none'; img-src 'self'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'; base-uri 'none'")
self.end_headers()
self.wfile.write(body)
def token(self):
return f''
def redirect(self, path):
self.send_response(303)
self.send_header('Location', u(path))
self.send_header('Set-Cookie', self.cookie())
self.send_header('Content-Length', '0')
self.end_headers()
def send_bytes(self, data, mime):
self.send_response(200)
self.send_header('Content-Type', mime)
self.send_header('Content-Length', str(len(data)))
self.send_header('X-Content-Type-Options', 'nosniff')
self.send_header('Cache-Control', 'private, max-age=3600')
self.end_headers()
self.wfile.write(data)
def send_stored_image(self, key):
path = image_path(key)
if path is None:
self.send_page('
Bild nicht gefunden
', 404)
return
data = path.read_bytes()
kind = image_kind(data)
if not kind:
self.send_page('
Bild nicht gefunden
', 404)
return
self.send_bytes(data, IMAGE_TYPES[kind])
def do_GET(self):
self.session()
path = self.route()
if path and path.startswith('/beispiel/'):
name = path[len('/beispiel/'):]
self.send_stored_image('beispiel:' + name if name in EXAMPLES else None)
return
if path and path.startswith('/bild/'):
ident = path[len('/bild/'):]
row = None
if ident.isdigit():
with connect() as db:
row = db.execute('SELECT image FROM products WHERE id=?', (int(ident),)).fetchone()
self.send_stored_image(row['image'] if row else None)
return
if path == '/':
tier, cut = catalog_selection(self.path)
with connect() as db:
products = db.execute('SELECT * FROM products ORDER BY name').fetchall()
self.send_page(self.catalog_page(products, tier, cut))
elif path == '/admin':
self.admin_page()
else:
self.send_page(f'
Seite nicht gefunden
Zum Bestand', 404)
def catalog_page(self, products, tier, cut):
stocked = [p for p in products if p['stock'] > 0]
known = [p for p in stocked if p['animal'] in ANIMALS]
other = [p for p in stocked if p['animal'] not in ANIMALS]
hero = 'Wildfleisch aus dem aktuellen Bestand
Gutes Wild. Einfach reservieren.
Wähle die Tierart. Die Packungen erscheinen direkt, das Teilstück filterst du nur bei Bedarf.
'
if not known:
return hero + self.pack_form(other or products)
animals = ''.join(choice_chip(animal, catalog_href(animal), animal == tier, sum(1 for p in known if p['animal'] == animal)) for animal in ANIMALS)
if not tier:
extra = '
Weitere Packungen
' + self.pack_form(other) if other else ''
return hero + '
Tierart
' + animals + '
' + extra
chosen = [p for p in known if p['animal'] == tier]
chosen.sort(key=lambda p: (CUTS.index(p['cut']) if p['cut'] in CUTS else len(CUTS), p['unit'] or ''))
packs = [p for p in chosen if p['cut'] == cut] if cut else chosen
cuts = choice_chip('Alle', catalog_href(tier), not cut, len(chosen))
cuts += ''.join(choice_chip(name, catalog_href(tier, name), name == cut, sum(1 for p in chosen if p['cut'] == name)) for name in CUTS)
title = f'{tier} · {cut}' if cut else tier
crumbs = f'
'
return hero + body + self.pack_form(packs)
def pack_form(self, products):
cards = ''
for p in products:
if p['stock'] < 1:
continue
photo = f'' if p['image'] else ''
title = p['cut'] or p['name']
price = ''
if p['price_cents'] is not None:
price = f'
{esc(format_price(p["price_cents"]))}
'
piece = piece_price(p['price_cents'], p['unit'], p['description'])
if piece:
price += f'
{esc(piece)}
'
cards += f'{photo}
{esc(p["unit"])}
{esc(title)}
{esc(p["description"])}
{price}
{p["stock"]} verfügbar
'
if not cards:
return '
Aktuell ist noch kein Bestand eingestellt. Schau bald wieder vorbei.
'
return f''
def alternative_offer(self, conflict, name, contact):
back = u('/')
blocks = ''
choice = False
for item in conflict.conflicts:
product = item['product']
label = ' · '.join(part for part in ((product or {}).get('unit'), (product or {}).get('cut'), (product or {}).get('animal')) if part) or 'Die gewählte Packung'
if product and product['animal'] in ANIMALS:
back = catalog_href(product['animal'], product['cut'] if product['cut'] in CUTS else '')
if product and product['stock'] > 0:
text = f'Davon sind nur noch {product["stock"]} da, nicht die gewünschten {item["qty"]}. Nichts davon wurde reserviert.'
else:
text = 'Die Packung ist nicht mehr da und wurde nicht reserviert.'
blocks += f'
{esc(label)}
{text}
'
if item['alternatives']:
choice = True
blocks += '
Stattdessen noch verfügbar:
'
for alt in item['alternatives']:
piece = piece_price(alt['price_cents'], alt['unit'], alt['description']) if alt['price_cents'] is not None else ''
blocks += f'
'
def preset_select(self, name, label, options, current):
current = current or ''
opts = ''.join(f'' for option in options)
blank = ' selected' if current not in options else ''
return f''
def image_choices(self, current, creating):
content = 'Bild'
return content
def admin_page(self):
if not self.sess['admin']:
self.send_page(f'
Verwaltung
')
return
query = parse_qs(self.path.split('?', 1)[1] if '?' in self.path else '', max_num_fields=5)
edit_raw = query.get('edit', [''])[0]
edit_id = int(edit_raw) if edit_raw.isdigit() else None
with connect() as db:
products = db.execute('SELECT * FROM products').fetchall()
reservations = db.execute('SELECT * FROM reservations ORDER BY id DESC').fetchall()
items = db.execute('SELECT i.*,p.name,p.unit,p.animal,p.cut FROM items i JOIN products p ON p.id=i.product_id').fetchall()
products = sorted(products, key=lambda p: (
ANIMALS.index(p['animal']) if p['animal'] in ANIMALS else len(ANIMALS),
CUTS.index(p['cut']) if p['cut'] in CUTS else len(CUTS),
weight_grams(p['unit'] or '') or Decimal('999999'),
p['id'],
))
editing = next((p for p in products if p['id'] == edit_id), None)
content = f'
Verwaltung
'
if editing:
price_value = format_price(editing['price_cents']).replace(' € / kg', '') if editing['price_cents'] is not None else ''
content += f''
else:
content += f''
content += '
Bestand pflegen
Jede Zeile ist eine Packung. Bestand ist die noch freie Menge.
'
if not products:
content += '
Noch keine Packungen.
'
last = None
for p in products:
key = (p['animal'] or 'Ohne Tierart', p['cut'] or 'Ohne Teilstück')
if key != last:
if last:
content += ''
photo = f'' if p['image'] else ''
content += f'
{photo}
{esc(key[0])} · {esc(key[1])}
{esc(p["description"])}
'
last = key
price_value = format_price(p['price_cents']).replace(' € / kg', '') if p['price_cents'] is not None else ''
content += '
'
listed = False
for i in items:
if i['reservation_id'] != r['id']:
continue
listed = True
where = ' · '.join(part for part in (i['animal'], i['cut']) if part)
content += f'
{i["quantity"]}× {esc(i["unit"])}{" · " + esc(where) if where else ""} · {esc(i["name"])}
'
if not listed:
content += '
Keine Positionen.
'
content += '
'
if r['status'] not in ('abgeholt', 'storniert'):
content += f''
hint = 'Offene und bestätigte Anfragen geben den Bestand zurück.' if r['status'] in ('offen', 'bestätigt') else 'Der Bestand bleibt unverändert.'
content += f''
self.send_page(content)
def do_POST(self):
self.session()
path = self.route()
try:
length = int(self.headers.get('Content-Length', '0'))
ctype = self.headers.get('Content-Type', '')
multipart = 'multipart/form-data' in ctype
large_ok = multipart and path == '/product' and self.sess['admin']
if length < 1 or length > (2_100_000 if large_ok else 16384):
raise ValueError('Ungültige Anfragegröße.')
raw = self.rfile.read(length)
upload = None
if multipart:
if path != '/product':
raise ValueError('Ungültige Anfrage.')
data, upload = parse_multipart(raw, ctype)
get = lambda key: data.get(key, '')
else:
data = parse_qs(raw.decode('utf-8'), max_num_fields=100)
get = lambda key: data.get(key, [''])[0]
if not hmac.compare_digest(get('csrf'), self.sess['csrf']):
self.send_page(f'
Sicherheitsprüfung fehlgeschlagen
Seite erneut öffnen', 403)
return
if path == '/login':
address = self.client_address[0]
now = time.time()
failures = [t for t in ATTEMPTS.get(address, []) if t > now - 300]
if len(failures) >= 5:
self.send_page('
Bitte fünf Minuten warten
', 429)
return
with connect() as db:
admin = db.execute('SELECT * FROM admin WHERE id=1').fetchone()
if not admin or not hmac.compare_digest(password_hash(get('password'), admin['salt']), admin['hash']):
ATTEMPTS[address] = failures + [now]
raise ValueError('Anmeldung fehlgeschlagen. Admin-Passwort zunächst per CLI einrichten.')
ATTEMPTS.pop(address, None)
SESSIONS.pop(self.sid, None)
self.sid = secrets.token_urlsafe(32)
self.sess = {'csrf': secrets.token_urlsafe(32), 'admin': True, 'expires': now + 3600}
SESSIONS[self.sid] = self.sess
self.redirect('/admin')
elif path == '/reserve':
quantities = {}
for key, value in data.items():
if key.startswith('q_'):
quantities[key[2:]] = value[0] if isinstance(value, list) else value
try:
ref = reserve(get('name'), get('contact'), quantities)
except StockConflict as conflict:
self.send_page(self.alternative_offer(conflict, get('name'), get('contact')))
return
self.send_page(f'
Anfrage eingegangen.
Deine Reservierungsnummer: {ref}
Der Bestand ist für dich zurückgelegt. Eine Bestätigung und die Abstimmung zur Abholung erfolgen persönlich.
Zurück zum Bestand')
elif path in ('/product','/status','/logout','/product-delete','/reservation-delete'):
if not self.sess['admin']:
self.send_page('
Anmeldung erforderlich
', 403)
return
if path == '/logout':
SESSIONS.pop(self.sid, None)
self.session()
elif path == '/status':
change_status(int(get('id')), get('status'))
elif path == '/product-delete':
if get('confirm') != '1':
raise ValueError('Löschen bitte bestätigen.')
drop_upload(delete_product(int(get('id'))))
elif path == '/reservation-delete':
if get('confirm') != '1':
raise ValueError('Löschen bitte bestätigen.')
delete_reservation(int(get('id')))
else:
name, desc = get('name').strip(), get('description').strip()
animal, cut = get('animal').strip(), get('cut').strip()
price = parse_price(get('price'))
if animal not in ANIMALS or cut not in CUTS:
raise ValueError('Bitte Tierart und Teilstück aus der Liste wählen.')
if len(desc) > 500 or len(name) > 500:
raise ValueError('Artikelangaben prüfen.')
choice = get('image_choice') or 'keep'
if get('id'):
unit = get('unit').strip()
stock = int(get('stock'))
if not name or not desc or not unit or len(unit) > 500 or not 0 <= stock <= 999999:
raise ValueError('Artikelangaben prüfen.')
with connect() as db:
row = db.execute('SELECT image FROM products WHERE id=?', (int(get('id')),)).fetchone()
if not row:
raise ValueError('Artikel nicht gefunden.')
image, old = store_image(row['image'], choice, upload)
db.execute('UPDATE products SET name=?,description=?,unit=?,stock=?,price_cents=?,image=?,animal=?,cut=? WHERE id=?', (name,desc,unit,stock,price,image,animal,cut,int(get('id'))))
drop_upload(old if image != old else None)
else:
grams_list = parse_pack_weights(get('weights'))
if not desc:
raise ValueError('Bitte eine Beschreibung angeben.')
if not name:
name = f'{animal} {cut}'
if choice == 'auto' and not upload:
fname = default_example(animal, cut)
choice = 'beispiel:' + fname if fname else 'keep'
stored = []
try:
image, _ = store_image(None, choice, upload)
stored.append(image)
for _ in grams_list[1:]:
if upload:
copy, _ = store_image(None, 'keep', upload)
stored.append(copy)
else:
stored.append(image)
with connect() as db:
for grams, image_key in zip(grams_list, stored):
unit = f'Packung à {grams} g'
db.execute('INSERT INTO products(name,description,unit,stock,price_cents,image,animal,cut) VALUES (?,?,?,?,?,?,?,?)', (name, desc, unit, 1, price, image_key, animal, cut))
except Exception:
for key in stored:
drop_upload(key)
raise
self.redirect('/admin')
else:
self.send_page('