refactor: move output file resolution to model method with log-based fallback and added unit tests

loko/common/admin_scheduled_tasks.py

@@ -180,22 +180,7 @@ class ScheduledCommandAdmin(admin.ModelAdmin):
             subject = task.email_subject or f"[{prod_name}] Tâche planifiée réussie: {task.name}"
             
             # Vérifier si un fichier a été généré
-            output_file = task.command_args.get('output', '') if isinstance(task.command_args, dict) else ''
+            output_file = task.resolve_output_file(log=log)
-            
-            # Substituer les variables dans le nom du fichier
-            if output_file and '%(year)s' in output_file:
-                from datetime import date
-                from dateutil.relativedelta import relativedelta
-                today = date.today()
-                prev_month = today.replace(day=1) - relativedelta(months=1)
-                output_file = output_file.replace('%(year)s', str(prev_month.year))
-                output_file = output_file.replace('%(month)s', f'{prev_month.month:02d}')
-                output_file = output_file.replace('%(day)s', f'{today.day:02d}')
-                output_file = output_file.replace('%(date)s', today.strftime('%Y-%m-%d'))
-                
-                # Convertir en chemin absolu
-                if not os.path.isabs(output_file):
-                    output_file = os.path.join(settings.BASE_DIR, output_file)
             
             body = f"""Bonjour,
 

loko/common/management/commands/run_scheduled_commands.py

@@ -140,7 +140,7 @@ class Command(BaseCommand):
             subject = task.email_subject or f"[{prod_name}] Tâche planifiée réussie: {task.name}"
             
             # Vérifier si un fichier a été généré
-            output_file = task.command_args.get('output', '')
+            output_file = task.resolve_output_file(log=log)
             
             body = f"""
 Bonjour,

loko/common/models_scheduled_tasks.py

@@ -7,6 +7,10 @@ from django.conf import settings
 from django.utils import timezone
 from django.core.validators import MinValueValidator, MaxValueValidator
 import json
+import os
+import re
+from datetime import date
+from dateutil.relativedelta import relativedelta
 
 
 class ScheduledCommand(models.Model):
@@ -216,6 +220,47 @@ class ScheduledCommand(models.Model):
         except:
             return str(self.command_args)
 
+    def resolve_output_file(self, log=None):
+        """
+        Résout le chemin du fichier généré par la commande (s'il existe).
+        Gère les variables temporelles (%(year)s, %(month)s, %(day)s, %(date)s)
+        et convertit les chemins relatifs par rapport à settings.BASE_DIR.
+        Inspecte également log.output en repli si le fichier n'est pas trouvé directement.
+        """
+        output_file = ''
+        if isinstance(self.command_args, dict):
+            output_file = self.command_args.get('output', '')
+        
+        if output_file:
+            today = date.today()
+            # Par cohérence avec les exports (ex: export_interventions_xlsx),
+            # pour le mois et l'année, on utilise le mois précédent par défaut
+            prev_month = today.replace(day=1) - relativedelta(months=1)
+            
+            output_file = str(output_file)
+            output_file = output_file.replace('%(year)s', str(prev_month.year))
+            output_file = output_file.replace('%(month)s', f'{prev_month.month:02d}')
+            output_file = output_file.replace('%(day)s', f'{today.day:02d}')
+            output_file = output_file.replace('%(date)s', today.strftime('%Y-%m-%d'))
+
+            if not os.path.isabs(output_file):
+                output_file = os.path.join(settings.BASE_DIR, output_file)
+            
+            if os.path.exists(output_file):
+                return output_file
+
+        # Repli : analyse de log.output pour trouver un fichier généré
+        if log and getattr(log, 'output', None):
+            match = re.search(r'(?:Fichier généré|Fichier de sortie)\s*:\s*(.+)', log.output)
+            if match:
+                candidate = match.group(1).strip()
+                if not os.path.isabs(candidate):
+                    candidate = os.path.join(settings.BASE_DIR, candidate)
+                if os.path.exists(candidate):
+                    return candidate
+
+        return output_file if (output_file and os.path.exists(output_file)) else None
+
 
 class CommandExecutionLog(models.Model):
     """

loko/common/tests.py

@@ -3,6 +3,7 @@ from django.contrib.messages import get_messages
 from django.test import TestCase
 from django.urls import reverse
 from django.conf import settings
+from django.utils import timezone
 from django.utils.translation import gettext as _
 
 import json
@@ -317,6 +318,80 @@ class UserConfigCopyTests(TestCase):
         self.assertTrue(target_ut.can_inspect)
 
 
+class ScheduledCommandResolveOutputFileTests(TestCase):
+    def test_resolve_output_file_with_placeholders(self):
+        import os
+        from unittest.mock import patch
+        from datetime import date
+        from dateutil.relativedelta import relativedelta
+        from common.models_scheduled_tasks import ScheduledCommand
+
+        cmd = ScheduledCommand(
+            name="Export test",
+            command_name="export_interventions_xlsx",
+            command_args={"output": "exports/interventions_export_%(year)s_%(month)s.xlsx"}
+        )
+
+        today = date.today()
+        prev_month = today.replace(day=1) - relativedelta(months=1)
+        expected_filename = f"interventions_export_{prev_month.year}_{prev_month.month:02d}.xlsx"
+        expected_path = os.path.join(settings.BASE_DIR, "exports", expected_filename)
+
+        with patch("os.path.exists", side_effect=lambda p: p == expected_path):
+            resolved = cmd.resolve_output_file()
+            self.assertEqual(resolved, expected_path)
+
+    def test_resolve_output_file_fallback_from_log_output(self):
+        import os
+        from unittest.mock import patch, MagicMock
+        from common.models_scheduled_tasks import ScheduledCommand
+
+        cmd = ScheduledCommand(
+            name="Export test",
+            command_name="export_interventions_xlsx",
+            command_args={"output": "unknown_file.xlsx"}
+        )
+
+        log = MagicMock()
+        log.output = "Démarrage...\n✓ Fichier généré: /tmp/custom_generated.xlsx\n"
+
+        with patch("os.path.exists", side_effect=lambda p: p == "/tmp/custom_generated.xlsx"):
+            resolved = cmd.resolve_output_file(log=log)
+            self.assertEqual(resolved, "/tmp/custom_generated.xlsx")
+
+    def test_send_success_email_attaches_resolved_file(self):
+        import os
+        from unittest.mock import patch, MagicMock
+        from common.models_scheduled_tasks import ScheduledCommand
+        from common.management.commands.run_scheduled_commands import Command as RunScheduledCommand
+
+        cmd = ScheduledCommand.objects.create(
+            name="Export mensuel",
+            command_name="export_interventions_xlsx",
+            command_args={"output": "exports/interventions_export_%(year)s_%(month)s.xlsx"},
+            send_email=True,
+            email_recipients="test@example.com"
+        )
+
+        log = MagicMock()
+        log.started_at = timezone.now()
+        log.finished_at = timezone.now()
+        log.duration_seconds = 1.0
+        log.output = "Succès"
+
+        fake_path = os.path.join(settings.BASE_DIR, "exports", "test.xlsx")
+
+        with patch.object(ScheduledCommand, "resolve_output_file", return_value=fake_path), \
+             patch("os.path.exists", return_value=True), \
+             patch("django.core.mail.EmailMessage.attach_file") as mock_attach, \
+             patch("django.core.mail.EmailMessage.send") as mock_send:
+            runner = RunScheduledCommand()
+            runner.send_success_email(cmd, log)
+
+            mock_attach.assert_called_once_with(fake_path)
+            mock_send.assert_called_once()
+
+