601 lines
22 KiB
Python
601 lines
22 KiB
Python
"""
|
|
Admin pour les tâches planifiées.
|
|
"""
|
|
|
|
from django.contrib import admin
|
|
from django.utils.html import format_html
|
|
from django.urls import reverse, path
|
|
from django.utils.safestring import mark_safe
|
|
from django.contrib import messages
|
|
from django.core.management import call_command
|
|
from django.core.mail import EmailMessage
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
from django.shortcuts import redirect
|
|
from django.http import HttpResponseRedirect
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.utils.decorators import method_decorator
|
|
from common.models_scheduled_tasks import ScheduledCommand, CommandExecutionLog
|
|
import json
|
|
import io
|
|
import traceback
|
|
import os
|
|
|
|
|
|
@admin.register(ScheduledCommand)
|
|
class ScheduledCommandAdmin(admin.ModelAdmin):
|
|
list_display = [
|
|
'name',
|
|
'command_name',
|
|
'frequency_display',
|
|
'schedule_display',
|
|
'is_active_display',
|
|
'last_run_display',
|
|
'next_run_display',
|
|
'execution_count',
|
|
'success_rate'
|
|
]
|
|
|
|
list_filter = ['is_active', 'frequency', 'send_email']
|
|
search_fields = ['name', 'command_name', 'description']
|
|
readonly_fields = ['last_run', 'next_run', 'created_at', 'created_by', 'execution_history_link', 'execute_now_button']
|
|
actions = ['execute_immediately']
|
|
|
|
fieldsets = [
|
|
('Informations générales', {
|
|
'fields': ['name', 'description', 'is_active']
|
|
}),
|
|
('Commande', {
|
|
'fields': ['command_name', 'command_args']
|
|
}),
|
|
('Planification', {
|
|
'fields': [
|
|
'frequency',
|
|
'hour',
|
|
'minute',
|
|
'day_of_month',
|
|
'day_of_week',
|
|
]
|
|
}),
|
|
('Notification par email', {
|
|
'fields': ['send_email', 'email_recipients', 'email_subject'],
|
|
'classes': ['collapse']
|
|
}),
|
|
('Exécution', {
|
|
'fields': ['execute_now_button'],
|
|
'description': 'Exécuter cette commande immédiatement, sans attendre la prochaine planification.'
|
|
}),
|
|
('Informations système', {
|
|
'fields': ['last_run', 'next_run', 'created_at', 'created_by', 'execution_history_link'],
|
|
'classes': ['collapse']
|
|
})
|
|
]
|
|
|
|
def get_urls(self):
|
|
"""Ajoute des URLs personnalisées pour l'admin."""
|
|
urls = super().get_urls()
|
|
custom_urls = [
|
|
path(
|
|
'<int:command_id>/execute/',
|
|
csrf_exempt(self.admin_site.admin_view(self.execute_command_view)),
|
|
name='execute_scheduled_command',
|
|
),
|
|
]
|
|
return custom_urls + urls
|
|
|
|
def execute_command_view(self, request, command_id):
|
|
"""Vue pour exécuter une commande immédiatement."""
|
|
# Accepter seulement les requêtes GET (pas de CSRF nécessaire)
|
|
if request.method != 'GET':
|
|
messages.error(request, "Méthode non autorisée.")
|
|
return redirect('admin:common_scheduledcommand_changelist')
|
|
|
|
# Récupérer la commande (hors du try pour pouvoir l'utiliser après)
|
|
try:
|
|
scheduled_command = ScheduledCommand.objects.get(pk=command_id)
|
|
except ScheduledCommand.DoesNotExist:
|
|
messages.error(request, "Commande introuvable.")
|
|
return redirect('admin:common_scheduledcommand_changelist')
|
|
|
|
# Créer un log d'exécution
|
|
log = CommandExecutionLog.objects.create(
|
|
scheduled_command=scheduled_command,
|
|
status='running'
|
|
)
|
|
|
|
try:
|
|
# Préparer les arguments
|
|
args = []
|
|
kwargs = {}
|
|
if scheduled_command.command_args:
|
|
if isinstance(scheduled_command.command_args, dict):
|
|
kwargs = scheduled_command.command_args
|
|
elif isinstance(scheduled_command.command_args, list):
|
|
args = scheduled_command.command_args
|
|
|
|
# Capturer la sortie
|
|
output_buffer = io.StringIO()
|
|
|
|
# Exécuter la commande
|
|
call_command(
|
|
scheduled_command.command_name,
|
|
*args,
|
|
stdout=output_buffer,
|
|
stderr=output_buffer,
|
|
**kwargs
|
|
)
|
|
|
|
# Mettre à jour le log
|
|
log.finished_at = timezone.now()
|
|
log.status = 'success'
|
|
log.output = output_buffer.getvalue()
|
|
log.save()
|
|
|
|
# Mettre à jour la commande planifiée
|
|
scheduled_command.last_run = log.started_at
|
|
scheduled_command.save()
|
|
|
|
# Envoyer l'email si configuré
|
|
if scheduled_command.send_email and scheduled_command.email_recipients:
|
|
self.send_success_email(scheduled_command, log, request)
|
|
|
|
messages.success(
|
|
request,
|
|
f"Commande '{scheduled_command.name}' exécutée avec succès. "
|
|
f"Consultez l'historique pour voir les détails."
|
|
)
|
|
|
|
except Exception as e:
|
|
# Mettre à jour le log avec l'erreur
|
|
log.finished_at = timezone.now()
|
|
log.status = 'error'
|
|
log.output = output_buffer.getvalue()
|
|
log.error_message = f"{str(e)}\n\n{traceback.format_exc()}"
|
|
log.save()
|
|
|
|
# Envoyer l'email d'erreur si configuré
|
|
if scheduled_command.send_email and scheduled_command.email_recipients:
|
|
self.send_error_email(scheduled_command, log, request)
|
|
|
|
messages.error(
|
|
request,
|
|
f"Erreur lors de l'exécution de '{scheduled_command.name}': {str(e)}"
|
|
)
|
|
|
|
# Rediriger vers la page de détail de la commande
|
|
return redirect('admin:common_scheduledcommand_change', scheduled_command.pk)
|
|
|
|
def send_success_email(self, task, log, request):
|
|
"""Envoie un email de succès après exécution depuis l'admin."""
|
|
try:
|
|
# Calculer la durée si pas déjà fait
|
|
if log.duration_seconds is None and log.finished_at and log.started_at:
|
|
log.calculate_duration()
|
|
log.save(update_fields=['duration_seconds'])
|
|
|
|
# Formater la durée
|
|
duration_str = f"{log.duration_seconds:.1f}" if log.duration_seconds is not None else "N/A"
|
|
|
|
subject = task.email_subject or f"[StreetUp] 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 ''
|
|
|
|
# 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,
|
|
|
|
La tâche planifiée "{task.name}" s'est exécutée avec succès depuis l'interface d'administration.
|
|
|
|
Détails:
|
|
- Commande: {task.command_name}
|
|
- Démarré: {log.started_at.strftime('%Y-%m-%d %H:%M:%S')}
|
|
- Terminé: {log.finished_at.strftime('%Y-%m-%d %H:%M:%S')}
|
|
- Durée: {duration_str} secondes
|
|
- Exécuté par: {request.user.username} ({request.user.email})
|
|
|
|
Prochaine exécution planifiée: {task.next_run.strftime('%Y-%m-%d %H:%M') if task.next_run else 'Non planifiée'}
|
|
|
|
---
|
|
Sortie de la commande:
|
|
{log.output}
|
|
|
|
---
|
|
Cordialement,
|
|
StreetUp - Système de tâches planifiées
|
|
"""
|
|
|
|
recipients = [email.strip() for email in task.email_recipients.split(',')]
|
|
|
|
email = EmailMessage(
|
|
subject=subject,
|
|
body=body,
|
|
from_email=settings.DEFAULT_FROM_EMAIL,
|
|
to=recipients,
|
|
)
|
|
|
|
# Attacher le fichier si il existe
|
|
if output_file and os.path.exists(output_file):
|
|
email.attach_file(output_file)
|
|
|
|
email.send(fail_silently=False)
|
|
|
|
log.email_sent = True
|
|
log.save(update_fields=['email_sent'])
|
|
|
|
messages.info(request, f"Email envoyé à: {', '.join(recipients)}")
|
|
|
|
except Exception as e:
|
|
messages.warning(request, f"Erreur lors de l'envoi de l'email: {str(e)}")
|
|
|
|
def send_error_email(self, task, log, request):
|
|
"""Envoie un email d'erreur après échec d'exécution depuis l'admin."""
|
|
try:
|
|
subject = f"[StreetUp] ERREUR - Tâche planifiée: {task.name}"
|
|
|
|
body = f"""Bonjour,
|
|
|
|
La tâche planifiée "{task.name}" a rencontré une erreur lors de son exécution depuis l'interface d'administration.
|
|
|
|
Détails:
|
|
- Commande: {task.command_name}
|
|
- Démarré: {log.started_at.strftime('%Y-%m-%d %H:%M:%S')}
|
|
- Terminé: {log.finished_at.strftime('%Y-%m-%d %H:%M:%S')}
|
|
- Durée: {log.duration_seconds:.1f} secondes
|
|
- Exécuté par: {request.user.username} ({request.user.email})
|
|
|
|
Prochaine exécution planifiée: {task.next_run.strftime('%Y-%m-%d %H:%M') if task.next_run else 'Non planifiée'}
|
|
|
|
---
|
|
Message d'erreur:
|
|
{log.error_message}
|
|
|
|
---
|
|
Sortie de la commande:
|
|
{log.output}
|
|
|
|
---
|
|
Veuillez vérifier les logs dans l'administration Django pour plus de détails.
|
|
|
|
Cordialement,
|
|
StreetUp - Système de tâches planifiées
|
|
"""
|
|
|
|
recipients = [email.strip() for email in task.email_recipients.split(',')]
|
|
|
|
email = EmailMessage(
|
|
subject=subject,
|
|
body=body,
|
|
from_email=settings.DEFAULT_FROM_EMAIL,
|
|
to=recipients,
|
|
)
|
|
|
|
email.send(fail_silently=False)
|
|
|
|
log.email_sent = True
|
|
log.save(update_fields=['email_sent'])
|
|
|
|
messages.warning(request, f"Email d'erreur envoyé à: {', '.join(recipients)}")
|
|
|
|
except Exception as e:
|
|
messages.warning(request, f"Erreur lors de l'envoi de l'email d'erreur: {str(e)}")
|
|
|
|
def save_model(self, request, obj, form, change):
|
|
if not change: # Nouvelle création
|
|
obj.created_by = request.user
|
|
|
|
# Calculer next_run si pas déjà défini
|
|
if not obj.next_run:
|
|
obj.next_run = obj.calculate_next_run()
|
|
|
|
super().save_model(request, obj, form, change)
|
|
|
|
def frequency_display(self, obj):
|
|
"""Affiche la fréquence avec une icône."""
|
|
icons = {
|
|
'hourly': '⏰',
|
|
'daily': '📅',
|
|
'weekly': '📆',
|
|
'monthly': '🗓️'
|
|
}
|
|
icon = icons.get(obj.frequency, '⏱️')
|
|
return f"{icon} {obj.get_frequency_display()}"
|
|
frequency_display.short_description = "Fréquence"
|
|
|
|
def schedule_display(self, obj):
|
|
"""Affiche le planning de manière lisible."""
|
|
schedule = f"{obj.hour:02d}:{obj.minute:02d}"
|
|
|
|
if obj.frequency == 'monthly' and obj.day_of_month:
|
|
schedule += f" le {obj.day_of_month}"
|
|
elif obj.frequency == 'weekly' and obj.day_of_week is not None:
|
|
days = ['Lun', 'Mar', 'Mer', 'Jeu', 'Ven', 'Sam', 'Dim']
|
|
schedule += f" le {days[obj.day_of_week]}"
|
|
|
|
return schedule
|
|
schedule_display.short_description = "Planning"
|
|
|
|
def is_active_display(self, obj):
|
|
"""Affiche le statut avec une icône."""
|
|
if obj.is_active:
|
|
return mark_safe('<span style="color: green;">✓ Active</span>')
|
|
return mark_safe('<span style="color: red;">✗ Inactive</span>')
|
|
is_active_display.short_description = "Statut"
|
|
|
|
def last_run_display(self, obj):
|
|
"""Affiche la dernière exécution."""
|
|
if obj.last_run:
|
|
return obj.last_run.strftime('%Y-%m-%d %H:%M')
|
|
return "-"
|
|
last_run_display.short_description = "Dernière exécution"
|
|
|
|
def next_run_display(self, obj):
|
|
"""Affiche la prochaine exécution."""
|
|
if obj.next_run:
|
|
return obj.next_run.strftime('%Y-%m-%d %H:%M')
|
|
return "-"
|
|
next_run_display.short_description = "Prochaine exécution"
|
|
|
|
def execution_count(self, obj):
|
|
"""Nombre total d'exécutions."""
|
|
return obj.execution_logs.count()
|
|
execution_count.short_description = "Exécutions"
|
|
|
|
def success_rate(self, obj):
|
|
"""Taux de succès."""
|
|
total = obj.execution_logs.count()
|
|
if total == 0:
|
|
return "-"
|
|
|
|
success = obj.execution_logs.filter(status='success').count()
|
|
rate = (success / total) * 100
|
|
rate_formatted = "{:.1f}".format(rate)
|
|
|
|
color = 'green' if rate >= 90 else 'orange' if rate >= 70 else 'red'
|
|
return format_html(
|
|
'<span style="color: {};">{}% ({}/{})</span>',
|
|
color, rate_formatted, success, total
|
|
)
|
|
success_rate.short_description = "Taux de succès"
|
|
|
|
def execution_history_link(self, obj):
|
|
"""Lien vers l'historique d'exécution."""
|
|
if obj.pk:
|
|
count = obj.execution_logs.count()
|
|
url = reverse('admin:common_commandexecutionlog_changelist') + '?scheduled_command__id__exact={}'.format(obj.pk)
|
|
return format_html(
|
|
'<a href="{}" class="button">Voir l\'historique ({} exécutions)</a>',
|
|
url, count
|
|
)
|
|
return "-"
|
|
execution_history_link.short_description = "Historique"
|
|
|
|
def execute_now_button(self, obj):
|
|
"""Bouton pour exécuter la commande immédiatement."""
|
|
if obj.pk:
|
|
url = reverse('admin:execute_scheduled_command', args=[obj.pk])
|
|
return format_html(
|
|
'''
|
|
<div style="padding: 10px; background: #f8f8f8; border-radius: 5px;">
|
|
<button type="button" onclick="if(confirm('Voulez-vous vraiment exécuter cette commande maintenant ?')) {{ window.location.href='{}'; }}"
|
|
class="button" style="background: #417690; color: white; padding: 10px 20px; font-weight: bold; cursor: pointer;">
|
|
▶️ Exécuter maintenant
|
|
</button>
|
|
<p style="margin-top: 10px; color: #666; font-size: 12px;">
|
|
La commande sera exécutée immédiatement dans une tâche en arrière-plan.
|
|
Consultez l'historique d'exécution pour voir le résultat.
|
|
</p>
|
|
</div>
|
|
''',
|
|
url
|
|
)
|
|
return "-"
|
|
execute_now_button.short_description = "Exécution immédiate"
|
|
|
|
def execute_immediately(self, request, queryset):
|
|
"""Action pour exécuter les commandes sélectionnées immédiatement."""
|
|
executed_count = 0
|
|
error_count = 0
|
|
|
|
for scheduled_command in queryset:
|
|
# Créer un log d'exécution
|
|
log = CommandExecutionLog.objects.create(
|
|
scheduled_command=scheduled_command,
|
|
status='running'
|
|
)
|
|
|
|
try:
|
|
# Préparer les arguments
|
|
args = []
|
|
kwargs = {}
|
|
if scheduled_command.command_args:
|
|
if isinstance(scheduled_command.command_args, dict):
|
|
kwargs = scheduled_command.command_args
|
|
elif isinstance(scheduled_command.command_args, list):
|
|
args = scheduled_command.command_args
|
|
|
|
# Capturer la sortie
|
|
output_buffer = io.StringIO()
|
|
|
|
# Exécuter la commande
|
|
call_command(
|
|
scheduled_command.command_name,
|
|
*args,
|
|
stdout=output_buffer,
|
|
stderr=output_buffer,
|
|
**kwargs
|
|
)
|
|
|
|
# Mettre à jour le log
|
|
log.finished_at = timezone.now()
|
|
log.status = 'success'
|
|
log.output = output_buffer.getvalue()
|
|
log.save()
|
|
|
|
# Mettre à jour la commande planifiée
|
|
scheduled_command.last_run = log.started_at
|
|
scheduled_command.save()
|
|
|
|
executed_count += 1
|
|
|
|
except Exception as e:
|
|
# Mettre à jour le log avec l'erreur
|
|
log.finished_at = timezone.now()
|
|
log.status = 'error'
|
|
log.error_message = f"{str(e)}\n\n{traceback.format_exc()}"
|
|
log.save()
|
|
|
|
error_count += 1
|
|
|
|
# Messages de retour
|
|
if executed_count > 0:
|
|
self.message_user(
|
|
request,
|
|
f"{executed_count} commande(s) exécutée(s) avec succès.",
|
|
messages.SUCCESS
|
|
)
|
|
|
|
if error_count > 0:
|
|
self.message_user(
|
|
request,
|
|
f"{error_count} commande(s) en erreur. Consultez l'historique pour les détails.",
|
|
messages.ERROR
|
|
)
|
|
|
|
execute_immediately.short_description = "▶️ Exécuter immédiatement les commandes sélectionnées"
|
|
|
|
|
|
@admin.register(CommandExecutionLog)
|
|
class CommandExecutionLogAdmin(admin.ModelAdmin):
|
|
list_display = [
|
|
'scheduled_command',
|
|
'started_at_display',
|
|
'status_display',
|
|
'duration_display',
|
|
'email_sent_display'
|
|
]
|
|
|
|
list_filter = ['status', 'email_sent', 'scheduled_command']
|
|
search_fields = ['scheduled_command__name', 'output', 'error_message']
|
|
readonly_fields = [
|
|
'scheduled_command',
|
|
'started_at',
|
|
'finished_at',
|
|
'status',
|
|
'output',
|
|
'error_message',
|
|
'duration_seconds',
|
|
'email_sent',
|
|
'output_formatted',
|
|
'error_formatted'
|
|
]
|
|
|
|
fieldsets = [
|
|
('Exécution', {
|
|
'fields': [
|
|
'scheduled_command',
|
|
'started_at',
|
|
'finished_at',
|
|
'status',
|
|
'duration_seconds',
|
|
'email_sent'
|
|
]
|
|
}),
|
|
('Sortie', {
|
|
'fields': ['output_formatted'],
|
|
}),
|
|
('Erreurs', {
|
|
'fields': ['error_formatted'],
|
|
'classes': ['collapse']
|
|
})
|
|
]
|
|
|
|
def has_add_permission(self, request):
|
|
"""Empêche la création manuelle de logs."""
|
|
return False
|
|
|
|
def has_change_permission(self, request, obj=None):
|
|
"""Lecture seule."""
|
|
return True
|
|
|
|
def has_delete_permission(self, request, obj=None):
|
|
"""Permet la suppression."""
|
|
return True
|
|
|
|
def started_at_display(self, obj):
|
|
"""Affiche la date de début."""
|
|
return obj.started_at.strftime('%Y-%m-%d %H:%M:%S')
|
|
started_at_display.short_description = "Démarré à"
|
|
|
|
def status_display(self, obj):
|
|
"""Affiche le statut avec une couleur."""
|
|
colors = {
|
|
'running': 'blue',
|
|
'success': 'green',
|
|
'error': 'red'
|
|
}
|
|
icons = {
|
|
'running': '⏳',
|
|
'success': '✓',
|
|
'error': '✗'
|
|
}
|
|
color = colors.get(obj.status, 'gray')
|
|
icon = icons.get(obj.status, '•')
|
|
|
|
return format_html(
|
|
'<span style="color: {};">{} {}</span>',
|
|
color, icon, obj.get_status_display()
|
|
)
|
|
status_display.short_description = "Statut"
|
|
|
|
def duration_display(self, obj):
|
|
"""Affiche la durée."""
|
|
if obj.duration_seconds is not None:
|
|
if obj.duration_seconds < 60:
|
|
return f"{obj.duration_seconds:.1f}s"
|
|
else:
|
|
minutes = int(obj.duration_seconds // 60)
|
|
seconds = int(obj.duration_seconds % 60)
|
|
return f"{minutes}m {seconds}s"
|
|
return "-"
|
|
duration_display.short_description = "Durée"
|
|
|
|
def email_sent_display(self, obj):
|
|
"""Affiche si l'email a été envoyé."""
|
|
if obj.email_sent:
|
|
return mark_safe('<span style="color: green;">✓</span>')
|
|
return mark_safe('<span style="color: gray;">-</span>')
|
|
email_sent_display.short_description = "Email"
|
|
|
|
def output_formatted(self, obj):
|
|
"""Affiche la sortie formatée."""
|
|
if obj.output:
|
|
return format_html(
|
|
'<pre style="background: #f5f5f5; padding: 10px; border-radius: 5px; max-height: 400px; overflow: auto;">{}</pre>',
|
|
obj.output
|
|
)
|
|
return "-"
|
|
output_formatted.short_description = "Sortie de la commande"
|
|
|
|
def error_formatted(self, obj):
|
|
"""Affiche les erreurs formatées."""
|
|
if obj.error_message:
|
|
return format_html(
|
|
'<pre style="background: #fee; padding: 10px; border-radius: 5px; color: #c00; max-height: 400px; overflow: auto;">{}</pre>',
|
|
obj.error_message
|
|
)
|
|
return "-"
|
|
error_formatted.short_description = "Message d'erreur"
|