146 lines
4.5 KiB
Python
146 lines
4.5 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
|
|
|
|
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
|
|
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:
|
|
rows = conn.execute(
|
|
"SELECT * FROM dogs WHERE user_id=? ORDER BY id", (user["id"],)
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
@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}")
|
|
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 speichern
|
|
ext = os.path.splitext(file.filename or "")[1] or ".jpg"
|
|
filename = f"dog_{dog_id}_{uuid.uuid4().hex[:8]}{ext}"
|
|
path = os.path.join(MEDIA_DIR, "dogs", filename)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
|
|
content = await file.read()
|
|
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}
|
|
|
|
|
|
# Ö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,
|
|
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.")
|
|
return dict(dog)
|