133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
"""Reportes ciudadanos sobre unidades (incidents)."""
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
|
|
|
|
from app.core.deps import get_current_user
|
|
from app.core.supabase_client import supabase_admin
|
|
from app.schemas.incidents import (
|
|
IncidentOut,
|
|
UnitPublic,
|
|
)
|
|
|
|
router = APIRouter(prefix="/incidents", tags=["incidents"])
|
|
|
|
BUCKET = "incident-photos"
|
|
VALID_CATEGORIES = {"derrame", "dano_propiedad", "conducta", "no_recoleccion", "otro"}
|
|
|
|
|
|
@router.get("/units", response_model=list[UnitPublic])
|
|
def list_units(_user: dict = Depends(get_current_user)):
|
|
"""Lista unidades activas para que el ciudadano elija al reportar."""
|
|
try:
|
|
res = (
|
|
supabase_admin.table("units")
|
|
.select("id, plate, status")
|
|
.eq("status", "active")
|
|
.order("id")
|
|
.execute()
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Error al listar unidades: {e}")
|
|
|
|
return [UnitPublic(**r) for r in (res.data or [])]
|
|
|
|
|
|
def _upload_photo(user_id: str, photo: UploadFile) -> str:
|
|
ext = (photo.filename or "").rsplit(".", 1)[-1].lower() or "jpg"
|
|
if ext not in {"jpg", "jpeg", "png", "webp"}:
|
|
ext = "jpg"
|
|
path = f"{user_id}/{uuid.uuid4().hex}.{ext}"
|
|
|
|
content = photo.file.read()
|
|
content_type = photo.content_type or f"image/{ext}"
|
|
|
|
try:
|
|
supabase_admin.storage.from_(BUCKET).upload(
|
|
path=path,
|
|
file=content,
|
|
file_options={"content-type": content_type, "upsert": "false"},
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Error al subir foto: {e}")
|
|
|
|
try:
|
|
public = supabase_admin.storage.from_(BUCKET).get_public_url(path)
|
|
return public if isinstance(public, str) else str(public)
|
|
except Exception:
|
|
return path
|
|
|
|
|
|
@router.post("", response_model=IncidentOut, status_code=201)
|
|
async def create_incident(
|
|
category: str = Form(...),
|
|
description: str = Form(...),
|
|
unit_id: Optional[int] = Form(None),
|
|
photo: Optional[UploadFile] = File(None),
|
|
current_user: dict = Depends(get_current_user),
|
|
):
|
|
if category not in VALID_CATEGORIES:
|
|
raise HTTPException(status_code=400, detail="Categoría inválida")
|
|
if len(description.strip()) < 3:
|
|
raise HTTPException(status_code=400, detail="Descripción demasiado corta")
|
|
|
|
user_id = current_user["user_id"]
|
|
photo_url: Optional[str] = None
|
|
if photo is not None and photo.filename:
|
|
photo_url = _upload_photo(user_id, photo)
|
|
|
|
payload = {
|
|
"user_id": user_id,
|
|
"unit_id": unit_id,
|
|
"category": category,
|
|
"description": description.strip(),
|
|
"photo_url": photo_url,
|
|
}
|
|
try:
|
|
res = supabase_admin.table("incidents").insert(payload).execute()
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Error al guardar reporte: {e}")
|
|
|
|
row = (res.data or [{}])[0]
|
|
return IncidentOut(
|
|
id=row.get("id"),
|
|
user_id=row.get("user_id"),
|
|
unit_id=row.get("unit_id"),
|
|
category=row.get("category"),
|
|
description=row.get("description"),
|
|
photo_url=row.get("photo_url"),
|
|
status=row.get("status") or "open",
|
|
created_at=row.get("created_at") and str(row["created_at"]),
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=list[IncidentOut])
|
|
def my_incidents(current_user: dict = Depends(get_current_user)):
|
|
try:
|
|
res = (
|
|
supabase_admin.table("incidents")
|
|
.select("*")
|
|
.eq("user_id", current_user["user_id"])
|
|
.order("created_at", desc=True)
|
|
.execute()
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Error al listar reportes: {e}")
|
|
|
|
out: list[IncidentOut] = []
|
|
for row in res.data or []:
|
|
out.append(
|
|
IncidentOut(
|
|
id=row.get("id"),
|
|
user_id=row.get("user_id"),
|
|
unit_id=row.get("unit_id"),
|
|
category=row.get("category"),
|
|
description=row.get("description"),
|
|
photo_url=row.get("photo_url"),
|
|
status=row.get("status") or "open",
|
|
created_at=row.get("created_at") and str(row["created_at"]),
|
|
)
|
|
)
|
|
return out
|