214 lines
7.2 KiB
Python
214 lines
7.2 KiB
Python
"""Routes G6/C6 — Sondages et votes de conscience de groupe."""
|
|
from datetime import datetime, timezone
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, desc, func
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_membre, require_executif
|
|
from app.models.sondage import Sondage
|
|
from app.models.option_sondage import OptionSondage
|
|
from app.models.vote import Vote
|
|
from app.models.membre import Membre
|
|
from app.schemas.phase8 import SondageCreate, SondageOut, SondageUpdate, OptionOut, VoteCreate
|
|
|
|
router = APIRouter(prefix="/sondages", tags=["G6/C6 — Sondages et votes"])
|
|
|
|
TYPES_VALIDES = {"oui_non", "choix_multiple", "unanimite"}
|
|
|
|
|
|
async def _build_sondage_out(sondage: Sondage, membre_id, db: AsyncSession) -> SondageOut:
|
|
"""Construit la sortie complète d'un sondage avec comptage des votes."""
|
|
cree = await db.get(Membre, sondage.cree_par) if sondage.cree_par else None
|
|
|
|
# Options avec comptage
|
|
result = await db.execute(
|
|
select(OptionSondage)
|
|
.where(OptionSondage.sondage_id == sondage.id)
|
|
.order_by(OptionSondage.ordre)
|
|
)
|
|
options = result.scalars().all()
|
|
|
|
options_out = []
|
|
total = 0
|
|
for opt in options:
|
|
count_result = await db.execute(
|
|
select(func.count(Vote.id)).where(Vote.option_id == opt.id)
|
|
)
|
|
nb = count_result.scalar() or 0
|
|
total += nb
|
|
options_out.append(OptionOut(
|
|
id=opt.id, texte=opt.texte, ordre=opt.ordre, nb_votes=nb
|
|
))
|
|
|
|
# Le membre a-t-il voté?
|
|
vote_result = await db.execute(
|
|
select(Vote).where(Vote.sondage_id == sondage.id, Vote.membre_id == membre_id)
|
|
)
|
|
a_vote = vote_result.scalar_one_or_none() is not None
|
|
|
|
return SondageOut(
|
|
id=sondage.id,
|
|
titre=sondage.titre,
|
|
description=sondage.description,
|
|
type=sondage.type,
|
|
cree_par_prenom=cree.prenom if cree else None,
|
|
date_ouverture=sondage.date_ouverture,
|
|
date_fermeture=sondage.date_fermeture,
|
|
statut=sondage.statut,
|
|
options=options_out,
|
|
total_votes=total,
|
|
a_vote=a_vote,
|
|
)
|
|
|
|
|
|
@router.post("/", response_model=SondageOut)
|
|
async def creer_sondage(
|
|
req: SondageCreate,
|
|
membre=Depends(require_executif),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""G6 — Créer un sondage de conscience de groupe."""
|
|
if req.type not in TYPES_VALIDES:
|
|
raise HTTPException(status_code=400, detail=f"Type invalide. Choix: {list(TYPES_VALIDES)}")
|
|
|
|
if req.date_fermeture <= req.date_ouverture:
|
|
raise HTTPException(status_code=400, detail="La date de fermeture doit être après l'ouverture")
|
|
|
|
# Pour oui_non, créer automatiquement les options
|
|
options_data = req.options
|
|
if req.type == "oui_non" and len(options_data) == 0:
|
|
from app.schemas.phase8 import OptionCreate
|
|
options_data = [OptionCreate(texte="Oui"), OptionCreate(texte="Non")]
|
|
|
|
if len(options_data) < 2:
|
|
raise HTTPException(status_code=400, detail="Au moins 2 options sont requises")
|
|
|
|
sondage = Sondage(
|
|
groupe_id=membre.groupe_id,
|
|
titre=req.titre.strip(),
|
|
description=req.description,
|
|
type=req.type,
|
|
cree_par=membre.id,
|
|
date_ouverture=req.date_ouverture,
|
|
date_fermeture=req.date_fermeture,
|
|
statut="brouillon",
|
|
)
|
|
db.add(sondage)
|
|
await db.flush()
|
|
|
|
for i, opt in enumerate(options_data):
|
|
option = OptionSondage(
|
|
sondage_id=sondage.id,
|
|
texte=opt.texte.strip(),
|
|
ordre=i,
|
|
)
|
|
db.add(option)
|
|
await db.flush()
|
|
|
|
return await _build_sondage_out(sondage, membre.id, db)
|
|
|
|
|
|
@router.get("/", response_model=list[SondageOut])
|
|
async def liste_sondages(
|
|
statut: str | None = None,
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Lister les sondages du groupe."""
|
|
query = select(Sondage).where(Sondage.groupe_id == membre.groupe_id)
|
|
if statut:
|
|
query = query.where(Sondage.statut == statut)
|
|
query = query.order_by(desc(Sondage.date_ouverture))
|
|
|
|
result = await db.execute(query)
|
|
sondages = result.scalars().all()
|
|
|
|
return [await _build_sondage_out(s, membre.id, db) for s in sondages]
|
|
|
|
|
|
@router.get("/{sondage_id}", response_model=SondageOut)
|
|
async def detail_sondage(
|
|
sondage_id: str,
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
result = await db.execute(
|
|
select(Sondage).where(Sondage.id == sondage_id, Sondage.groupe_id == membre.groupe_id)
|
|
)
|
|
sondage = result.scalar_one_or_none()
|
|
if not sondage:
|
|
raise HTTPException(status_code=404, detail="Sondage introuvable")
|
|
return await _build_sondage_out(sondage, membre.id, db)
|
|
|
|
|
|
@router.patch("/{sondage_id}", response_model=SondageOut)
|
|
async def modifier_statut_sondage(
|
|
sondage_id: str,
|
|
req: SondageUpdate,
|
|
membre=Depends(require_executif),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Ouvrir ou fermer un sondage."""
|
|
result = await db.execute(
|
|
select(Sondage).where(Sondage.id == sondage_id, Sondage.groupe_id == membre.groupe_id)
|
|
)
|
|
sondage = result.scalar_one_or_none()
|
|
if not sondage:
|
|
raise HTTPException(status_code=404, detail="Sondage introuvable")
|
|
|
|
if req.statut and req.statut in ("brouillon", "ouvert", "ferme"):
|
|
sondage.statut = req.statut
|
|
await db.flush()
|
|
|
|
return await _build_sondage_out(sondage, membre.id, db)
|
|
|
|
|
|
@router.post("/{sondage_id}/voter")
|
|
async def voter(
|
|
sondage_id: str,
|
|
req: VoteCreate,
|
|
membre=Depends(get_current_membre),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""C6 — Voter dans un sondage ouvert. Un membre = un vote."""
|
|
result = await db.execute(
|
|
select(Sondage).where(Sondage.id == sondage_id, Sondage.groupe_id == membre.groupe_id)
|
|
)
|
|
sondage = result.scalar_one_or_none()
|
|
if not sondage:
|
|
raise HTTPException(status_code=404, detail="Sondage introuvable")
|
|
|
|
if sondage.statut != "ouvert":
|
|
raise HTTPException(status_code=400, detail="Ce sondage n'est pas ouvert au vote")
|
|
|
|
now = datetime.now(timezone.utc)
|
|
if now < sondage.date_ouverture or now > sondage.date_fermeture:
|
|
raise HTTPException(status_code=400, detail="La période de vote est terminée ou pas encore ouverte")
|
|
|
|
# Vérifier qu'il n'a pas déjà voté
|
|
existing = await db.execute(
|
|
select(Vote).where(Vote.sondage_id == sondage.id, Vote.membre_id == membre.id)
|
|
)
|
|
if existing.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Vous avez déjà voté dans ce sondage")
|
|
|
|
# Vérifier que l'option appartient au sondage
|
|
opt_result = await db.execute(
|
|
select(OptionSondage).where(
|
|
OptionSondage.id == req.option_id,
|
|
OptionSondage.sondage_id == sondage.id,
|
|
)
|
|
)
|
|
if not opt_result.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Option invalide")
|
|
|
|
vote = Vote(
|
|
sondage_id=sondage.id,
|
|
option_id=req.option_id,
|
|
membre_id=membre.id,
|
|
)
|
|
db.add(vote)
|
|
await db.flush()
|
|
|
|
return {"message": "Vote enregistré", "option_id": str(req.option_id)}
|