from django.contrib import admin, messages from django.db.models import Avg, Count, Q from django.utils import timezone from unfold.admin import ModelAdmin, TabularInline from unfold.contrib.filters.admin import RangeDateTimeFilter from unfold.decorators import display from core.admin_ui import PortalAdminMixin from core.status_palette import MONITOR_CHECK_LABELS, MONITOR_INCIDENT_LABELS, MONITOR_STATUS_LABELS from .models import MonitorCheck, MonitorIncident, WebsiteMonitor from .services import run_monitor_check class RecentCheckInline(TabularInline): model = MonitorCheck extra = 0 max_num = 10 can_delete = False fields = ("checked_at", "result", "http_status", "response_time_ms", "ssl_days_remaining", "error_message") readonly_fields = fields tab = True verbose_name_plural = "Últimas verificações" def has_add_permission(self, request, obj=None): return False @admin.action(description="Executar monitorização agora") def run_now(modeladmin, request, queryset): success = 0 failed = 0 for monitor in queryset[:25]: try: run_monitor_check(monitor, triggered_by=request.user) success += 1 except Exception as exc: failed += 1 modeladmin.message_user(request, f"{monitor}: {exc}", level=messages.ERROR) if success: modeladmin.message_user(request, f"{success} monitor(es) verificado(s).", level=messages.SUCCESS) if queryset.count() > 25: modeladmin.message_user(request, "Por segurança, a ação processou apenas os primeiros 25 monitores.", level=messages.WARNING) if failed: modeladmin.message_user(request, f"{failed} verificação(ões) falharam ao executar.", level=messages.WARNING) @admin.action(description="Ativar monitores selecionados") def activate_monitors(modeladmin, request, queryset): updated = queryset.update(is_active=True, status=WebsiteMonitor.Status.PENDING, next_check_at=timezone.now()) modeladmin.message_user(request, f"{updated} monitor(es) ativado(s).", level=messages.SUCCESS) @admin.action(description="Pausar monitores selecionados") def pause_monitors(modeladmin, request, queryset): updated = queryset.update(is_active=False, status=WebsiteMonitor.Status.PAUSED, next_check_at=None) modeladmin.message_user(request, f"{updated} monitor(es) pausado(s).", level=messages.SUCCESS) @admin.register(WebsiteMonitor) class WebsiteMonitorAdmin(PortalAdminMixin, ModelAdmin): portal_icon = "monitor_heart" portal_kicker = "MONITORIZAÇÃO" portal_description = "Disponibilidade, tempos de resposta, SSL, alertas e automatização de incidentes dos websites geridos." portal_tone = "navy" portal_stats = ( {"label": "Monitores", "icon": "monitor_heart", "tone": "primary", "caption": "Total configurado"}, {"label": "Operacionais", "icon": "check_circle", "tone": "success", "filters": {"status": WebsiteMonitor.Status.UP}, "caption": "Sem falhas"}, {"label": "Degradados", "icon": "speed", "tone": "warning", "filters": {"status": WebsiteMonitor.Status.DEGRADED}, "caption": "Requer atenção"}, {"label": "Indisponíveis", "icon": "cloud_off", "tone": "danger", "filters": {"status": WebsiteMonitor.Status.DOWN}, "caption": "Incidente ativo"}, ) portal_related_links = ( {"label": "Centro de monitorização", "icon": "space_dashboard", "url_name": "monitoring_center"}, {"label": "Verificações", "icon": "fact_check", "url_name": "admin:monitoring_monitorcheck_changelist"}, {"label": "Incidentes", "icon": "crisis_alert", "url_name": "admin:monitoring_monitorincident_changelist"}, {"label": "Ativos técnicos", "icon": "developer_board", "url_name": "admin:technical_technicalasset_changelist"}, ) list_display = ( "name", "technical_asset", "url_link", "show_status", "last_http_status", "show_response", "show_uptime", "last_checked_at", "next_check_at", "is_active", ) list_filter = ( "status", "is_active", "client_visible", "http_method", "interval_minutes", "verify_ssl", "notify_client", ("last_checked_at", RangeDateTimeFilter), ) search_fields = ( "name", "url", "technical_asset__label", "technical_asset__domain_name", "technical_asset__service__name", "technical_asset__service__client__name", ) autocomplete_fields = ("technical_asset",) readonly_fields = ( "public_id", "status", "last_checked_at", "next_check_at", "last_success_at", "last_failure_at", "last_response_ms", "last_http_status", "last_error", "consecutive_failures", "consecutive_successes", "created_at", "updated_at", ) actions = (run_now, activate_monitors, pause_monitors) inlines = (RecentCheckInline,) list_fullwidth = True list_filter_submit = True fieldsets = ( ("Monitor", {"fields": ("technical_asset", "name", "url", "is_active", "client_visible", "status"), "description": "Website, URL e visibilidade para o cliente."}), ("Pedido HTTP", {"fields": ("http_method", "interval_minutes", "timeout_seconds", "expected_status_codes", "expected_text", "follow_redirects"), "description": "Critérios usados para validar a resposta."}), ("Desempenho e SSL", {"fields": ("warning_response_ms", "verify_ssl", "inspect_ssl_certificate", "ssl_warning_days", "allow_private_networks")}), ("Incidentes e alertas", {"fields": ("alert_after_failures", "recover_after_successes", "create_incident", "create_task", "notify_internal", "notify_client", "maintenance_until")}), ("Último resultado", {"fields": ("last_checked_at", "next_check_at", "last_success_at", "last_failure_at", "last_response_ms", "last_http_status", "last_error", "consecutive_failures", "consecutive_successes")}), ("Sistema", {"fields": ("public_id", "created_at", "updated_at"), "classes": ("collapse",)}), ) @display(description="Estado", label=MONITOR_STATUS_LABELS, ordering="status") def show_status(self, obj): return obj.get_status_display() @display(description="URL") def url_link(self, obj): return obj.url @display(description="Resposta", label=True, ordering="last_response_ms") def show_response(self, obj): if obj.last_response_ms is None: return "Por medir", "info" if obj.last_response_ms > obj.warning_response_ms: return f"{obj.last_response_ms} ms", "warning" return f"{obj.last_response_ms} ms", "success" @display(description="Uptime 30d", label=True) def show_uptime(self, obj): uptime = obj.uptime_percentage(days=30) if uptime is None: return "Sem dados", "info" if uptime >= 99.9: return f"{uptime:.2f}%", "success" if uptime >= 98: return f"{uptime:.2f}%", "warning" return f"{uptime:.2f}%", "danger" @admin.register(MonitorCheck) class MonitorCheckAdmin(PortalAdminMixin, ModelAdmin): portal_icon = "fact_check" portal_kicker = "MONITORIZAÇÃO" portal_description = "Histórico imutável das verificações HTTP, tempos de resposta, SSL, DNS e erros detetados." portal_tone = "cyan" portal_stats = ( {"label": "Verificações", "icon": "fact_check", "tone": "primary", "caption": "Histórico total"}, {"label": "Operacionais", "icon": "check_circle", "tone": "success", "filters": {"result": MonitorCheck.Result.UP}, "caption": "Resposta válida"}, {"label": "Degradadas", "icon": "speed", "tone": "warning", "filters": {"result": MonitorCheck.Result.DEGRADED}, "caption": "Lentas ou SSL próximo"}, {"label": "Falhas", "icon": "cancel", "tone": "danger", "filters": {"result": MonitorCheck.Result.DOWN}, "caption": "Indisponibilidade"}, ) portal_related_links = ( {"label": "Monitores", "icon": "monitor_heart", "url_name": "admin:monitoring_websitemonitor_changelist"}, {"label": "Incidentes", "icon": "crisis_alert", "url_name": "admin:monitoring_monitorincident_changelist"}, ) list_display = ( "monitor", "checked_at", "show_result", "http_status", "response_time_ms", "resolved_ip", "ssl_days_remaining", "error_type", ) list_filter = ("result", "error_type", "ssl_valid", ("checked_at", RangeDateTimeFilter)) search_fields = ("monitor__name", "monitor__url", "monitor__technical_asset__service__client__name", "error_message", "resolved_ip") readonly_fields = ( "monitor", "checked_at", "result", "http_status", "response_time_ms", "final_url", "redirect_count", "resolved_ip", "response_size_bytes", "content_match", "ssl_valid", "ssl_issuer", "ssl_expires_at", "ssl_days_remaining", "error_type", "error_message", "details", ) list_fullwidth = True fieldsets = ( ("Resultado", {"fields": ("monitor", "checked_at", "result", "http_status", "response_time_ms", "final_url", "redirect_count")}), ("Rede e conteúdo", {"fields": ("resolved_ip", "response_size_bytes", "content_match")}), ("SSL", {"fields": ("ssl_valid", "ssl_issuer", "ssl_expires_at", "ssl_days_remaining")}), ("Erro e diagnóstico", {"fields": ("error_type", "error_message", "details")}), ) def has_add_permission(self, request): return False def has_change_permission(self, request, obj=None): return False def has_delete_permission(self, request, obj=None): return request.user.is_superuser @display(description="Resultado", label=MONITOR_CHECK_LABELS, ordering="result") def show_result(self, obj): return obj.get_result_display() @admin.action(description="Reconhecer incidentes selecionados") def acknowledge_incidents(modeladmin, request, queryset): count = 0 for incident in queryset: if incident.status == MonitorIncident.Status.OPEN: incident.acknowledge(user=request.user) count += 1 modeladmin.message_user(request, f"{count} incidente(s) reconhecido(s).", level=messages.SUCCESS) @admin.action(description="Marcar incidentes como resolvidos") def resolve_incidents(modeladmin, request, queryset): count = 0 for incident in queryset.exclude(status__in=[MonitorIncident.Status.RESOLVED, MonitorIncident.Status.IGNORED]): incident.resolve(user=request.user, note="Resolvido manualmente no backoffice.") count += 1 modeladmin.message_user(request, f"{count} incidente(s) resolvido(s).", level=messages.SUCCESS) @admin.register(MonitorIncident) class MonitorIncidentAdmin(PortalAdminMixin, ModelAdmin): portal_icon = "crisis_alert" portal_kicker = "MONITORIZAÇÃO" portal_description = "Incidentes automáticos, reconhecimento, resolução, duração da indisponibilidade e tarefas técnicas associadas." portal_tone = "coral" portal_stats = ( {"label": "Incidentes", "icon": "crisis_alert", "tone": "primary", "caption": "Histórico total"}, {"label": "Abertos", "icon": "error", "tone": "danger", "filters": {"status": MonitorIncident.Status.OPEN}, "caption": "Requer ação"}, {"label": "Reconhecidos", "icon": "visibility", "tone": "warning", "filters": {"status": MonitorIncident.Status.ACKNOWLEDGED}, "caption": "Em investigação"}, {"label": "Resolvidos", "icon": "task_alt", "tone": "success", "filters": {"status": MonitorIncident.Status.RESOLVED}, "caption": "Recuperados"}, ) portal_related_links = ( {"label": "Centro de monitorização", "icon": "space_dashboard", "url_name": "monitoring_center"}, {"label": "Monitores", "icon": "monitor_heart", "url_name": "admin:monitoring_websitemonitor_changelist"}, {"label": "Tarefas", "icon": "task_alt", "url_name": "admin:operations_internaltask_changelist"}, ) list_display = ( "title", "monitor", "show_severity", "show_status", "opened_at", "show_duration", "failure_count", "internal_task", "client_visible", ) list_filter = ("status", "severity", "client_visible", ("opened_at", RangeDateTimeFilter), ("resolved_at", RangeDateTimeFilter)) search_fields = ("title", "reason", "monitor__name", "monitor__url", "monitor__technical_asset__service__client__name") autocomplete_fields = ("monitor", "internal_task", "acknowledged_by", "resolved_by") readonly_fields = ("public_id", "opened_at", "acknowledged_at", "resolved_at", "first_check", "last_check", "failure_count", "created_at", "updated_at") actions = (acknowledge_incidents, resolve_incidents) list_fullwidth = True fieldsets = ( ("Incidente", {"fields": ("monitor", "title", "reason", "severity", "status", "client_visible")}), ("Cronologia", {"fields": ("opened_at", "acknowledged_at", "acknowledged_by", "resolved_at", "resolved_by", "failure_count")}), ("Relações", {"fields": ("first_check", "last_check", "internal_task")}), ("Resolução", {"fields": ("resolution_note",)}), ("Sistema", {"fields": ("public_id", "created_at", "updated_at"), "classes": ("collapse",)}), ) @display(description="Gravidade", label={ MonitorIncident.Severity.WARNING: "warning", MonitorIncident.Severity.CRITICAL: "danger", }, ordering="severity") def show_severity(self, obj): return obj.get_severity_display() @display(description="Estado", label=MONITOR_INCIDENT_LABELS, ordering="status") def show_status(self, obj): return obj.get_status_display() @display(description="Duração") def show_duration(self, obj): seconds = obj.duration_seconds if seconds < 60: return f"{seconds}s" if seconds < 3600: return f"{seconds // 60} min" return f"{seconds // 3600}h {(seconds % 3600) // 60}m"