"""BAN YARO — Züchter-Datenexport (HTML + ODS)"""
import io
import zipfile
import logging
from datetime import date, datetime
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from database import db
from auth import get_current_user
router = APIRouter()
logger = logging.getLogger(__name__)
def _require_breeder(user=Depends(get_current_user)):
if user["rolle"] not in ("breeder", "admin"):
raise HTTPException(403, "Nur für Züchter.")
return user
# ------------------------------------------------------------------
# Datenabruf
# ------------------------------------------------------------------
def _collect_data(user_id: int) -> dict:
with db() as conn:
profile = conn.execute(
"SELECT bp.*, u.name AS user_name, u.email "
"FROM breeder_profiles bp JOIN users u ON u.id = bp.user_id "
"WHERE bp.user_id=?", (user_id,)
).fetchone()
if not profile:
return None
bid = profile["id"]
hunde = conn.execute(
"SELECT * FROM zucht_hunde WHERE breeder_id=? ORDER BY name", (bid,)
).fetchall()
hund_ids = [h["id"] for h in hunde]
def rows_for(table, id_col):
if not hund_ids:
return []
placeholders = ",".join("?" * len(hund_ids))
return conn.execute(
f"SELECT * FROM {table} WHERE {id_col} IN ({placeholders}) ORDER BY {id_col}",
hund_ids
).fetchall()
health = rows_for("dog_health_tests", "hund_id")
genetic = rows_for("dog_genetic_tests", "hund_id")
titles = rows_for("dog_titles", "hund_id")
litters = conn.execute(
"SELECT * FROM litters WHERE breeder_id=? ORDER BY created_at DESC", (bid,)
).fetchall()
litter_ids = [l["id"] for l in litters]
puppies, weights = [], []
if litter_ids:
placeholders = ",".join("?" * len(litter_ids))
puppies = conn.execute(
f"SELECT * FROM puppies WHERE wurf_id IN ({placeholders}) ORDER BY wurf_id",
litter_ids
).fetchall()
puppy_ids = [p["id"] for p in puppies]
if puppy_ids:
ph2 = ",".join("?" * len(puppy_ids))
weights = conn.execute(
f"SELECT * FROM puppy_weights WHERE welpe_id IN ({ph2}) ORDER BY welpe_id, gemessen_am",
puppy_ids
).fetchall()
# Stammbaum pro Hund (3 Generationen, flach)
def pedigree_flat(hund_id, depth=3):
if not hund_id or depth == 0:
return {}
row = conn.execute("SELECT id, name, zuchtbuchnummer, geburtsdatum FROM zucht_hunde WHERE id=?", (hund_id,)).fetchone()
if not row:
return {}
result = {"name": row["name"], "nr": row["zuchtbuchnummer"] or "", "geb": row["geburtsdatum"] or ""}
row2 = conn.execute("SELECT vater_id, mutter_id FROM zucht_hunde WHERE id=?", (hund_id,)).fetchone()
if row2:
result["vater"] = pedigree_flat(row2["vater_id"], depth - 1)
result["mutter"] = pedigree_flat(row2["mutter_id"], depth - 1)
return result
pedigrees = {h["id"]: pedigree_flat(h["id"]) for h in hunde}
return {
"profile": dict(profile),
"hunde": [dict(h) for h in hunde],
"health": [dict(r) for r in health],
"genetic": [dict(r) for r in genetic],
"titles": [dict(r) for r in titles],
"litters": [dict(l) for l in litters],
"puppies": [dict(p) for p in puppies],
"weights": [dict(w) for w in weights],
"pedigrees": pedigrees,
}
# ------------------------------------------------------------------
# HTML-Export
# ------------------------------------------------------------------
def _fmt_date(val):
if not val:
return "—"
try:
return datetime.strptime(str(val)[:10], "%Y-%m-%d").strftime("%d.%m.%Y")
except Exception:
return str(val)
def _health_badge(test_typ, ergebnis):
color = "#6B7280"
e = (ergebnis or "").upper()
if test_typ == "HD":
color = {"A1": "#16a34a", "A2": "#16a34a", "B1": "#86efac", "B2": "#86efac",
"C": "#eab308", "C1": "#eab308", "C2": "#eab308",
"D": "#f97316", "D1": "#f97316", "D2": "#f97316",
"E": "#dc2626", "E1": "#dc2626", "E2": "#dc2626"}.get(e, "#6B7280")
elif test_typ == "ED":
color = {"0": "#16a34a", "1": "#eab308", "2": "#f97316", "3": "#dc2626"}.get(e, "#6B7280")
elif ergebnis in ("clear", "frei"):
color = "#16a34a"
elif ergebnis == "carrier":
color = "#eab308"
elif ergebnis == "affected":
color = "#dc2626"
return f'{ergebnis}'
def _pedigree_node_html(node, gen):
if not node:
return '
Unbekannt
'
geb = f"*{node['geb'][:4]}" if node.get("geb") else ""
nr = node.get("nr", "")
bg = "#7c3aed" if gen == 1 else ("#ede9fe" if gen == 2 else ("#f5f3ff" if gen == 3 else "#faf9ff"))
col = "#fff" if gen == 1 else "#1f2937"
return (
f'
'
f'
{node["name"]}
'
+ (f'
{nr}
' if nr else "")
+ (f'
{geb}
' if geb else "")
+ "
"
)
def _pedigree_html(tree):
# 4-Gen horizontaler Stammbaum als Tabelle
def nodes(t, gen, max_gen=4):
if gen > max_gen:
return []
item = [{"node": t, "gen": gen}]
item += nodes(t.get("vater") if t else None, gen + 1, max_gen)
item += nodes(t.get("mutter") if t else None, gen + 1, max_gen)
return item
rows = 8 # 2^3 für 4 Generationen
cols = 4
def collect(node, gen, row_start, row_span):
if gen > cols:
return []
items = [{"node": node, "gen": gen, "row": row_start, "span": row_span}]
half = row_span // 2
items += collect(node.get("vater") if node else None, gen + 1, row_start, half)
items += collect(node.get("mutter") if node else None, gen + 1, row_start + half, half)
return items
cells = collect(tree, 1, 0, rows)
gen_labels = {1: "Proband", 2: "Eltern", 3: "Großeltern", 4: "Urgroßeltern"}
# Grid als Tabelle (row_height = 50px)
row_h = 52
total_h = rows * row_h
html = (
f'
'
f'
'
)
for c in cells:
node, gen, row, span = c["node"], c["gen"], c["row"], c["span"]
html += (
f'
'
+ _pedigree_node_html(node, gen)
+ "
"
)
html += "
"
return html
def _generate_html(data: dict) -> str:
p = data["profile"]
today = date.today().strftime("%d.%m.%Y")
hunde_html = ""
for h in data["hunde"]:
hid = h["id"]
h_health = [r for r in data["health"] if r["hund_id"] == hid]
h_genetic = [r for r in data["genetic"] if r["hund_id"] == hid]
h_titles = [r for r in data["titles"] if r["hund_id"] == hid]
tree = data["pedigrees"].get(hid, {})
health_rows = "".join(
f'
{r["test_typ"]}
{_health_badge(r["test_typ"], r["ergebnis"])}
'
f'
{_fmt_date(r["untersuch_am"])}
{r.get("labor","") or ""}
'
f'
{r.get("untersucher","") or ""}
'
for r in h_health
) or "
Keine Einträge
"
genetic_rows = "".join(
f'
{r["marker_name"]}
{_health_badge("DNA", r["ergebnis_klasse"])}
'
f'
{_fmt_date(r["getestet_am"])}
{r.get("labor","") or ""}
'
for r in h_genetic
) or "
Keine Einträge
"
title_rows = "".join(
f'
{r["titel_name"]}
{r["titel_typ"]}
'
f'
{_fmt_date(r["verliehen_am"])}
{r.get("ort","") or ""}
'
f'
{r.get("richter","") or ""}
'
for r in sorted(h_titles, key=lambda x: x["verliehen_am"] or "", reverse=True)
) or "