222 lines
6.4 KiB
Python

"""
Routes pour les élections et les candidats.
"""
from fastapi import APIRouter, HTTPException, status, Depends
from sqlalchemy.orm import Session
from .. import schemas, services
from ..dependencies import get_db, get_current_voter
from ..models import Voter
router = APIRouter(prefix="/api/elections", tags=["elections"])
@router.get("/debug/all")
def debug_all_elections(db: Session = Depends(get_db)):
"""DEBUG: Return all elections with dates for troubleshooting"""
from datetime import datetime
from .. import models
now = datetime.utcnow()
all_elections = db.query(models.Election).all()
return {
"current_time": now.isoformat(),
"elections": [
{
"id": e.id,
"name": e.name,
"is_active": e.is_active,
"start_date": e.start_date.isoformat() if e.start_date else None,
"end_date": e.end_date.isoformat() if e.end_date else None,
"should_be_active": (
e.start_date <= now <= e.end_date and e.is_active
if e.start_date and e.end_date
else False
),
}
for e in all_elections
],
}
@router.get("/active", response_model=list[schemas.ElectionResponse])
def get_active_elections(db: Session = Depends(get_db)):
"""Récupérer toutes les élections actives en cours"""
from datetime import datetime, timedelta
from .. import models
now = datetime.utcnow()
# Allow 1 hour buffer for timezone issues
start_buffer = now - timedelta(hours=1)
end_buffer = now + timedelta(hours=1)
active = db.query(models.Election).filter(
(models.Election.start_date <= end_buffer) &
(models.Election.end_date >= start_buffer) &
(models.Election.is_active == True)
).order_by(models.Election.id.asc()).all()
return active
@router.get("/completed")
def get_completed_elections(db: Session = Depends(get_db)):
"""Récupérer tous les votes passés/terminés"""
from datetime import datetime
elections = db.query(services.models.Election).filter(
services.models.Election.end_date < datetime.utcnow(),
services.models.Election.results_published == True
).all()
return elections
@router.get("/upcoming")
def get_upcoming_elections(db: Session = Depends(get_db)):
"""Récupérer tous les votes à venir"""
from datetime import datetime
elections = db.query(services.models.Election).filter(
services.models.Election.start_date > datetime.utcnow()
).all()
return elections
@router.get("/active/results")
def get_active_election_results(db: Session = Depends(get_db)):
"""Récupérer les résultats de l'élection active"""
election = services.ElectionService.get_active_election(db)
if not election:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="No active election"
)
results = services.VoteService.get_election_results(db, election.id)
return results
@router.get("/{election_id}/candidates")
def get_election_candidates(election_id: int, db: Session = Depends(get_db)):
"""Récupérer les candidats d'une élection"""
election = services.ElectionService.get_election(db, election_id)
if not election:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Election not found"
)
return election.candidates
@router.get("/{election_id}", response_model=schemas.ElectionResponse)
def get_election(election_id: int, db: Session = Depends(get_db)):
"""Récupérer une élection par son ID"""
election = services.ElectionService.get_election(db, election_id)
if not election:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Election not found"
)
return election
@router.get("/{election_id}/results", response_model=schemas.ElectionResultResponse)
def get_election_results(
election_id: int,
db: Session = Depends(get_db)
):
"""
Récupérer les résultats d'une élection.
Disponible après la fermeture du scrutin.
"""
election = services.ElectionService.get_election(db, election_id)
if not election:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Election not found"
)
if not election.results_published:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Results not yet published"
)
results = services.VoteService.get_election_results(db, election_id)
return schemas.ElectionResultResponse(
election_id=election.id,
election_name=election.name,
total_votes=sum(r.vote_count for r in results),
results=results
)
@router.post("/{election_id}/publish-results")
def publish_results(
election_id: int,
db: Session = Depends(get_db)
):
"""
Publier les résultats d'une élection (admin only).
À utiliser après la fermeture du scrutin.
"""
election = services.ElectionService.get_election(db, election_id)
if not election:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Election not found"
)
# Marquer les résultats comme publiés
election.results_published = True
db.commit()
return {
"message": "Results published successfully",
"election_id": election.id,
"election_name": election.name
}
@router.get("/completed", response_model=list[schemas.ElectionResponse])
def get_completed_elections(db: Session = Depends(get_db)):
"""Récupérer toutes les élections terminées (archives)"""
from datetime import datetime
from .. import models
completed = db.query(models.Election).filter(
models.Election.end_date < datetime.utcnow(),
models.Election.results_published == True
).all()
return completed
@router.get("/upcoming", response_model=list[schemas.ElectionResponse])
def get_upcoming_elections(db: Session = Depends(get_db)):
"""Récupérer toutes les élections futures"""
from datetime import datetime
from .. import models
upcoming = db.query(models.Election).filter(
models.Election.start_date > datetime.utcnow()
).all()
return upcoming