""" Modèles pour la gestion des tâches planifiées. """ from django.db import models from django.conf import settings from django.utils import timezone from django.core.validators import MinValueValidator, MaxValueValidator import json class ScheduledCommand(models.Model): """ Commande Django planifiée pour exécution automatique. """ FREQUENCY_CHOICES = [ ('hourly', 'Toutes les heures'), ('daily', 'Quotidien'), ('weekly', 'Hebdomadaire'), ('monthly', 'Mensuel'), ] WEEKDAY_CHOICES = [ (0, 'Lundi'), (1, 'Mardi'), (2, 'Mercredi'), (3, 'Jeudi'), (4, 'Vendredi'), (5, 'Samedi'), (6, 'Dimanche'), ] name = models.CharField( max_length=200, unique=True, verbose_name="Nom de la tâche", help_text="Nom descriptif de la tâche planifiée" ) command_name = models.CharField( max_length=200, verbose_name="Nom de la commande", help_text="Nom de la commande Django (ex: export_interventions_xlsx)" ) command_args = models.JSONField( default=dict, blank=True, verbose_name="Arguments de la commande", help_text="Arguments de la commande au format JSON (ex: {\"output\": \"file.xlsx\"})" ) frequency = models.CharField( max_length=20, choices=FREQUENCY_CHOICES, default='daily', verbose_name="Fréquence" ) hour = models.IntegerField( validators=[MinValueValidator(0), MaxValueValidator(23)], default=0, verbose_name="Heure", help_text="Heure d'exécution (0-23)" ) minute = models.IntegerField( validators=[MinValueValidator(0), MaxValueValidator(59)], default=0, verbose_name="Minute", help_text="Minute d'exécution (0-59)" ) day_of_month = models.IntegerField( null=True, blank=True, validators=[MinValueValidator(1), MaxValueValidator(31)], verbose_name="Jour du mois", help_text="Pour les tâches mensuelles : jour du mois (1-31)" ) day_of_week = models.IntegerField( null=True, blank=True, choices=WEEKDAY_CHOICES, verbose_name="Jour de la semaine", help_text="Pour les tâches hebdomadaires : 0=Lundi, 6=Dimanche" ) is_active = models.BooleanField( default=True, verbose_name="Active", help_text="Si décoché, la tâche ne sera pas exécutée" ) send_email = models.BooleanField( default=False, verbose_name="Envoyer un email", help_text="Envoyer un email après l'exécution" ) email_recipients = models.TextField( blank=True, verbose_name="Destinataires", help_text="Adresses email séparées par des virgules" ) email_subject = models.CharField( max_length=200, blank=True, verbose_name="Sujet de l'email" ) last_run = models.DateTimeField( null=True, blank=True, verbose_name="Dernière exécution" ) next_run = models.DateTimeField( null=True, blank=True, verbose_name="Prochaine exécution" ) created_at = models.DateTimeField( auto_now_add=True, verbose_name="Créé le" ) created_by = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="created_scheduled_commands", verbose_name="Créé par" ) description = models.TextField( blank=True, verbose_name="Description" ) class Meta: verbose_name = "Commande planifiée" verbose_name_plural = "Commandes planifiées" ordering = ['name'] def __str__(self): return f"{self.name} ({self.get_frequency_display()})" def should_run_now(self): """ Vérifie si la commande doit être exécutée maintenant. """ if not self.is_active: return False now = timezone.now() # Si jamais exécutée ou si next_run est dépassé if self.next_run is None or now >= self.next_run: return True return False def calculate_next_run(self, from_time=None): """ Calcule la prochaine date d'exécution. """ if from_time is None: from_time = timezone.now() next_run = from_time.replace(hour=self.hour, minute=self.minute, second=0, microsecond=0) if self.frequency == 'hourly': # Prochaine heure if next_run <= from_time: next_run = from_time.replace(minute=self.minute, second=0, microsecond=0) if next_run <= from_time: next_run = (next_run + timezone.timedelta(hours=1)) elif self.frequency == 'daily': # Lendemain si l'heure est dépassée aujourd'hui if next_run <= from_time: next_run += timezone.timedelta(days=1) elif self.frequency == 'weekly': # Prochain jour de la semaine spécifié if self.day_of_week is not None: days_ahead = self.day_of_week - next_run.weekday() if days_ahead <= 0 or (days_ahead == 0 and next_run <= from_time): days_ahead += 7 next_run += timezone.timedelta(days=days_ahead) elif self.frequency == 'monthly': # Prochain jour du mois spécifié if self.day_of_month is not None: next_run = next_run.replace(day=min(self.day_of_month, 28)) # Sécurité pour éviter les erreurs if next_run <= from_time: # Mois suivant if next_run.month == 12: next_run = next_run.replace(year=next_run.year + 1, month=1) else: next_run = next_run.replace(month=next_run.month + 1) return next_run def get_command_args_display(self): """Retourne les arguments sous forme lisible.""" if not self.command_args: return "Aucun argument" try: return json.dumps(self.command_args, indent=2) except: return str(self.command_args) class CommandExecutionLog(models.Model): """ Historique d'exécution des commandes planifiées. """ STATUS_CHOICES = [ ('running', 'En cours'), ('success', 'Succès'), ('error', 'Erreur'), ] scheduled_command = models.ForeignKey( ScheduledCommand, on_delete=models.CASCADE, related_name="execution_logs", verbose_name="Commande planifiée" ) started_at = models.DateTimeField( auto_now_add=True, verbose_name="Démarré à" ) finished_at = models.DateTimeField( null=True, blank=True, verbose_name="Terminé à" ) status = models.CharField( max_length=20, choices=STATUS_CHOICES, default='running', verbose_name="Statut" ) output = models.TextField( blank=True, verbose_name="Sortie" ) error_message = models.TextField( blank=True, verbose_name="Message d'erreur" ) duration_seconds = models.FloatField( null=True, blank=True, verbose_name="Durée (secondes)" ) email_sent = models.BooleanField( default=False, verbose_name="Email envoyé" ) class Meta: verbose_name = "Journal d'exécution" verbose_name_plural = "Journaux d'exécution" ordering = ['-started_at'] def __str__(self): return f"{self.scheduled_command.name} - {self.started_at.strftime('%Y-%m-%d %H:%M')} - {self.get_status_display()}" def calculate_duration(self): """Calcule la durée d'exécution.""" if self.finished_at and self.started_at: delta = self.finished_at - self.started_at self.duration_seconds = delta.total_seconds() self.save(update_fields=['duration_seconds'])