307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""BAN YARO — Läufigkeit, Progesterontests & Trächtigkeit (Züchter)"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
from datetime import date, timedelta
|
|
|
|
from database import db
|
|
from auth import get_current_user
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _require_breeder(user=Depends(get_current_user)):
|
|
if user["rolle"] not in ("breeder", "admin"):
|
|
raise HTTPException(403, "Nur für verifizierte Züchter.")
|
|
return user
|
|
|
|
|
|
def _check_hund_owner(hund_id: int, user: dict, conn):
|
|
row = conn.execute(
|
|
"""SELECT zh.id, bp.user_id AS owner_user_id FROM zucht_hunde zh
|
|
JOIN breeder_profiles bp ON bp.id = zh.breeder_id
|
|
WHERE zh.id=?""", (hund_id,)
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Hund nicht gefunden.")
|
|
if user["rolle"] != "admin" and row["owner_user_id"] != user["id"]:
|
|
raise HTTPException(403, "Kein Zugriff.")
|
|
return row
|
|
|
|
|
|
def _check_laeufi_owner(laeufi_id: int, user: dict, conn):
|
|
row = conn.execute(
|
|
"""SELECT l.id, bp.user_id AS owner_user_id FROM laeufi_log l
|
|
JOIN zucht_hunde zh ON zh.id = l.hund_id
|
|
JOIN breeder_profiles bp ON bp.id = zh.breeder_id
|
|
WHERE l.id=?""", (laeufi_id,)
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Läufigkeit nicht gefunden.")
|
|
if user["rolle"] != "admin" and row["owner_user_id"] != user["id"]:
|
|
raise HTTPException(403, "Kein Zugriff.")
|
|
return row
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Meilensteine berechnen
|
|
# ------------------------------------------------------------------
|
|
_MEILENSTEINE = [
|
|
(21, "Frühester Ultraschall möglich"),
|
|
(25, "Ultraschall empfohlen (Welpen erkennbar)"),
|
|
(35, "Bauch wird sichtbar"),
|
|
(45, "Röntgen möglich (Skelettanzahl)"),
|
|
(56, "Wurfbox aufstellen"),
|
|
(58, "Tägliche Temperaturmessung beginnen"),
|
|
(63, "Erwarteter Geburtstermin"),
|
|
(65, "Tierarzt konsultieren wenn keine Geburt"),
|
|
]
|
|
|
|
def _calc_meilensteine(deckdatum_str: str) -> list:
|
|
try:
|
|
deck = date.fromisoformat(deckdatum_str)
|
|
except Exception:
|
|
return []
|
|
return [
|
|
{
|
|
"tag": tag,
|
|
"datum": (deck + timedelta(days=tag)).isoformat(),
|
|
"label": label,
|
|
"vorbei": (deck + timedelta(days=tag)) < date.today(),
|
|
}
|
|
for tag, label in _MEILENSTEINE
|
|
]
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Schemas
|
|
# ------------------------------------------------------------------
|
|
class LaeufiCreate(BaseModel):
|
|
beginn: str
|
|
ende: Optional[str] = None
|
|
notiz: Optional[str] = None
|
|
|
|
class LaeufiUpdate(BaseModel):
|
|
beginn: Optional[str] = None
|
|
ende: Optional[str] = None
|
|
notiz: Optional[str] = None
|
|
|
|
class ProgestCreate(BaseModel):
|
|
datum: str
|
|
wert: Optional[float] = None
|
|
einheit: str = "ng/ml"
|
|
labor: Optional[str] = None
|
|
notiz: Optional[str] = None
|
|
|
|
class ProgestUpdate(BaseModel):
|
|
datum: Optional[str] = None
|
|
wert: Optional[float] = None
|
|
einheit: Optional[str] = None
|
|
labor: Optional[str] = None
|
|
notiz: Optional[str] = None
|
|
|
|
class DeckCreate(BaseModel):
|
|
deckdatum: str
|
|
laeufi_id: Optional[int] = None
|
|
ruede_id: Optional[int] = None
|
|
ruede_name: Optional[str] = None
|
|
deckart: str = "natuerlich"
|
|
traechtig: int = 0
|
|
ultraschall_datum: Optional[str] = None
|
|
notiz: Optional[str] = None
|
|
|
|
class DeckUpdate(BaseModel):
|
|
deckdatum: Optional[str] = None
|
|
ruede_id: Optional[int] = None
|
|
ruede_name: Optional[str] = None
|
|
deckart: Optional[str] = None
|
|
traechtig: Optional[int] = None
|
|
ultraschall_datum: Optional[str] = None
|
|
notiz: Optional[str] = None
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Läufigkeit
|
|
# ------------------------------------------------------------------
|
|
@router.get("/laeufi/{hund_id}")
|
|
async def list_laeufi(hund_id: int, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_hund_owner(hund_id, user, conn)
|
|
rows = conn.execute(
|
|
"SELECT * FROM laeufi_log WHERE hund_id=? ORDER BY beginn DESC",
|
|
(hund_id,)
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
@router.post("/laeufi/{hund_id}", status_code=201)
|
|
async def add_laeufi(hund_id: int, body: LaeufiCreate, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_hund_owner(hund_id, user, conn)
|
|
cur = conn.execute(
|
|
"INSERT INTO laeufi_log (hund_id, beginn, ende, notiz) VALUES (?,?,?,?)",
|
|
(hund_id, body.beginn, body.ende, body.notiz)
|
|
)
|
|
row = conn.execute("SELECT * FROM laeufi_log WHERE id=?", (cur.lastrowid,)).fetchone()
|
|
return dict(row)
|
|
|
|
|
|
@router.put("/laeufi/entry/{laeufi_id}")
|
|
async def update_laeufi(laeufi_id: int, body: LaeufiUpdate, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_laeufi_owner(laeufi_id, user, conn)
|
|
fields = {k: v for k, v in body.model_dump().items() if v is not None}
|
|
if fields:
|
|
sets = ", ".join(f"{k}=?" for k in fields)
|
|
conn.execute(f"UPDATE laeufi_log SET {sets} WHERE id=?", (*fields.values(), laeufi_id))
|
|
row = conn.execute("SELECT * FROM laeufi_log WHERE id=?", (laeufi_id,)).fetchone()
|
|
return dict(row)
|
|
|
|
|
|
@router.delete("/laeufi/entry/{laeufi_id}", status_code=204)
|
|
async def delete_laeufi(laeufi_id: int, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_laeufi_owner(laeufi_id, user, conn)
|
|
conn.execute("DELETE FROM laeufi_log WHERE id=?", (laeufi_id,))
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Progesterontests
|
|
# ------------------------------------------------------------------
|
|
@router.get("/laeufi/entry/{laeufi_id}/prog")
|
|
async def list_prog(laeufi_id: int, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_laeufi_owner(laeufi_id, user, conn)
|
|
rows = conn.execute(
|
|
"SELECT * FROM progesteron_tests WHERE laeufi_id=? ORDER BY datum",
|
|
(laeufi_id,)
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
@router.post("/laeufi/entry/{laeufi_id}/prog", status_code=201)
|
|
async def add_prog(laeufi_id: int, body: ProgestCreate, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_laeufi_owner(laeufi_id, user, conn)
|
|
cur = conn.execute(
|
|
"INSERT INTO progesteron_tests (laeufi_id, datum, wert, einheit, labor, notiz) VALUES (?,?,?,?,?,?)",
|
|
(laeufi_id, body.datum, body.wert, body.einheit, body.labor, body.notiz)
|
|
)
|
|
row = conn.execute("SELECT * FROM progesteron_tests WHERE id=?", (cur.lastrowid,)).fetchone()
|
|
return dict(row)
|
|
|
|
|
|
@router.put("/laeufi/prog/{prog_id}")
|
|
async def update_prog(prog_id: int, body: ProgestUpdate, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
pt = conn.execute(
|
|
"""SELECT pt.id, bp.user_id AS owner_user_id FROM progesteron_tests pt
|
|
JOIN laeufi_log l ON l.id = pt.laeufi_id
|
|
JOIN zucht_hunde zh ON zh.id = l.hund_id
|
|
JOIN breeder_profiles bp ON bp.id = zh.breeder_id
|
|
WHERE pt.id=?""", (prog_id,)
|
|
).fetchone()
|
|
if not pt:
|
|
raise HTTPException(404)
|
|
if user["rolle"] != "admin" and pt["owner_user_id"] != user["id"]:
|
|
raise HTTPException(403)
|
|
fields = {k: v for k, v in body.model_dump().items() if v is not None}
|
|
if fields:
|
|
sets = ", ".join(f"{k}=?" for k in fields)
|
|
conn.execute(f"UPDATE progesteron_tests SET {sets} WHERE id=?", (*fields.values(), prog_id))
|
|
row = conn.execute("SELECT * FROM progesteron_tests WHERE id=?", (prog_id,)).fetchone()
|
|
return dict(row)
|
|
|
|
|
|
@router.delete("/laeufi/prog/{prog_id}", status_code=204)
|
|
async def delete_prog(prog_id: int, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
pt = conn.execute(
|
|
"""SELECT pt.id, bp.user_id AS owner_user_id FROM progesteron_tests pt
|
|
JOIN laeufi_log l ON l.id = pt.laeufi_id
|
|
JOIN zucht_hunde zh ON zh.id = l.hund_id
|
|
JOIN breeder_profiles bp ON bp.id = zh.breeder_id
|
|
WHERE pt.id=?""", (prog_id,)
|
|
).fetchone()
|
|
if not pt:
|
|
raise HTTPException(404)
|
|
if user["rolle"] != "admin" and pt["owner_user_id"] != user["id"]:
|
|
raise HTTPException(403)
|
|
conn.execute("DELETE FROM progesteron_tests WHERE id=?", (prog_id,))
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Deckdaten & Trächtigkeit
|
|
# ------------------------------------------------------------------
|
|
@router.get("/laeufi/deck/{hund_id}")
|
|
async def list_deck(hund_id: int, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_hund_owner(hund_id, user, conn)
|
|
rows = conn.execute(
|
|
"SELECT * FROM deckdaten WHERE hund_id=? ORDER BY deckdatum DESC",
|
|
(hund_id,)
|
|
).fetchall()
|
|
result = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
d["meilensteine"] = _calc_meilensteine(d["deckdatum"])
|
|
result.append(d)
|
|
return result
|
|
|
|
|
|
@router.post("/laeufi/deck/{hund_id}", status_code=201)
|
|
async def add_deck(hund_id: int, body: DeckCreate, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
_check_hund_owner(hund_id, user, conn)
|
|
cur = conn.execute(
|
|
"""INSERT INTO deckdaten
|
|
(hund_id, laeufi_id, deckdatum, ruede_id, ruede_name, deckart,
|
|
traechtig, ultraschall_datum, notiz)
|
|
VALUES (?,?,?,?,?,?,?,?,?)""",
|
|
(hund_id, body.laeufi_id, body.deckdatum, body.ruede_id, body.ruede_name,
|
|
body.deckart, body.traechtig, body.ultraschall_datum, body.notiz)
|
|
)
|
|
row = conn.execute("SELECT * FROM deckdaten WHERE id=?", (cur.lastrowid,)).fetchone()
|
|
d = dict(row)
|
|
d["meilensteine"] = _calc_meilensteine(d["deckdatum"])
|
|
return d
|
|
|
|
|
|
@router.put("/laeufi/deck/entry/{deck_id}")
|
|
async def update_deck(deck_id: int, body: DeckUpdate, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
dk = conn.execute(
|
|
"""SELECT d.id, bp.user_id AS owner_user_id FROM deckdaten d
|
|
JOIN zucht_hunde zh ON zh.id = d.hund_id
|
|
JOIN breeder_profiles bp ON bp.id = zh.breeder_id
|
|
WHERE d.id=?""", (deck_id,)
|
|
).fetchone()
|
|
if not dk:
|
|
raise HTTPException(404)
|
|
if user["rolle"] != "admin" and dk["owner_user_id"] != user["id"]:
|
|
raise HTTPException(403)
|
|
fields = {k: v for k, v in body.model_dump().items() if v is not None}
|
|
if fields:
|
|
sets = ", ".join(f"{k}=?" for k in fields)
|
|
conn.execute(f"UPDATE deckdaten SET {sets} WHERE id=?", (*fields.values(), deck_id))
|
|
row = conn.execute("SELECT * FROM deckdaten WHERE id=?", (deck_id,)).fetchone()
|
|
d = dict(row)
|
|
d["meilensteine"] = _calc_meilensteine(d["deckdatum"])
|
|
return d
|
|
|
|
|
|
@router.delete("/laeufi/deck/entry/{deck_id}", status_code=204)
|
|
async def delete_deck(deck_id: int, user=Depends(_require_breeder)):
|
|
with db() as conn:
|
|
dk = conn.execute(
|
|
"""SELECT d.id, bp.user_id AS owner_user_id FROM deckdaten d
|
|
JOIN zucht_hunde zh ON zh.id = d.hund_id
|
|
JOIN breeder_profiles bp ON bp.id = zh.breeder_id
|
|
WHERE d.id=?""", (deck_id,)
|
|
).fetchone()
|
|
if not dk:
|
|
raise HTTPException(404)
|
|
if user["rolle"] != "admin" and dk["owner_user_id"] != user["id"]:
|
|
raise HTTPException(403)
|
|
conn.execute("DELETE FROM deckdaten WHERE id=?", (deck_id,))
|