524 lines
19 KiB
Python
524 lines
19 KiB
Python
"""BAN YARO — Hunde-Profil Routes"""
|
|
|
|
import os
|
|
import uuid
|
|
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
|
|
from pydantic import BaseModel
|
|
from typing import Optional
|
|
from database import db
|
|
from auth import get_current_user
|
|
from routes.push import send_push_to_user
|
|
from media_utils import safe_media_path, preview_url_from
|
|
|
|
router = APIRouter()
|
|
MEDIA_DIR = os.getenv("MEDIA_DIR", "/data/media")
|
|
|
|
|
|
class DogCreate(BaseModel):
|
|
name: str
|
|
rasse: Optional[str] = None
|
|
geburtstag: Optional[str] = None
|
|
geschlecht: Optional[str] = None
|
|
gewicht_kg: Optional[float] = None
|
|
chip_nr: Optional[str] = None
|
|
bio: Optional[str] = None
|
|
is_public: bool = False
|
|
|
|
|
|
class DogUpdate(BaseModel):
|
|
name: Optional[str] = None
|
|
rasse: Optional[str] = None
|
|
rasse_id: Optional[int] = None
|
|
geburtstag: Optional[str] = None
|
|
geschlecht: Optional[str] = None
|
|
gewicht_kg: Optional[float] = None
|
|
chip_nr: Optional[str] = None
|
|
bio: Optional[str] = None
|
|
is_public: Optional[bool] = None
|
|
|
|
|
|
@router.get("")
|
|
async def list_dogs(user=Depends(get_current_user)):
|
|
with db() as conn:
|
|
own = conn.execute(
|
|
"SELECT *, NULL AS shared_by, NULL AS share_role FROM dogs WHERE user_id=? ORDER BY id",
|
|
(user["id"],)
|
|
).fetchall()
|
|
shared = conn.execute(
|
|
"""SELECT d.*, u.name AS shared_by, ds.role AS share_role
|
|
FROM dog_shares ds
|
|
JOIN dogs d ON d.id = ds.dog_id
|
|
JOIN users u ON u.id = ds.owner_id
|
|
WHERE ds.shared_with_id = ? AND ds.accepted_at IS NOT NULL""",
|
|
(user["id"],)
|
|
).fetchall()
|
|
guest_rows = conn.execute("""
|
|
SELECT d.*, ss.id AS sub_id, ss.valid_until AS sitting_until,
|
|
u.name AS owner_name, NULL AS shared_by, NULL AS share_role
|
|
FROM sitting_subscriptions ss
|
|
JOIN dogs d ON d.id = ss.dog_id
|
|
JOIN users u ON u.id = ss.owner_id
|
|
WHERE ss.sitter_id = ?
|
|
AND ss.valid_until >= date('now')
|
|
""", (user["id"],)).fetchall()
|
|
|
|
result = []
|
|
for r in own:
|
|
d = dict(r)
|
|
d["is_guest"] = False
|
|
result.append(d)
|
|
for r in shared:
|
|
d = dict(r)
|
|
d["is_guest"] = False
|
|
result.append(d)
|
|
for r in guest_rows:
|
|
d = dict(r)
|
|
d["is_guest"] = True
|
|
result.append(d)
|
|
return result
|
|
|
|
|
|
@router.post("")
|
|
async def create_dog(data: DogCreate, user=Depends(get_current_user)):
|
|
with db() as conn:
|
|
conn.execute(
|
|
"""INSERT INTO dogs (user_id, name, rasse, geburtstag, geschlecht,
|
|
gewicht_kg, chip_nr, bio, is_public)
|
|
VALUES (?,?,?,?,?,?,?,?,?)""",
|
|
(user["id"], data.name, data.rasse, data.geburtstag,
|
|
data.geschlecht, data.gewicht_kg, data.chip_nr,
|
|
data.bio, int(data.is_public))
|
|
)
|
|
dog = conn.execute(
|
|
"SELECT * FROM dogs WHERE user_id=? ORDER BY id DESC LIMIT 1",
|
|
(user["id"],)
|
|
).fetchone()
|
|
return dict(dog)
|
|
|
|
|
|
@router.get("/{dog_id}/welcome-dashboard")
|
|
async def get_welcome_dashboard(dog_id: int, user=Depends(get_current_user)):
|
|
"""Liefert kompakte Dashboard-Daten für die Welcome-Ansicht eines Hundes."""
|
|
import random as _random
|
|
with db() as conn:
|
|
# Besitz prüfen
|
|
dog = conn.execute(
|
|
"SELECT id FROM dogs WHERE id=? AND user_id=?", (dog_id, user["id"])
|
|
).fetchone()
|
|
if not dog:
|
|
raise HTTPException(404, "Hund nicht gefunden.")
|
|
|
|
# Zufälliges Foto aus den letzten 100 Tagebuchbildern
|
|
photos = conn.execute(
|
|
"""SELECT dm.url FROM diary_media dm
|
|
JOIN diary d ON d.id = dm.diary_id
|
|
WHERE d.dog_id=? AND dm.media_type='image'
|
|
ORDER BY d.datum DESC LIMIT 100""",
|
|
(dog_id,)
|
|
).fetchall()
|
|
random_photo = None
|
|
if photos:
|
|
chosen_url = _random.choice(photos)["url"]
|
|
random_photo = {
|
|
"url": chosen_url,
|
|
"preview_url": preview_url_from(chosen_url),
|
|
}
|
|
|
|
# Neuester Tagebucheintrag
|
|
last_diary_row = conn.execute(
|
|
"SELECT titel, datum FROM diary WHERE dog_id=? ORDER BY datum DESC LIMIT 1",
|
|
(dog_id,)
|
|
).fetchone()
|
|
last_diary = dict(last_diary_row) if last_diary_row else None
|
|
|
|
# Nächster Termin (kein Gewicht, nur innerhalb 60 Tage)
|
|
next_appt_row = conn.execute(
|
|
"""SELECT bezeichnung, naechstes, typ FROM health
|
|
WHERE dog_id=? AND naechstes IS NOT NULL
|
|
AND naechstes >= date('now')
|
|
AND naechstes <= date('now', '+60 days')
|
|
AND typ != 'gewicht'
|
|
ORDER BY naechstes ASC LIMIT 1""",
|
|
(dog_id,)
|
|
).fetchone()
|
|
next_appointment = dict(next_appt_row) if next_appt_row else None
|
|
|
|
# Letztes Gewicht
|
|
last_weight_row = conn.execute(
|
|
"""SELECT wert, einheit, datum FROM health
|
|
WHERE dog_id=? AND typ='gewicht'
|
|
ORDER BY datum DESC LIMIT 1""",
|
|
(dog_id,)
|
|
).fetchone()
|
|
last_weight = dict(last_weight_row) if last_weight_row else None
|
|
|
|
# Anzahl Tagebucheinträge
|
|
diary_count = conn.execute(
|
|
"SELECT COUNT(*) AS n FROM diary WHERE dog_id=?", (dog_id,)
|
|
).fetchone()["n"]
|
|
|
|
# Tagesübung — personalisiert aus exercise_progress, tagesstabil
|
|
import datetime as _dt
|
|
day_num = (_dt.date.today() - _dt.date(2024, 1, 1)).days
|
|
|
|
# Übungen in Bearbeitung (noch-nicht / manchmal / meistens), älteste zuerst
|
|
in_progress = conn.execute(
|
|
"""SELECT ep.exercise_id, te.name, te.kategorie, te.schwierigkeit
|
|
FROM exercise_progress ep
|
|
JOIN training_exercises te ON te.exercise_id = ep.exercise_id
|
|
WHERE ep.user_id = ? AND ep.status IN ('noch-nicht', 'manchmal', 'meistens')
|
|
ORDER BY ep.updated_at ASC LIMIT 20""",
|
|
(user["id"],)
|
|
).fetchall()
|
|
|
|
daily_exercise = None
|
|
if in_progress:
|
|
daily_exercise = dict(in_progress[day_num % len(in_progress)])
|
|
else:
|
|
# Fallback: globale Rotation über alle Übungen
|
|
exercise_count = conn.execute(
|
|
"SELECT COUNT(*) AS n FROM training_exercises"
|
|
).fetchone()["n"]
|
|
if exercise_count:
|
|
ex_row = conn.execute(
|
|
"SELECT exercise_id, name, kategorie, schwierigkeit FROM training_exercises ORDER BY id LIMIT 1 OFFSET ?",
|
|
(day_num % exercise_count,)
|
|
).fetchone()
|
|
daily_exercise = dict(ex_row) if ex_row else None
|
|
|
|
return {
|
|
"random_photo": random_photo,
|
|
"last_diary": last_diary,
|
|
"next_appointment": next_appointment,
|
|
"last_weight": last_weight,
|
|
"diary_count": diary_count,
|
|
"daily_exercise": daily_exercise,
|
|
}
|
|
|
|
|
|
@router.get("/{dog_id}")
|
|
async def get_dog(dog_id: int, user=Depends(get_current_user)):
|
|
with db() as conn:
|
|
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 dict(dog)
|
|
|
|
|
|
@router.patch("/{dog_id}")
|
|
async def update_dog(dog_id: int, data: DogUpdate, user=Depends(get_current_user)):
|
|
fields = {k: v for k, v in data.model_dump().items() if v is not None}
|
|
if not fields:
|
|
raise HTTPException(400, "Keine Änderungen angegeben.")
|
|
|
|
set_clause = ", ".join(f"{k}=?" for k in fields)
|
|
values = list(fields.values()) + [dog_id, user["id"]]
|
|
|
|
with db() as conn:
|
|
conn.execute(
|
|
f"UPDATE dogs SET {set_clause} WHERE id=? AND user_id=?", values
|
|
)
|
|
dog = conn.execute(
|
|
"SELECT * FROM dogs WHERE id=?", (dog_id,)
|
|
).fetchone()
|
|
return dict(dog)
|
|
|
|
|
|
@router.delete("/{dog_id}", status_code=204)
|
|
async def delete_dog(dog_id: int, user=Depends(get_current_user)):
|
|
with db() as conn:
|
|
conn.execute(
|
|
"DELETE FROM dogs WHERE id=? AND user_id=?", (dog_id, user["id"])
|
|
)
|
|
|
|
|
|
@router.post("/{dog_id}/photo")
|
|
async def upload_photo(
|
|
dog_id: int,
|
|
file: UploadFile = File(...),
|
|
user=Depends(get_current_user)
|
|
):
|
|
# Hund gehört dem User?
|
|
with db() as conn:
|
|
dog = conn.execute(
|
|
"SELECT id FROM dogs WHERE id=? AND user_id=?", (dog_id, user["id"])
|
|
).fetchone()
|
|
if not dog:
|
|
raise HTTPException(404, "Hund nicht gefunden.")
|
|
|
|
# Datei immer als JPEG speichern (HEIC/PNG/WebP → kompatibel für alle Browser)
|
|
import io
|
|
from PIL import Image
|
|
try:
|
|
import pillow_heif
|
|
pillow_heif.register_heif_opener()
|
|
except ImportError:
|
|
pass
|
|
|
|
content = await file.read()
|
|
try:
|
|
from PIL import ImageOps
|
|
img = Image.open(io.BytesIO(content))
|
|
img = ImageOps.exif_transpose(img) # EXIF-Orientierung anwenden
|
|
img = img.convert("RGB")
|
|
buf = io.BytesIO()
|
|
img.save(buf, format="JPEG", quality=90)
|
|
content = buf.getvalue()
|
|
except Exception:
|
|
pass # Fallback: Originaldaten speichern
|
|
|
|
filename = f"dog_{dog_id}_{uuid.uuid4().hex[:8]}.jpg"
|
|
path = os.path.join(MEDIA_DIR, "dogs", filename)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
|
|
with open(path, "wb") as f:
|
|
f.write(content)
|
|
|
|
foto_url = f"/media/dogs/{filename}"
|
|
with db() as conn:
|
|
conn.execute("UPDATE dogs SET foto_url=? WHERE id=?", (foto_url, dog_id))
|
|
|
|
return {"foto_url": foto_url}
|
|
|
|
|
|
class PhotoPosition(BaseModel):
|
|
zoom: float = 1.0
|
|
offset_x: float = 0.0
|
|
offset_y: float = 0.0
|
|
|
|
|
|
@router.patch("/{dog_id}/photo-position")
|
|
async def update_photo_position(dog_id: int, pos: PhotoPosition, user=Depends(get_current_user)):
|
|
with db() as conn:
|
|
updated = conn.execute(
|
|
"UPDATE dogs SET foto_zoom=?, foto_offset_x=?, foto_offset_y=? WHERE id=? AND user_id=?",
|
|
(pos.zoom, pos.offset_x, pos.offset_y, dog_id, user["id"])
|
|
).rowcount
|
|
if not updated:
|
|
raise HTTPException(404, "Hund nicht gefunden.")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.delete("/{dog_id}/photo", status_code=204)
|
|
async def delete_photo(dog_id: int, user=Depends(get_current_user)):
|
|
with db() as conn:
|
|
row = conn.execute(
|
|
"SELECT foto_url FROM dogs WHERE id=? AND user_id=?", (dog_id, user["id"])
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Hund nicht gefunden.")
|
|
if row["foto_url"]:
|
|
path = safe_media_path(MEDIA_DIR, row["foto_url"])
|
|
if path and os.path.exists(path):
|
|
os.remove(path)
|
|
with db() as conn:
|
|
conn.execute(
|
|
"UPDATE dogs SET foto_url=NULL, foto_zoom=1.0, foto_offset_x=0.0, foto_offset_y=0.0 WHERE id=?",
|
|
(dog_id,)
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Fähigkeiten / Kommandos (für Profil + öffentliche Seite)
|
|
# ------------------------------------------------------------------
|
|
|
|
def _parse_exercise_name(exercise_id: str) -> str:
|
|
"""grundkommandos_Hier__Komm → 'Hier / Komm'"""
|
|
parts = exercise_id.split("_", 1)
|
|
if len(parts) < 2:
|
|
return exercise_id
|
|
return parts[1].replace("__", " / ").replace("_", " ")
|
|
|
|
|
|
def _load_skills(conn, dog_id: int, user_id: int) -> list:
|
|
"""Gibt Übungen mit Status 'sitzt' oder 'meistens' zurück, die mit diesem Hund trainiert wurden."""
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT ep.exercise_id, ep.status,
|
|
(SELECT ts.exercise_name FROM training_sessions ts
|
|
WHERE ts.user_id = ep.user_id AND ts.dog_id = ?
|
|
AND ts.exercise_id = ep.exercise_id
|
|
ORDER BY ts.datum DESC, ts.created_at DESC LIMIT 1) AS exercise_name
|
|
FROM exercise_progress ep
|
|
WHERE ep.user_id = ?
|
|
AND ep.status IN ('sitzt', 'meistens')
|
|
AND EXISTS (SELECT 1 FROM training_sessions ts2
|
|
WHERE ts2.user_id = ep.user_id AND ts2.dog_id = ?
|
|
AND ts2.exercise_id = ep.exercise_id)
|
|
ORDER BY ep.status DESC, ep.exercise_id
|
|
""",
|
|
(dog_id, user_id, dog_id)
|
|
).fetchall()
|
|
|
|
return [
|
|
{
|
|
"exercise_id": r["exercise_id"],
|
|
"exercise_name": r["exercise_name"] or _parse_exercise_name(r["exercise_id"]),
|
|
"status": r["status"],
|
|
"tab": r["exercise_id"].split("_")[0],
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
|
|
@router.get("/{dog_id}/skills")
|
|
async def get_dog_skills(dog_id: int, user=Depends(get_current_user)):
|
|
uid = user["id"]
|
|
with db() as conn:
|
|
dog = conn.execute(
|
|
"SELECT id, user_id FROM dogs WHERE id=? AND (user_id=? OR id IN (SELECT dog_id FROM sitting_subscriptions WHERE sitter_id=? AND valid_until >= date('now')))",
|
|
(dog_id, uid, uid)
|
|
).fetchone()
|
|
if not dog:
|
|
raise HTTPException(404, "Hund nicht gefunden.")
|
|
return _load_skills(conn, dog_id, dog["user_id"])
|
|
|
|
|
|
# Öffentliches Profil (für NFC-Tag, kein Login nötig)
|
|
@router.get("/public/{dog_id}")
|
|
async def public_dog_profile(dog_id: int):
|
|
with db() as conn:
|
|
dog = conn.execute(
|
|
"""SELECT d.id, d.name, d.rasse, d.geburtstag, d.foto_url, d.bio,
|
|
d.user_id, u.name as besitzer_name
|
|
FROM dogs d JOIN users u ON d.user_id=u.id
|
|
WHERE d.id=? AND d.is_public=1""",
|
|
(dog_id,)
|
|
).fetchone()
|
|
if not dog:
|
|
raise HTTPException(404, "Profil nicht gefunden oder nicht öffentlich.")
|
|
skills = _load_skills(conn, dog_id, dog["user_id"])
|
|
result = dict(dog)
|
|
result.pop("user_id", None)
|
|
result["skills"] = skills
|
|
return result
|
|
|
|
|
|
class FoundReport(BaseModel):
|
|
message: Optional[str] = None
|
|
kontakt: Optional[str] = None
|
|
|
|
|
|
# Gefunden-Meldung (kein Login nötig)
|
|
@router.post("/public/{dog_id}/found")
|
|
async def report_found(dog_id: int, data: FoundReport = FoundReport()):
|
|
with db() as conn:
|
|
row = conn.execute(
|
|
"""SELECT d.id, d.name, d.user_id
|
|
FROM dogs d
|
|
WHERE d.id=? AND d.is_public=1""",
|
|
(dog_id,)
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Profil nicht gefunden oder nicht öffentlich.")
|
|
|
|
dog_name = row["name"]
|
|
user_id = row["user_id"]
|
|
|
|
body = data.message.strip() if data.message and data.message.strip() \
|
|
else "Jemand hat deinen Hund gefunden. Öffne die App für Details."
|
|
|
|
if data.kontakt and data.kontakt.strip():
|
|
body += f" Kontakt: {data.kontakt.strip()}"
|
|
|
|
send_push_to_user(user_id, {
|
|
"title": f"🐾 {dog_name} wurde gefunden!",
|
|
"body": body,
|
|
"data": {"page": "diary", "found": True},
|
|
"tag": f"found-{dog_id}",
|
|
})
|
|
|
|
return {"ok": True}
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# GET /api/dogs/{id}/pflege — Pflegetipps für diesen Hund
|
|
# ------------------------------------------------------------------
|
|
@router.get("/{dog_id}/pflege")
|
|
async def get_pflege_tipps(dog_id: int, user=Depends(get_current_user)):
|
|
import json as _json
|
|
with db() as conn:
|
|
dog = conn.execute(
|
|
"SELECT id, name, rasse, rasse_id FROM dogs WHERE id=? AND user_id=?",
|
|
(dog_id, user["id"])
|
|
).fetchone()
|
|
if not dog:
|
|
raise HTTPException(404, "Hund nicht gefunden.")
|
|
|
|
# Rassen-Infos für Fell-Typ
|
|
rasse_info = None
|
|
with db() as conn:
|
|
if dog["rasse_id"]:
|
|
rasse_info = conn.execute(
|
|
"SELECT name, groesse, beschreibung FROM wiki_rassen WHERE id=?",
|
|
(dog["rasse_id"],)
|
|
).fetchone()
|
|
elif dog["rasse"]:
|
|
rasse_info = conn.execute(
|
|
"SELECT name, groesse, beschreibung FROM wiki_rassen WHERE name LIKE ? LIMIT 1",
|
|
(f"%{dog['rasse']}%",)
|
|
).fetchone()
|
|
|
|
# Fell-Typ und Pflegeart ableiten
|
|
fell_filter = None
|
|
fell_pflege_art_filter = None
|
|
if rasse_info:
|
|
beschr = (rasse_info["beschreibung"] or "").lower()
|
|
if any(w in beschr for w in ["lockig", "wellig", "kraus", "pudel", "doodle"]):
|
|
fell_filter = "lockig"
|
|
elif any(w in beschr for w in ["langhaar", "seidiges", "fließendes", "langes fell"]):
|
|
fell_filter = "lang"
|
|
elif any(w in beschr for w in ["kurzhaar", "kurzes fell", "glatthaarig"]):
|
|
fell_filter = "kurz"
|
|
elif rasse_info["groesse"] in ("gross", "sehr_gross"):
|
|
fell_filter = "doppel"
|
|
|
|
# Pflegeart: Trimmen vs. Schneiden
|
|
if any(w in beschr for w in ["trimm", "hand-stripping", "stripping", "rauhhaar", "drahthaar", "rauhaar"]):
|
|
fell_pflege_art_filter = "trimmen"
|
|
elif any(w in beschr for w in ["schneid", "geschoren", "schere", "clipper"]):
|
|
fell_pflege_art_filter = "schneiden"
|
|
|
|
with db() as conn:
|
|
alle_tipps = conn.execute(
|
|
"SELECT * FROM pflege_tipps ORDER BY kategorie, titel"
|
|
).fetchall()
|
|
|
|
# Relevante Tipps: kein Fell-Filter oder passend
|
|
from datetime import date
|
|
heute_saison = {1:"winter",2:"winter",3:"fruehling",4:"fruehling",5:"fruehling",
|
|
6:"sommer",7:"sommer",8:"sommer",9:"herbst",10:"herbst",
|
|
11:"herbst",12:"winter"}[date.today().month]
|
|
|
|
result = []
|
|
for t in alle_tipps:
|
|
t = dict(t)
|
|
# Fell-Typ-Filter
|
|
if fell_filter and t["fell_typ"] and t["fell_typ"] != "alle":
|
|
if fell_filter not in t["fell_typ"].split(","):
|
|
continue
|
|
# Pflegeart-Filter: Trimm-Tipps nicht bei Schneidehunden und umgekehrt
|
|
tipp_art = t.get("fell_pflege_art")
|
|
if tipp_art and tipp_art != "alle" and fell_pflege_art_filter:
|
|
if tipp_art != fell_pflege_art_filter:
|
|
continue
|
|
t["schritte"] = _json.loads(t["schritte"] or "[]")
|
|
t["saisonal_aktuell"] = bool(t["saison"] and heute_saison in t["saison"])
|
|
result.append(t)
|
|
|
|
# Tipp des Tages: erster aktuell-saisonaler oder zufällig deterministisch
|
|
from hashlib import md5
|
|
day_hash = int(md5(str(date.today()).encode()).hexdigest(), 16)
|
|
saisonal = [t for t in result if t["saisonal_aktuell"]]
|
|
tipp_des_tages = (saisonal or result)[day_hash % len(saisonal or result)] if result else None
|
|
|
|
return {
|
|
"dog_name": dog["name"],
|
|
"rasse_name": rasse_info["name"] if rasse_info else dog["rasse"],
|
|
"tipp_des_tages": tipp_des_tages,
|
|
"tipps": result,
|
|
"kategorien": list(dict.fromkeys(t["kategorie"] for t in result)),
|
|
"fell_pflege_art": fell_pflege_art_filter, # 'schneiden' | 'trimmen' | None
|
|
}
|