from fastapi import FastAPI, APIRouter, Depends, HTTPException, status, BackgroundTasks from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from dotenv import load_dotenv from starlette.middleware.cors import CORSMiddleware from motor.motor_asyncio import AsyncIOMotorClient import os import logging from pathlib import Path from pydantic import BaseModel, Field, ConfigDict, EmailStr from typing import List, Optional import uuid from datetime import datetime, timezone, timedelta import jwt from passlib.context import CryptContext import asyncio # SendGrid for email alerts try: from sendgrid import SendGridAPIClient from sendgrid.helpers.mail import Mail SENDGRID_AVAILABLE = True except ImportError: SENDGRID_AVAILABLE = False ROOT_DIR = Path(__file__).parent load_dotenv(ROOT_DIR / '.env') # MongoDB connection mongo_url = os.environ['MONGO_URL'] client = AsyncIOMotorClient(mongo_url) db = client[os.environ['DB_NAME']] # JWT Config SECRET_KEY = os.environ.get('JWT_SECRET', 'maintech-pro-secret-key-2024') ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_HOURS = 24 # Email Config SENDGRID_API_KEY = os.environ.get('SENDGRID_API_KEY', '') SENDER_EMAIL = os.environ.get('SENDER_EMAIL', 'noreply@maintech-pro.com') ALERT_EMAIL = os.environ.get('ALERT_EMAIL', 'info@maintech-pro.fr') # Password hashing pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # Security security = HTTPBearer() # Create the main app app = FastAPI(title="MainTech-pro API") # Create a router with the /api prefix api_router = APIRouter(prefix="/api") # ============== MODELS ============== class UserBase(BaseModel): email: EmailStr nom: str prenom: str role: str = "agent" # admin or agent class UserCreate(UserBase): password: str class UserUpdate(BaseModel): email: Optional[EmailStr] = None nom: Optional[str] = None prenom: Optional[str] = None role: Optional[str] = None password: Optional[str] = None class User(UserBase): model_config = ConfigDict(extra="ignore") id: str = Field(default_factory=lambda: str(uuid.uuid4())) created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) class UserResponse(BaseModel): id: str email: str nom: str prenom: str role: str created_at: str class LoginRequest(BaseModel): email: EmailStr password: str class TokenResponse(BaseModel): access_token: str token_type: str = "bearer" user: UserResponse class PointageBase(BaseModel): date: str # Format: YYYY-MM-DD heure_arrivee: str # Format: HH:MM heure_depart: str # Format: HH:MM pause_minutes: int = 0 # Pause en minutes à déduire reference_machine: str is_absent: bool = False # Case absence class PointageCreate(PointageBase): pass class PointageUpdate(BaseModel): heure_arrivee: Optional[str] = None heure_depart: Optional[str] = None pause_minutes: Optional[int] = None reference_machine: Optional[str] = None is_absent: Optional[bool] = None class FactureRequest(BaseModel): month: str # Format: YYYY-MM taux_horaire: float = 37.72 majoration_samedi: float = 50.0 # Pourcentage de majoration class FactureResponse(BaseModel): month: str taux_horaire: float taux_horaire_samedi: float heures_normales: float heures_samedi: float montant_heures_normales: float montant_heures_samedi: float total_ht: float tva: float total_ttc: float details_par_agent: list def calculate_hours_worked(heure_arrivee: str, heure_depart: str, pause_minutes: int = 0) -> float: """Calculate hours worked between arrival and departure times minus pause""" try: arr_parts = heure_arrivee.split(":") dep_parts = heure_depart.split(":") arr_minutes = int(arr_parts[0]) * 60 + int(arr_parts[1]) dep_minutes = int(dep_parts[0]) * 60 + int(dep_parts[1]) # Handle overnight shifts if dep_minutes < arr_minutes: dep_minutes += 24 * 60 diff_minutes = dep_minutes - arr_minutes - pause_minutes if diff_minutes < 0: diff_minutes = 0 return round(diff_minutes / 60, 2) except: return 0.0 class Pointage(PointageBase): model_config = ConfigDict(extra="ignore") id: str = Field(default_factory=lambda: str(uuid.uuid4())) user_id: str user_nom: str = "" user_prenom: str = "" heures_travaillees: float = 0.0 created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) class PointageResponse(BaseModel): id: str user_id: str user_nom: str user_prenom: str date: str heure_arrivee: str heure_depart: str pause_minutes: int heures_travaillees: float reference_machine: str is_absent: bool created_at: str # ============== HELPERS ============== def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) def create_access_token(data: dict) -> str: to_encode = data.copy() expire = datetime.now(timezone.utc) + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) # ============== EMAIL FUNCTIONS ============== def send_alert_email(subject: str, content: str): """Send alert email via SendGrid""" if not SENDGRID_AVAILABLE or not SENDGRID_API_KEY: logger.warning(f"SendGrid not configured. Alert: {subject}") return False try: message = Mail( from_email=SENDER_EMAIL, to_emails=ALERT_EMAIL, subject=subject, html_content=content ) sg = SendGridAPIClient(SENDGRID_API_KEY) response = sg.send(message) return response.status_code == 202 except Exception as e: logger.error(f"Failed to send email: {str(e)}") return False async def check_missing_pointages_and_alert(): """Check for agents who haven't clocked in today (run after 20:00 weekdays)""" now = datetime.now(timezone.utc) # Only check Monday to Friday if now.weekday() >= 5: # Saturday=5, Sunday=6 return [] today = now.date().isoformat() # Get all agents agents = await db.users.find({"role": "agent"}, {"_id": 0}).to_list(100) # Get today's pointages today_pointages = await db.pointages.find({"date": today}, {"_id": 0}).to_list(100) agents_with_pointage = {p['user_id'] for p in today_pointages} # Find agents without pointage missing_agents = [a for a in agents if a['id'] not in agents_with_pointage] return missing_agents def send_missing_pointage_alert(missing_agents: list, date: str): """Send email alert for missing pointages""" if not missing_agents: return False subject = f"⚠️ Alerte Pointage MainTech-pro - {len(missing_agents)} agent(s) non pointé(s)" agents_list = "".join([ f"
  • {a['prenom']} {a['nom']} ({a['email']})
    • " for a in missing_agents ]) content = f"""

    ⚠️ Alerte Pointage

    Les agents suivants n'ont pas pointé le {date} :

    MainTech-pro - Gestion d'effectif Tecma Aries
    Ce message est envoyé automatiquement après 20h00 les jours de semaine (lundi-vendredi).

    """ return send_alert_email(subject, content) async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): try: token = credentials.credentials payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) user_id = payload.get("sub") if user_id is None: raise HTTPException(status_code=401, detail="Token invalide") user = await db.users.find_one({"id": user_id}, {"_id": 0, "password_hash": 0}) if user is None: raise HTTPException(status_code=401, detail="Utilisateur non trouvé") if isinstance(user.get('created_at'), str): user['created_at'] = datetime.fromisoformat(user['created_at']) return user except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Token expiré") except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Token invalide") async def get_admin_user(current_user: dict = Depends(get_current_user)): if current_user.get("role") != "admin": raise HTTPException(status_code=403, detail="Accès réservé aux administrateurs") return current_user def user_to_response(user: dict) -> UserResponse: created_at = user.get('created_at', '') if isinstance(created_at, datetime): created_at = created_at.isoformat() return UserResponse( id=user['id'], email=user['email'], nom=user['nom'], prenom=user['prenom'], role=user['role'], created_at=created_at ) def pointage_to_response(pointage: dict) -> PointageResponse: created_at = pointage.get('created_at', '') if isinstance(created_at, datetime): created_at = created_at.isoformat() return PointageResponse( id=pointage['id'], user_id=pointage['user_id'], user_nom=pointage.get('user_nom', ''), user_prenom=pointage.get('user_prenom', ''), date=pointage['date'], heure_arrivee=pointage.get('heure_arrivee', ''), heure_depart=pointage.get('heure_depart', ''), pause_minutes=pointage.get('pause_minutes', 0), heures_travaillees=pointage.get('heures_travaillees', 0.0), reference_machine=pointage.get('reference_machine', ''), is_absent=pointage.get('is_absent', False), created_at=created_at ) # ============== AUTH ROUTES ============== @api_router.post("/auth/login", response_model=TokenResponse) async def login(request: LoginRequest): user = await db.users.find_one({"email": request.email}, {"_id": 0}) if not user: raise HTTPException(status_code=401, detail="Email ou mot de passe incorrect") if not verify_password(request.password, user.get('password_hash', '')): raise HTTPException(status_code=401, detail="Email ou mot de passe incorrect") access_token = create_access_token({"sub": user['id']}) user_response = user_to_response(user) return TokenResponse(access_token=access_token, user=user_response) @api_router.get("/auth/me", response_model=UserResponse) async def get_me(current_user: dict = Depends(get_current_user)): return user_to_response(current_user) # ============== USER ROUTES ============== @api_router.get("/users", response_model=List[UserResponse]) async def get_users(current_user: dict = Depends(get_admin_user)): users = await db.users.find({}, {"_id": 0, "password_hash": 0}).to_list(1000) return [user_to_response(u) for u in users] @api_router.post("/users", response_model=UserResponse) async def create_user(user_data: UserCreate, current_user: dict = Depends(get_admin_user)): # Check if email already exists existing = await db.users.find_one({"email": user_data.email}) if existing: raise HTTPException(status_code=400, detail="Cet email est déjà utilisé") user_id = str(uuid.uuid4()) now = datetime.now(timezone.utc) user_doc = { "id": user_id, "email": user_data.email, "nom": user_data.nom, "prenom": user_data.prenom, "role": user_data.role, "password_hash": hash_password(user_data.password), "created_at": now.isoformat() } await db.users.insert_one(user_doc) return UserResponse( id=user_id, email=user_data.email, nom=user_data.nom, prenom=user_data.prenom, role=user_data.role, created_at=now.isoformat() ) @api_router.put("/users/{user_id}", response_model=UserResponse) async def update_user(user_id: str, user_data: UserUpdate, current_user: dict = Depends(get_admin_user)): existing = await db.users.find_one({"id": user_id}, {"_id": 0}) if not existing: raise HTTPException(status_code=404, detail="Utilisateur non trouvé") update_data = {} if user_data.email: # Check if new email is taken by another user email_check = await db.users.find_one({"email": user_data.email, "id": {"$ne": user_id}}) if email_check: raise HTTPException(status_code=400, detail="Cet email est déjà utilisé") update_data["email"] = user_data.email if user_data.nom: update_data["nom"] = user_data.nom if user_data.prenom: update_data["prenom"] = user_data.prenom if user_data.role: update_data["role"] = user_data.role if user_data.password: update_data["password_hash"] = hash_password(user_data.password) if update_data: await db.users.update_one({"id": user_id}, {"$set": update_data}) updated = await db.users.find_one({"id": user_id}, {"_id": 0, "password_hash": 0}) return user_to_response(updated) @api_router.delete("/users/{user_id}") async def delete_user(user_id: str, current_user: dict = Depends(get_admin_user)): if user_id == current_user['id']: raise HTTPException(status_code=400, detail="Vous ne pouvez pas supprimer votre propre compte") result = await db.users.delete_one({"id": user_id}) if result.deleted_count == 0: raise HTTPException(status_code=404, detail="Utilisateur non trouvé") # Also delete user's pointages await db.pointages.delete_many({"user_id": user_id}) return {"message": "Utilisateur supprimé avec succès"} # ============== POINTAGE ROUTES ============== @api_router.get("/pointages", response_model=List[PointageResponse]) async def get_pointages( user_id: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None, current_user: dict = Depends(get_current_user) ): query = {} # Agents can only see their own pointages if current_user['role'] != 'admin': query['user_id'] = current_user['id'] elif user_id: query['user_id'] = user_id if start_date: query['date'] = {"$gte": start_date} if end_date: if 'date' in query: query['date']['$lte'] = end_date else: query['date'] = {"$lte": end_date} pointages = await db.pointages.find(query, {"_id": 0}).sort("date", -1).to_list(1000) return [pointage_to_response(p) for p in pointages] @api_router.post("/pointages", response_model=PointageResponse) async def create_pointage(pointage_data: PointageCreate, current_user: dict = Depends(get_current_user)): # Check if pointage already exists for this date existing = await db.pointages.find_one({ "user_id": current_user['id'], "date": pointage_data.date }) if existing: raise HTTPException(status_code=400, detail="Un pointage existe déjà pour cette date") # Calculate hours worked automatically (minus pause) - 0 if absent if pointage_data.is_absent: heures_travaillees = 0.0 else: heures_travaillees = calculate_hours_worked( pointage_data.heure_arrivee, pointage_data.heure_depart, pointage_data.pause_minutes ) pointage_id = str(uuid.uuid4()) now = datetime.now(timezone.utc) pointage_doc = { "id": pointage_id, "user_id": current_user['id'], "user_nom": current_user['nom'], "user_prenom": current_user['prenom'], "date": pointage_data.date, "heure_arrivee": pointage_data.heure_arrivee if not pointage_data.is_absent else "", "heure_depart": pointage_data.heure_depart if not pointage_data.is_absent else "", "pause_minutes": pointage_data.pause_minutes if not pointage_data.is_absent else 0, "heures_travaillees": heures_travaillees, "reference_machine": pointage_data.reference_machine if not pointage_data.is_absent else "", "is_absent": pointage_data.is_absent, "created_at": now.isoformat() } await db.pointages.insert_one(pointage_doc) return pointage_to_response(pointage_doc) @api_router.put("/pointages/{pointage_id}", response_model=PointageResponse) async def update_pointage( pointage_id: str, pointage_data: PointageUpdate, current_user: dict = Depends(get_current_user) ): query = {"id": pointage_id} # Agents can only update their own pointages if current_user['role'] != 'admin': query['user_id'] = current_user['id'] existing = await db.pointages.find_one(query, {"_id": 0}) if not existing: raise HTTPException(status_code=404, detail="Pointage non trouvé") update_data = {} if pointage_data.heure_arrivee is not None: update_data["heure_arrivee"] = pointage_data.heure_arrivee if pointage_data.heure_depart is not None: update_data["heure_depart"] = pointage_data.heure_depart if pointage_data.pause_minutes is not None: update_data["pause_minutes"] = pointage_data.pause_minutes if pointage_data.reference_machine is not None: update_data["reference_machine"] = pointage_data.reference_machine if pointage_data.is_absent is not None: update_data["is_absent"] = pointage_data.is_absent if pointage_data.is_absent: update_data["heures_travaillees"] = 0.0 # Recalculate hours if times or pause changed (and not absent) is_absent = update_data.get("is_absent", existing.get("is_absent", False)) if not is_absent and ("heure_arrivee" in update_data or "heure_depart" in update_data or "pause_minutes" in update_data): arr = update_data.get("heure_arrivee", existing.get("heure_arrivee", "07:00")) dep = update_data.get("heure_depart", existing.get("heure_depart", "17:00")) pause = update_data.get("pause_minutes", existing.get("pause_minutes", 0)) update_data["heures_travaillees"] = calculate_hours_worked(arr, dep, pause) if update_data: await db.pointages.update_one({"id": pointage_id}, {"$set": update_data}) updated = await db.pointages.find_one({"id": pointage_id}, {"_id": 0}) return pointage_to_response(updated) @api_router.delete("/pointages/{pointage_id}") async def delete_pointage(pointage_id: str, current_user: dict = Depends(get_current_user)): query = {"id": pointage_id} # Agents can only delete their own pointages if current_user['role'] != 'admin': query['user_id'] = current_user['id'] result = await db.pointages.delete_one(query) if result.deleted_count == 0: raise HTTPException(status_code=404, detail="Pointage non trouvé") return {"message": "Pointage supprimé avec succès"} # ============== REPORTS ROUTES ============== @api_router.get("/reports/summary") async def get_reports_summary( period: str = "week", # week or month user_id: Optional[str] = None, current_user: dict = Depends(get_current_user) ): today = datetime.now(timezone.utc).date() if period == "week": start_date = today - timedelta(days=today.weekday()) end_date = start_date + timedelta(days=6) else: # month start_date = today.replace(day=1) if today.month == 12: end_date = today.replace(year=today.year + 1, month=1, day=1) - timedelta(days=1) else: end_date = today.replace(month=today.month + 1, day=1) - timedelta(days=1) query = { "date": { "$gte": start_date.isoformat(), "$lte": end_date.isoformat() } } # Agents can only see their own reports if current_user['role'] != 'admin': query['user_id'] = current_user['id'] elif user_id: query['user_id'] = user_id pointages = await db.pointages.find(query, {"_id": 0}).to_list(1000) # Calculate totals per user user_totals = {} for p in pointages: uid = p['user_id'] if uid not in user_totals: user_totals[uid] = { "user_id": uid, "user_nom": p.get('user_nom', ''), "user_prenom": p.get('user_prenom', ''), "total_heures": 0, "nombre_jours": 0, "machines": [] } user_totals[uid]['total_heures'] += p.get('heures_travaillees', 0) user_totals[uid]['nombre_jours'] += 1 if p['reference_machine'] not in user_totals[uid]['machines']: user_totals[uid]['machines'].append(p['reference_machine']) return { "period": period, "start_date": start_date.isoformat(), "end_date": end_date.isoformat(), "summaries": list(user_totals.values()), "total_heures_equipe": sum(u['total_heures'] for u in user_totals.values()), "total_jours_equipe": sum(u['nombre_jours'] for u in user_totals.values()) } @api_router.get("/reports/monthly-stats") async def get_monthly_stats( user_id: Optional[str] = None, current_user: dict = Depends(get_current_user) ): """Get monthly statistics for current year including Saturday hours""" today = datetime.now(timezone.utc).date() year_start = today.replace(month=1, day=1) query = { "date": { "$gte": year_start.isoformat(), "$lte": today.isoformat() } } # Agents can only see their own stats if current_user['role'] != 'admin': query['user_id'] = current_user['id'] elif user_id: query['user_id'] = user_id pointages = await db.pointages.find(query, {"_id": 0}).to_list(10000) # Group by month monthly_data = {} for p in pointages: month = p['date'][:7] # YYYY-MM if month not in monthly_data: monthly_data[month] = { "month": month, "total_heures": 0, "heures_samedi": 0, "nombre_jours": 0, "nombre_samedis": 0, "users": {} } heures = p.get('heures_travaillees', 0) monthly_data[month]['total_heures'] += heures monthly_data[month]['nombre_jours'] += 1 # Check if it's Saturday (weekday 5) date_obj = datetime.strptime(p['date'], "%Y-%m-%d").date() is_saturday = date_obj.weekday() == 5 if is_saturday: monthly_data[month]['heures_samedi'] += heures monthly_data[month]['nombre_samedis'] += 1 uid = p['user_id'] if uid not in monthly_data[month]['users']: monthly_data[month]['users'][uid] = { "user_id": uid, "user_nom": p.get('user_nom', ''), "user_prenom": p.get('user_prenom', ''), "heures": 0, "heures_samedi": 0, "jours": 0, "samedis": 0 } monthly_data[month]['users'][uid]['heures'] += heures monthly_data[month]['users'][uid]['jours'] += 1 if is_saturday: monthly_data[month]['users'][uid]['heures_samedi'] += heures monthly_data[month]['users'][uid]['samedis'] += 1 # Convert users dict to list and sort by month result = [] for month_key in sorted(monthly_data.keys(), reverse=True): data = monthly_data[month_key] data['users'] = list(data['users'].values()) result.append(data) # Current month stats current_month = today.strftime("%Y-%m") current_month_data = monthly_data.get(current_month, { "total_heures": 0, "heures_samedi": 0, "nombre_jours": 0, "nombre_samedis": 0 }) return { "year": today.year, "current_month": current_month, "current_month_heures": current_month_data.get('total_heures', 0), "current_month_heures_samedi": current_month_data.get('heures_samedi', 0), "current_month_jours": current_month_data.get('nombre_jours', 0), "current_month_samedis": current_month_data.get('nombre_samedis', 0), "monthly_data": result } # ============== INIT DATA ============== @api_router.post("/init-data") async def init_default_data(): """Initialize default users for MainTech-pro""" # Check if users already exist existing = await db.users.count_documents({}) if existing > 0: return {"message": "Les données existent déjà", "users_count": existing} default_users = [ { "id": str(uuid.uuid4()), "email": "ali.khattabi@maintech-pro.com", "nom": "KHATTABI", "prenom": "Ali", "role": "admin", "password_hash": hash_password("admin123"), "created_at": datetime.now(timezone.utc).isoformat() }, { "id": str(uuid.uuid4()), "email": "omar.ouabdelmoumen@maintech-pro.com", "nom": "OUABDELMOUMEN", "prenom": "Omar", "role": "agent", "password_hash": hash_password("agent123"), "created_at": datetime.now(timezone.utc).isoformat() }, { "id": str(uuid.uuid4()), "email": "amine.lajdel@maintech-pro.com", "nom": "LAJDEL", "prenom": "Amine", "role": "agent", "password_hash": hash_password("agent123"), "created_at": datetime.now(timezone.utc).isoformat() }, { "id": str(uuid.uuid4()), "email": "abdelillah.mallouk@maintech-pro.com", "nom": "MALLOUK", "prenom": "Abdelillah", "role": "agent", "password_hash": hash_password("agent123"), "created_at": datetime.now(timezone.utc).isoformat() } ] await db.users.insert_many(default_users) return { "message": "Utilisateurs créés avec succès", "users": [ {"email": u["email"], "password": "admin123" if u["role"] == "admin" else "agent123", "role": u["role"]} for u in default_users ] } @api_router.get("/") async def root(): return {"message": "MainTech-pro API", "version": "1.0.0"} @api_router.post("/facture/calculate", response_model=FactureResponse) async def calculate_facture( request: FactureRequest, current_user: dict = Depends(get_admin_user) ): """Calculate invoice for a given month""" month = request.month taux_horaire = request.taux_horaire majoration = request.majoration_samedi / 100 # Convert percentage to decimal taux_horaire_samedi = taux_horaire * (1 + majoration) # Get all pointages for the month start_date = f"{month}-01" # Calculate end of month year, mon = map(int, month.split("-")) if mon == 12: end_date = f"{year + 1}-01-01" else: end_date = f"{year}-{mon + 1:02d}-01" pointages = await db.pointages.find({ "date": {"$gte": start_date, "$lt": end_date} }, {"_id": 0}).to_list(10000) # Calculate totals heures_normales = 0.0 heures_samedi = 0.0 agent_details = {} for p in pointages: heures = p.get('heures_travaillees', 0) date_obj = datetime.strptime(p['date'], "%Y-%m-%d").date() is_saturday = date_obj.weekday() == 5 uid = p['user_id'] if uid not in agent_details: agent_details[uid] = { "user_id": uid, "user_nom": p.get('user_nom', ''), "user_prenom": p.get('user_prenom', ''), "heures_normales": 0, "heures_samedi": 0, "montant": 0 } if is_saturday: heures_samedi += heures agent_details[uid]['heures_samedi'] += heures else: heures_normales += heures agent_details[uid]['heures_normales'] += heures # Calculate amounts per agent for uid in agent_details: agent = agent_details[uid] agent['montant'] = (agent['heures_normales'] * taux_horaire) + (agent['heures_samedi'] * taux_horaire_samedi) montant_heures_normales = heures_normales * taux_horaire montant_heures_samedi = heures_samedi * taux_horaire_samedi total_ht = montant_heures_normales + montant_heures_samedi tva = total_ht * 0.20 # TVA 20% total_ttc = total_ht + tva return FactureResponse( month=month, taux_horaire=taux_horaire, taux_horaire_samedi=round(taux_horaire_samedi, 2), heures_normales=round(heures_normales, 2), heures_samedi=round(heures_samedi, 2), montant_heures_normales=round(montant_heures_normales, 2), montant_heures_samedi=round(montant_heures_samedi, 2), total_ht=round(total_ht, 2), tva=round(tva, 2), total_ttc=round(total_ttc, 2), details_par_agent=list(agent_details.values()) ) # ============== ALERT ROUTES ============== @api_router.get("/alerts/check-missing-pointages") async def check_missing_pointages(current_user: dict = Depends(get_admin_user)): """Check for agents who haven't clocked in today and send alert""" today = datetime.now(timezone.utc).date() # Only check Monday to Friday if today.weekday() >= 5: return {"message": "Pas de vérification le weekend", "missing_agents": []} missing_agents = await check_missing_pointages_and_alert() return { "date": today.isoformat(), "missing_count": len(missing_agents), "missing_agents": [ {"nom": a['nom'], "prenom": a['prenom'], "email": a['email']} for a in missing_agents ] } @api_router.post("/alerts/send-missing-pointage-alert") async def send_alert(current_user: dict = Depends(get_admin_user)): """Manually trigger the missing pointage alert email""" today = datetime.now(timezone.utc).date() missing_agents = await check_missing_pointages_and_alert() if not missing_agents: return {"message": "Tous les agents ont pointé aujourd'hui", "sent": False} success = send_missing_pointage_alert(missing_agents, today.isoformat()) return { "message": f"Alerte envoyée pour {len(missing_agents)} agent(s)" if success else "Échec de l'envoi (SendGrid non configuré)", "sent": success, "missing_agents": [ {"nom": a['nom'], "prenom": a['prenom']} for a in missing_agents ] } @api_router.get("/alerts/config") async def get_alert_config(current_user: dict = Depends(get_admin_user)): """Get current alert configuration""" return { "sendgrid_configured": bool(SENDGRID_API_KEY), "sender_email": SENDER_EMAIL, "alert_email": ALERT_EMAIL, "alert_time": "20:00", "alert_days": "Lundi - Vendredi" } # Include the router in the main app app.include_router(api_router) app.add_middleware( CORSMiddleware, allow_credentials=True, allow_origins=os.environ.get('CORS_ORIGINS', '*').split(','), allow_methods=["*"], allow_headers=["*"], ) # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @app.on_event("shutdown") async def shutdown_db_client(): client.close()