import uuid from datetime import datetime, timezone from fastapi import HTTPException from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import hash_password_async, verify_password_async from app.models.utilisateur import Utilisateur from app.schemas.compte import CompteCreate, CompteUpdate class CompteService: def __init__(self, db: AsyncSession) -> None: self.db = db def _serialize(self, u: Utilisateur) -> dict: return { "id": str(u.id), "email": u.email, "nom": u.nom, "prenom": u.prenom, "typeUtilisateur": u.type_utilisateur, "bureauGenreId": u.bureau_genre_id, "telephone": u.telephone, "filiere": u.filiere, "campus": u.campus, "avatarUrl": u.avatar_url, "estActif": u.est_actif, "dateCreation": u.date_creation, "dateModification": u.date_modification, "dernierConnexion": u.derniere_connexion, } async def get_me(self, user: Utilisateur) -> dict: return self._serialize(user) async def create(self, data: CompteCreate) -> dict: requested_role = (data.typeUtilisateur or "UTILISATEUR").upper() if requested_role == "SUPERVISEUR": requested_role = "ADMIN_SYSTEME" if requested_role == "MODERATEUR": requested_role = "GESTIONNAIRE" allowed_roles = {"UTILISATEUR", "GESTIONNAIRE", "ADMIN_SYSTEME"} if requested_role not in allowed_roles: raise HTTPException(status_code=400, detail="Rôle utilisateur invalide") existing = ( await self.db.execute(select(Utilisateur).where(Utilisateur.email == data.email)) ).scalar_one_or_none() if existing: raise HTTPException(status_code=409, detail="Email déjà utilisé") hashed_pwd = await hash_password_async(data.motDePasse) user = Utilisateur( email=data.email, mot_de_passe_hash=hashed_pwd, nom=data.nom, prenom=data.prenom, type_utilisateur=requested_role, bureau_genre_id=data.bureauGenreId, telephone=data.telephone, premiere_connexion=True, ) self.db.add(user) await self.db.commit() await self.db.refresh(user) return self._serialize(user) async def get_by_id(self, compte_id: str) -> dict: user = await self._get_or_404(compte_id) return self._serialize(user) async def list_all(self, page: int = 0, size: int = 20) -> tuple[list[dict], int]: total = (await self.db.execute(select(func.count(Utilisateur.id)))).scalar_one() users = ( await self.db.execute( select(Utilisateur).offset(page * size).limit(size).order_by(Utilisateur.date_creation.desc()) ) ).scalars().all() return [self._serialize(u) for u in users], total async def update(self, compte_id: str, data: CompteUpdate, current_user: Utilisateur) -> dict: user = await self._get_or_404(compte_id) # Only admin or the account owner can update if str(user.id) != str(current_user.id) and current_user.type_utilisateur != "ADMIN_SYSTEME": raise HTTPException(status_code=403, detail="Permission insuffisante") if data.email is not None: existing = ( await self.db.execute( select(Utilisateur).where( Utilisateur.email == data.email, Utilisateur.id != user.id ) ) ).scalar_one_or_none() if existing: raise HTTPException(status_code=409, detail="Email déjà utilisé") user.email = data.email if data.nom is not None: user.nom = data.nom if data.prenom is not None: user.prenom = data.prenom if data.telephone is not None: user.telephone = data.telephone if data.filiere is not None: user.filiere = data.filiere if data.campus is not None: user.campus = data.campus if data.avatarUrl is not None: user.avatar_url = data.avatarUrl await self.db.commit() await self.db.refresh(user) return self._serialize(user) async def change_password(self, compte_id: str, ancien: str, nouveau: str) -> None: user = await self._get_or_404(compte_id) if not (await verify_password_async(ancien, user.mot_de_passe_hash)): raise HTTPException(status_code=401, detail="Ancien mot de passe incorrect") user.mot_de_passe_hash = await hash_password_async(nouveau) user.premiere_connexion = False await self.db.commit() async def reset_password(self, compte_id: str, nouveau: str) -> None: user = await self._get_or_404(compte_id) user.mot_de_passe_hash = await hash_password_async(nouveau) user.premiere_connexion = False await self.db.commit() async def deactivate(self, compte_id: str) -> None: user = await self._get_or_404(compte_id) user.est_actif = False await self.db.commit() async def reactivate(self, compte_id: str) -> None: user = await self._get_or_404(compte_id) user.est_actif = True await self.db.commit() async def _get_or_404(self, compte_id: str) -> Utilisateur: user = ( await self.db.execute( select(Utilisateur).where(Utilisateur.id == uuid.UUID(compte_id)) ) ).scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="Compte non trouvé") return user # ── Preferences ─────────────────────────────────────────────────────────── async def get_preferences(self, user_id: uuid.UUID) -> dict: from sqlalchemy import select from app.models.preference import Preference pref = ( await self.db.execute( select(Preference).where(Preference.utilisateur_id == user_id) ) ).scalar_one_or_none() if pref is None: # Create default preferences on first access pref = Preference(utilisateur_id=user_id) self.db.add(pref) await self.db.commit() await self.db.refresh(pref) return { "id": str(pref.id), "utilisateurId": str(pref.utilisateur_id), "notificationsActivees": pref.notifications_activees, "notificationsEmail": pref.notifications_email, "notificationsPush": pref.notifications_push, "langue": pref.langue, "theme": pref.theme, "parametres": pref.parametres, } async def update_preferences(self, user_id: uuid.UUID, data) -> dict: from sqlalchemy import select from app.models.preference import Preference pref = ( await self.db.execute( select(Preference).where(Preference.utilisateur_id == user_id) ) ).scalar_one_or_none() if pref is None: pref = Preference(utilisateur_id=user_id) self.db.add(pref) await self.db.flush() if data.notificationsActivees is not None: pref.notifications_activees = data.notificationsActivees if data.notificationsEmail is not None: pref.notifications_email = data.notificationsEmail if data.notificationsPush is not None: pref.notifications_push = data.notificationsPush if data.langue is not None: pref.langue = data.langue if data.theme is not None: pref.theme = data.theme if data.parametres is not None: pref.parametres = data.parametres await self.db.commit() await self.db.refresh(pref) return { "id": str(pref.id), "utilisateurId": str(pref.utilisateur_id), "notificationsActivees": pref.notifications_activees, "notificationsEmail": pref.notifications_email, "notificationsPush": pref.notifications_push, "langue": pref.langue, "theme": pref.theme, "parametres": pref.parametres, }