rene
1ff66a7083
PYDANTIC max_length (38 Routen, ~400 Field-Constraints): Schützt vor DoS durch Riesen-Payloads (10MB Thread-Titel etc.). Pragmatische Limits: - Titel/Name: 200 · Beschreibung/Body: 10000 · Notiz: 5000 - Email: 254 (RFC 5321) · URL: 500 · Slug/Kategorie: 100 - Hund-Name/Rasse: 80 · Hund-Bio: 2000 Top-betroffen: forum.py, diary.py, health.py, dogs.py, expenses.py, notes.py, auth.py, profile.py. Manuelle len()-Checks in profile, chat, ki entfernt (jetzt durch Field abgedeckt). PYTEST COVERAGE (+19 Tests, 37 grün + 1 xfail): - test_security.py: require_owner (Places GET/PATCH/DELETE mit Fremduser → 403), JWT-Blacklist (Logout invalidiert Token), Login-Lockout (5 Fehlversuche → 429 + Retry-After Header) - test_race.py: Invoice-Counter (20 parallele Threads, alle unique), Founder-Number (atomare Vergabe, voll bei 100) - test_validation.py: Forum-Titel 30k Zeichen → 422, Diary-Text 50k → 422 (verifiziert Pydantic max_length-Sweep) A11Y (Tap-Targets ≥44×44 + Dark-Mode-Kontrast): - #header-user-btn 36→44px, .header-back 40→44, .header-menu-btn 40→44 - dog-profile Wrapped-Slider Prev/Next 40→44 - forum-Lightbox Close 40→44 - --c-text-muted Light: #B0A090 (2.37:1 FAIL) → #7F6B58 (4.74:1 PASS) - --c-text-muted Dark: #806A58 (3.58:1 FAIL) → #A08878 (5.46:1 PASS) - Branding-Farben unangetastet
377 lines
15 KiB
Python
377 lines
15 KiB
Python
"""BAN YARO — Digitaler Hundepass"""
|
||
|
||
import io
|
||
import secrets
|
||
from datetime import date, datetime, timedelta
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
from typing import Optional
|
||
from database import db
|
||
from auth import get_current_user
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Schemas
|
||
# ------------------------------------------------------------------
|
||
class PassportMeta(BaseModel):
|
||
blutgruppe: Optional[str] = Field(None, max_length=50)
|
||
allergien: Optional[str] = Field(None, max_length=2000)
|
||
besonderheiten: Optional[str] = Field(None, max_length=2000)
|
||
|
||
|
||
class VaccinationCreate(BaseModel):
|
||
krankheit: str = Field(..., max_length=200)
|
||
datum: str = Field(..., max_length=32)
|
||
naechste: Optional[str] = Field(None, max_length=32)
|
||
tierarzt: Optional[str] = Field(None, max_length=200)
|
||
charge_nr: Optional[str] = Field(None, max_length=100)
|
||
|
||
|
||
class MedicationCreate(BaseModel):
|
||
name: str = Field(..., max_length=200)
|
||
dosierung: Optional[str] = Field(None, max_length=200)
|
||
von: Optional[str] = Field(None, max_length=32)
|
||
bis: Optional[str] = Field(None, max_length=32)
|
||
notiz: Optional[str] = Field(None, max_length=2000)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Hilfsfunktion: Eigentümer-Prüfung
|
||
# ------------------------------------------------------------------
|
||
def _get_own_dog(conn, dog_id: int, user_id: int):
|
||
dog = conn.execute(
|
||
"SELECT * FROM dogs WHERE id=? AND user_id=?", (dog_id, user_id)
|
||
).fetchone()
|
||
if not dog:
|
||
raise HTTPException(404, "Hund nicht gefunden.")
|
||
return dog
|
||
|
||
|
||
def _load_passport_data(conn, dog_id: int) -> dict:
|
||
dog = conn.execute("SELECT * FROM dogs WHERE id=?", (dog_id,)).fetchone()
|
||
if not dog:
|
||
raise HTTPException(404, "Hund nicht gefunden.")
|
||
|
||
meta = conn.execute(
|
||
"SELECT * FROM dog_passport_meta WHERE dog_id=?", (dog_id,)
|
||
).fetchone()
|
||
vaccinations = conn.execute(
|
||
"SELECT * FROM vaccinations WHERE dog_id=? ORDER BY datum DESC", (dog_id,)
|
||
).fetchall()
|
||
medications = conn.execute(
|
||
"SELECT * FROM medications WHERE dog_id=? ORDER BY von DESC, id DESC", (dog_id,)
|
||
).fetchall()
|
||
|
||
return {
|
||
"dog": dict(dog),
|
||
"meta": dict(meta) if meta else {},
|
||
"vaccinations": [dict(v) for v in vaccinations],
|
||
"medications": [dict(m) for m in medications],
|
||
}
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# GET /passport/{dog_id} — vollständige Passdaten
|
||
# ------------------------------------------------------------------
|
||
@router.get("/{dog_id}")
|
||
async def get_passport(dog_id: int, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
return _load_passport_data(conn, dog_id)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# PUT /passport/{dog_id}/meta
|
||
# ------------------------------------------------------------------
|
||
@router.put("/{dog_id}/meta")
|
||
async def update_meta(dog_id: int, data: PassportMeta, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
conn.execute("""
|
||
INSERT INTO dog_passport_meta (dog_id, blutgruppe, allergien, besonderheiten, updated_at)
|
||
VALUES (?, ?, ?, ?, datetime('now'))
|
||
ON CONFLICT(dog_id) DO UPDATE SET
|
||
blutgruppe = excluded.blutgruppe,
|
||
allergien = excluded.allergien,
|
||
besonderheiten = excluded.besonderheiten,
|
||
updated_at = excluded.updated_at
|
||
""", (dog_id, data.blutgruppe, data.allergien, data.besonderheiten))
|
||
return {"ok": True}
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# POST /passport/{dog_id}/vaccinations
|
||
# ------------------------------------------------------------------
|
||
@router.post("/{dog_id}/vaccinations")
|
||
async def add_vaccination(dog_id: int, data: VaccinationCreate, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
conn.execute("""
|
||
INSERT INTO vaccinations (dog_id, krankheit, datum, naechste, tierarzt, charge_nr)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
""", (dog_id, data.krankheit, data.datum, data.naechste, data.tierarzt, data.charge_nr))
|
||
row = conn.execute(
|
||
"SELECT * FROM vaccinations WHERE dog_id=? ORDER BY id DESC LIMIT 1", (dog_id,)
|
||
).fetchone()
|
||
return dict(row)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# DELETE /passport/{dog_id}/vaccinations/{vacc_id}
|
||
# ------------------------------------------------------------------
|
||
@router.delete("/{dog_id}/vaccinations/{vacc_id}", status_code=204)
|
||
async def delete_vaccination(dog_id: int, vacc_id: int, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
conn.execute(
|
||
"DELETE FROM vaccinations WHERE id=? AND dog_id=?", (vacc_id, dog_id)
|
||
)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# POST /passport/{dog_id}/medications
|
||
# ------------------------------------------------------------------
|
||
@router.post("/{dog_id}/medications")
|
||
async def add_medication(dog_id: int, data: MedicationCreate, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
conn.execute("""
|
||
INSERT INTO medications (dog_id, name, dosierung, von, bis, notiz)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
""", (dog_id, data.name, data.dosierung, data.von, data.bis, data.notiz))
|
||
row = conn.execute(
|
||
"SELECT * FROM medications WHERE dog_id=? ORDER BY id DESC LIMIT 1", (dog_id,)
|
||
).fetchone()
|
||
return dict(row)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# DELETE /passport/{dog_id}/medications/{med_id}
|
||
# ------------------------------------------------------------------
|
||
@router.delete("/{dog_id}/medications/{med_id}", status_code=204)
|
||
async def delete_medication(dog_id: int, med_id: int, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
conn.execute(
|
||
"DELETE FROM medications WHERE id=? AND dog_id=?", (med_id, dog_id)
|
||
)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# POST /passport/{dog_id}/share — Share-Token erstellen
|
||
# ------------------------------------------------------------------
|
||
@router.post("/{dog_id}/share")
|
||
async def create_share(dog_id: int, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
token = secrets.token_urlsafe(32)
|
||
valid_until = (date.today() + timedelta(days=30)).isoformat()
|
||
conn.execute("""
|
||
INSERT INTO passport_shares (dog_id, token, valid_until)
|
||
VALUES (?, ?, ?)
|
||
""", (dog_id, token, valid_until))
|
||
return {
|
||
"token": token,
|
||
"valid_until": valid_until,
|
||
"url": f"/pass/{token}",
|
||
}
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# GET /passport/share/{token} — öffentlicher Endpunkt (kein Auth)
|
||
# ------------------------------------------------------------------
|
||
@router.get("/share/{token}")
|
||
async def get_shared_passport(token: str):
|
||
with db() as conn:
|
||
share = conn.execute(
|
||
"SELECT * FROM passport_shares WHERE token=?", (token,)
|
||
).fetchone()
|
||
if not share:
|
||
raise HTTPException(404, "Link nicht gefunden.")
|
||
if share["valid_until"] < date.today().isoformat():
|
||
raise HTTPException(410, "Dieser Link ist abgelaufen.")
|
||
return _load_passport_data(conn, share["dog_id"])
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# GET /passport/{dog_id}/pdf — PDF generieren
|
||
# ------------------------------------------------------------------
|
||
@router.get("/{dog_id}/pdf")
|
||
async def download_pdf(dog_id: int, user=Depends(get_current_user)):
|
||
with db() as conn:
|
||
_get_own_dog(conn, dog_id, user["id"])
|
||
data = _load_passport_data(conn, dog_id)
|
||
|
||
pdf_bytes = _generate_pdf(data)
|
||
dog_name = data["dog"]["name"].replace(" ", "_")
|
||
filename = f"Hundepass_{dog_name}.pdf"
|
||
|
||
return StreamingResponse(
|
||
io.BytesIO(pdf_bytes),
|
||
media_type="application/pdf",
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# PDF-Generierung mit fpdf2
|
||
# ------------------------------------------------------------------
|
||
def _generate_pdf(data: dict) -> bytes:
|
||
try:
|
||
from fpdf import FPDF
|
||
except ImportError:
|
||
raise HTTPException(500, "PDF-Bibliothek nicht verfügbar. Bitte fpdf2 installieren.")
|
||
|
||
dog = data["dog"]
|
||
meta = data["meta"]
|
||
vaccs = data["vaccinations"]
|
||
meds = data["medications"]
|
||
|
||
# Datumsformatierung DE
|
||
def _fmt_date(d):
|
||
if not d:
|
||
return "–"
|
||
try:
|
||
return datetime.strptime(d[:10], "%Y-%m-%d").strftime("%d.%m.%Y")
|
||
except Exception:
|
||
return d
|
||
|
||
# Geschlecht
|
||
geschlecht_map = {"m": "Rüde", "w": "Hündin"}
|
||
|
||
pdf = FPDF()
|
||
pdf.set_auto_page_break(auto=True, margin=20)
|
||
pdf.add_page()
|
||
|
||
# ---- Header ----
|
||
pdf.set_fill_color(40, 167, 100) # Ban Yaro Grün
|
||
pdf.rect(0, 0, 210, 38, style="F")
|
||
|
||
pdf.set_text_color(255, 255, 255)
|
||
pdf.set_font("Helvetica", style="B", size=20)
|
||
pdf.set_y(8)
|
||
pdf.cell(0, 10, "Ban Yaro", align="C", ln=True)
|
||
pdf.set_font("Helvetica", size=11)
|
||
pdf.cell(0, 8, f"Digitaler Hundepass — {dog['name']}", align="C", ln=True)
|
||
pdf.set_font("Helvetica", size=8)
|
||
pdf.cell(0, 6, f"Erstellt am {date.today().strftime('%d.%m.%Y')}", align="C", ln=True)
|
||
|
||
pdf.set_text_color(30, 30, 30)
|
||
pdf.set_y(46)
|
||
|
||
# ---- Hundedaten ----
|
||
pdf.set_fill_color(245, 250, 247)
|
||
pdf.set_draw_color(200, 200, 200)
|
||
pdf.set_font("Helvetica", style="B", size=12)
|
||
pdf.set_fill_color(235, 247, 240)
|
||
pdf.cell(0, 8, " Hundeangaben", ln=True, fill=True, border="B")
|
||
pdf.ln(3)
|
||
|
||
def _info_row(label, value):
|
||
pdf.set_font("Helvetica", style="B", size=9)
|
||
pdf.cell(45, 6, label + ":", ln=False)
|
||
pdf.set_font("Helvetica", size=9)
|
||
pdf.cell(0, 6, str(value) if value else "–", ln=True)
|
||
|
||
_info_row("Name", dog["name"])
|
||
_info_row("Rasse", dog.get("rasse") or "–")
|
||
_info_row("Geburtstag", _fmt_date(dog.get("geburtstag")))
|
||
_info_row("Geschlecht", geschlecht_map.get(dog.get("geschlecht", ""), "–"))
|
||
_info_row("Chip-Nr.", dog.get("chip_nr") or "–")
|
||
if meta.get("blutgruppe"):
|
||
_info_row("Blutgruppe", meta["blutgruppe"])
|
||
|
||
pdf.ln(5)
|
||
|
||
# ---- Allergien & Besonderheiten ----
|
||
if meta.get("allergien") or meta.get("besonderheiten"):
|
||
pdf.set_font("Helvetica", style="B", size=12)
|
||
pdf.set_fill_color(235, 247, 240)
|
||
pdf.cell(0, 8, " Allergien & Besonderheiten", ln=True, fill=True, border="B")
|
||
pdf.ln(3)
|
||
if meta.get("allergien"):
|
||
pdf.set_font("Helvetica", style="B", size=9)
|
||
pdf.cell(45, 6, "Allergien:", ln=False)
|
||
pdf.set_font("Helvetica", size=9)
|
||
pdf.multi_cell(0, 6, meta["allergien"])
|
||
if meta.get("besonderheiten"):
|
||
pdf.set_font("Helvetica", style="B", size=9)
|
||
pdf.cell(45, 6, "Besonderheiten:", ln=False)
|
||
pdf.set_font("Helvetica", size=9)
|
||
pdf.multi_cell(0, 6, meta["besonderheiten"])
|
||
pdf.ln(5)
|
||
|
||
# ---- Impfungen ----
|
||
pdf.set_font("Helvetica", style="B", size=12)
|
||
pdf.set_fill_color(235, 247, 240)
|
||
pdf.cell(0, 8, " Impfungen", ln=True, fill=True, border="B")
|
||
pdf.ln(3)
|
||
|
||
if vaccs:
|
||
# Tabellen-Header
|
||
pdf.set_fill_color(220, 240, 228)
|
||
pdf.set_font("Helvetica", style="B", size=8)
|
||
pdf.cell(50, 6, "Krankheit", border=1, fill=True)
|
||
pdf.cell(25, 6, "Datum", border=1, fill=True)
|
||
pdf.cell(25, 6, "Nächste fällig", border=1, fill=True)
|
||
pdf.cell(55, 6, "Tierarzt", border=1, fill=True)
|
||
pdf.cell(35, 6, "Charge-Nr.", border=1, fill=True, ln=True)
|
||
|
||
pdf.set_font("Helvetica", size=8)
|
||
for i, v in enumerate(vaccs):
|
||
fill = (i % 2 == 0)
|
||
pdf.set_fill_color(248, 252, 250) if fill else pdf.set_fill_color(255, 255, 255)
|
||
pdf.cell(50, 6, (v["krankheit"] or "")[:28], border=1, fill=fill)
|
||
pdf.cell(25, 6, _fmt_date(v["datum"]), border=1, fill=fill)
|
||
pdf.cell(25, 6, _fmt_date(v["naechste"]), border=1, fill=fill)
|
||
pdf.cell(55, 6, (v["tierarzt"] or "–")[:32], border=1, fill=fill)
|
||
pdf.cell(35, 6, (v["charge_nr"] or "–")[:20], border=1, fill=fill, ln=True)
|
||
else:
|
||
pdf.set_font("Helvetica", style="I", size=9)
|
||
pdf.set_text_color(140, 140, 140)
|
||
pdf.cell(0, 6, "Keine Impfungen eingetragen.", ln=True)
|
||
pdf.set_text_color(30, 30, 30)
|
||
|
||
pdf.ln(5)
|
||
|
||
# ---- Medikamente ----
|
||
pdf.set_font("Helvetica", style="B", size=12)
|
||
pdf.set_fill_color(235, 247, 240)
|
||
pdf.cell(0, 8, " Medikamente", ln=True, fill=True, border="B")
|
||
pdf.ln(3)
|
||
|
||
if meds:
|
||
pdf.set_fill_color(220, 240, 228)
|
||
pdf.set_font("Helvetica", style="B", size=8)
|
||
pdf.cell(55, 6, "Medikament", border=1, fill=True)
|
||
pdf.cell(35, 6, "Dosierung", border=1, fill=True)
|
||
pdf.cell(25, 6, "Von", border=1, fill=True)
|
||
pdf.cell(25, 6, "Bis", border=1, fill=True)
|
||
pdf.cell(50, 6, "Notiz", border=1, fill=True, ln=True)
|
||
|
||
pdf.set_font("Helvetica", size=8)
|
||
for i, m in enumerate(meds):
|
||
fill = (i % 2 == 0)
|
||
pdf.set_fill_color(248, 252, 250) if fill else pdf.set_fill_color(255, 255, 255)
|
||
pdf.cell(55, 6, (m["name"] or "")[:32], border=1, fill=fill)
|
||
pdf.cell(35, 6, (m["dosierung"] or "–")[:22], border=1, fill=fill)
|
||
pdf.cell(25, 6, _fmt_date(m["von"]), border=1, fill=fill)
|
||
bis = _fmt_date(m["bis"]) if m["bis"] else "dauerhaft"
|
||
pdf.cell(25, 6, bis, border=1, fill=fill)
|
||
pdf.cell(50, 6, (m["notiz"] or "–")[:30], border=1, fill=fill, ln=True)
|
||
else:
|
||
pdf.set_font("Helvetica", style="I", size=9)
|
||
pdf.set_text_color(140, 140, 140)
|
||
pdf.cell(0, 6, "Keine Medikamente eingetragen.", ln=True)
|
||
pdf.set_text_color(30, 30, 30)
|
||
|
||
# ---- Footer ----
|
||
pdf.set_y(-15)
|
||
pdf.set_font("Helvetica", style="I", size=8)
|
||
pdf.set_text_color(140, 140, 140)
|
||
pdf.cell(0, 5, "Erstellt mit Ban Yaro — banyaro.app", align="C", ln=True)
|
||
|
||
return bytes(pdf.output())
|