666 lines
19 KiB
Python
666 lines
19 KiB
Python
"""
|
||
Analytics app admin configuration.
|
||
"""
|
||
|
||
from django.contrib import admin
|
||
from django.utils.html import format_html
|
||
from django.urls import reverse
|
||
from django.utils.safestring import mark_safe
|
||
from decimal import Decimal
|
||
from datetime import datetime, timedelta
|
||
from .models import (
|
||
Dashboard, DashboardWidget, DataSource, Report,
|
||
ReportExecution, MetricDefinition, MetricValue
|
||
)
|
||
|
||
|
||
class DashboardWidgetInline(admin.TabularInline):
|
||
"""
|
||
Inline admin for dashboard widgets.
|
||
"""
|
||
model = DashboardWidget
|
||
extra = 0
|
||
fields = [
|
||
'name', 'widget_type', 'chart_type', 'data_source',
|
||
'position_x', 'position_y', 'width', 'height', 'is_active'
|
||
]
|
||
readonly_fields = ['widget_id']
|
||
|
||
|
||
@admin.register(Dashboard)
|
||
class DashboardAdmin(admin.ModelAdmin):
|
||
"""
|
||
Admin interface for dashboards.
|
||
"""
|
||
list_display = [
|
||
'name', 'dashboard_type',
|
||
'is_public', 'is_default', 'is_active',
|
||
'refresh_interval_display', 'created_at'
|
||
]
|
||
list_filter = [
|
||
'tenant', 'dashboard_type', 'is_public', 'is_default', 'is_active'
|
||
]
|
||
search_fields = [
|
||
'name', 'description'
|
||
]
|
||
readonly_fields = [
|
||
'dashboard_id',
|
||
'created_at', 'updated_at'
|
||
]
|
||
fieldsets = [
|
||
('Dashboard Information', {
|
||
'fields': [
|
||
'dashboard_id', 'tenant', 'name', 'description'
|
||
]
|
||
}),
|
||
('Dashboard Type', {
|
||
'fields': [
|
||
'dashboard_type'
|
||
]
|
||
}),
|
||
('Layout Configuration', {
|
||
'fields': [
|
||
'layout_config', 'refresh_interval'
|
||
]
|
||
}),
|
||
('Access Control', {
|
||
'fields': [
|
||
'is_public', 'allowed_users', 'allowed_roles'
|
||
]
|
||
}),
|
||
('Dashboard Status', {
|
||
'fields': [
|
||
'is_active', 'is_default'
|
||
]
|
||
}),
|
||
# ('Summary Information', {
|
||
# 'fields': [
|
||
# 'widget_count'
|
||
# ],
|
||
# 'classes': ['collapse']
|
||
# }),
|
||
('Metadata', {
|
||
'fields': [
|
||
'created_at', 'updated_at', 'created_by'
|
||
],
|
||
'classes': ['collapse']
|
||
})
|
||
]
|
||
inlines = [DashboardWidgetInline]
|
||
filter_horizontal = ['allowed_users']
|
||
|
||
# def widget_count_display(self, obj):
|
||
# """Display widget count."""
|
||
# return obj.value_count
|
||
# widget_count_display.short_description = 'Widgets'
|
||
|
||
def refresh_interval_display(self, obj):
|
||
"""Display refresh interval."""
|
||
if obj.refresh_interval >= 3600:
|
||
hours = obj.refresh_interval // 3600
|
||
return f"{hours}h"
|
||
elif obj.refresh_interval >= 60:
|
||
minutes = obj.refresh_interval // 60
|
||
return f"{minutes}m"
|
||
return f"{obj.refresh_interval}s"
|
||
refresh_interval_display.short_description = 'Refresh'
|
||
|
||
def get_queryset(self, request):
|
||
"""Filter by user's tenant."""
|
||
qs = super().get_queryset(request)
|
||
if hasattr(request.user, 'tenant'):
|
||
qs = qs.filter(tenant=request.user.tenant)
|
||
return qs.prefetch_related('widgets')
|
||
|
||
|
||
@admin.register(DashboardWidget)
|
||
class DashboardWidgetAdmin(admin.ModelAdmin):
|
||
"""
|
||
Admin interface for dashboard widgets.
|
||
"""
|
||
list_display = [
|
||
'name', 'dashboard_name', 'widget_type', 'chart_type',
|
||
'data_source', 'position_display', 'size_display',
|
||
'auto_refresh', 'is_active'
|
||
]
|
||
list_filter = [
|
||
'dashboard__tenant', 'widget_type', 'chart_type',
|
||
'auto_refresh', 'is_active'
|
||
]
|
||
search_fields = [
|
||
'name', 'description', 'dashboard__name'
|
||
]
|
||
readonly_fields = [
|
||
'widget_id', 'created_at', 'updated_at'
|
||
]
|
||
fieldsets = [
|
||
('Widget Information', {
|
||
'fields': [
|
||
'widget_id', 'dashboard', 'name', 'description'
|
||
]
|
||
}),
|
||
('Widget Type', {
|
||
'fields': [
|
||
'widget_type', 'chart_type'
|
||
]
|
||
}),
|
||
('Data Source', {
|
||
'fields': [
|
||
'data_source', 'query_config'
|
||
]
|
||
}),
|
||
('Layout', {
|
||
'fields': [
|
||
'position_x', 'position_y', 'width', 'height'
|
||
]
|
||
}),
|
||
('Display Configuration', {
|
||
'fields': [
|
||
'display_config', 'color_scheme'
|
||
]
|
||
}),
|
||
('Refresh Settings', {
|
||
'fields': [
|
||
'auto_refresh', 'refresh_interval'
|
||
]
|
||
}),
|
||
('Widget Status', {
|
||
'fields': [
|
||
'is_active'
|
||
]
|
||
}),
|
||
('Metadata', {
|
||
'fields': [
|
||
'created_at', 'updated_at'
|
||
],
|
||
'classes': ['collapse']
|
||
})
|
||
]
|
||
|
||
def dashboard_name(self, obj):
|
||
"""Display dashboard name."""
|
||
return obj.dashboard.name
|
||
dashboard_name.short_description = 'Dashboard'
|
||
|
||
def position_display(self, obj):
|
||
"""Display position."""
|
||
return f"({obj.position_x}, {obj.position_y})"
|
||
position_display.short_description = 'Position'
|
||
|
||
def size_display(self, obj):
|
||
"""Display size."""
|
||
return f"{obj.width}×{obj.height}"
|
||
size_display.short_description = 'Size'
|
||
|
||
def get_queryset(self, request):
|
||
"""Filter by user's tenant."""
|
||
qs = super().get_queryset(request)
|
||
if hasattr(request.user, 'tenant'):
|
||
qs = qs.filter(dashboard__tenant=request.user.tenant)
|
||
return qs.select_related('dashboard', 'data_source')
|
||
|
||
|
||
@admin.register(DataSource)
|
||
class DataSourceAdmin(admin.ModelAdmin):
|
||
"""
|
||
Admin interface for data sources.
|
||
"""
|
||
list_display = [
|
||
'name', 'source_type', 'connection_type',
|
||
'health_status_display', 'last_health_check',
|
||
'cache_duration_display', 'is_active'
|
||
]
|
||
list_filter = [
|
||
'tenant', 'source_type', 'connection_type',
|
||
'is_healthy', 'is_active'
|
||
]
|
||
search_fields = [
|
||
'name', 'description'
|
||
]
|
||
readonly_fields = [
|
||
'source_id', 'is_healthy', 'last_health_check',
|
||
'created_at', 'updated_at'
|
||
]
|
||
fieldsets = [
|
||
('Data Source Information', {
|
||
'fields': [
|
||
'source_id', 'tenant', 'name', 'description'
|
||
]
|
||
}),
|
||
('Source Configuration', {
|
||
'fields': [
|
||
'source_type', 'connection_type'
|
||
]
|
||
}),
|
||
('Connection Details', {
|
||
'fields': [
|
||
'connection_config', 'authentication_config'
|
||
]
|
||
}),
|
||
('Query Configuration', {
|
||
'fields': [
|
||
'query_template', 'parameters'
|
||
]
|
||
}),
|
||
('Data Processing', {
|
||
'fields': [
|
||
'data_transformation', 'cache_duration'
|
||
]
|
||
}),
|
||
('Health Monitoring', {
|
||
'fields': [
|
||
'is_healthy', 'last_health_check', 'health_check_interval'
|
||
]
|
||
}),
|
||
('Source Status', {
|
||
'fields': [
|
||
'is_active'
|
||
]
|
||
}),
|
||
('Metadata', {
|
||
'fields': [
|
||
'created_at', 'updated_at', 'created_by'
|
||
],
|
||
'classes': ['collapse']
|
||
})
|
||
]
|
||
|
||
def health_status_display(self, obj):
|
||
"""Display health status with color coding."""
|
||
if obj.is_healthy:
|
||
return format_html('<span style="color: green;">✓ Healthy</span>')
|
||
return format_html('<span style="color: red;">✗ Unhealthy</span>')
|
||
health_status_display.short_description = 'Health'
|
||
|
||
def cache_duration_display(self, obj):
|
||
"""Display cache duration."""
|
||
if obj.cache_duration >= 3600:
|
||
hours = obj.cache_duration // 3600
|
||
return f"{hours}h"
|
||
elif obj.cache_duration >= 60:
|
||
minutes = obj.cache_duration // 60
|
||
return f"{minutes}m"
|
||
return f"{obj.cache_duration}s"
|
||
cache_duration_display.short_description = 'Cache'
|
||
|
||
def get_queryset(self, request):
|
||
"""Filter by user's tenant."""
|
||
qs = super().get_queryset(request)
|
||
if hasattr(request.user, 'tenant'):
|
||
qs = qs.filter(tenant=request.user.tenant)
|
||
return qs
|
||
|
||
|
||
class ReportExecutionInline(admin.TabularInline):
|
||
"""
|
||
Inline admin for report executions.
|
||
"""
|
||
model = ReportExecution
|
||
extra = 0
|
||
fields = [
|
||
'execution_type', 'started_at', 'status',
|
||
'duration_seconds', 'record_count'
|
||
]
|
||
readonly_fields = [
|
||
'execution_id', 'started_at', 'completed_at',
|
||
'duration_seconds', 'record_count'
|
||
]
|
||
|
||
|
||
@admin.register(Report)
|
||
class ReportAdmin(admin.ModelAdmin):
|
||
"""
|
||
Admin interface for reports.
|
||
"""
|
||
list_display = [
|
||
'name', 'report_type', 'output_format',
|
||
'schedule_type', 'next_execution',
|
||
'execution_count_display', 'is_active'
|
||
]
|
||
list_filter = [
|
||
'tenant', 'report_type', 'output_format',
|
||
'schedule_type', 'is_active'
|
||
]
|
||
search_fields = [
|
||
'name', 'description'
|
||
]
|
||
readonly_fields = [
|
||
'report_id',
|
||
'created_at', 'updated_at'
|
||
]
|
||
fieldsets = [
|
||
('Report Information', {
|
||
'fields': [
|
||
'report_id', 'tenant', 'name', 'description'
|
||
]
|
||
}),
|
||
('Report Configuration', {
|
||
'fields': [
|
||
'report_type', 'data_source', 'query_config'
|
||
]
|
||
}),
|
||
('Output Configuration', {
|
||
'fields': [
|
||
'output_format', 'template_config'
|
||
]
|
||
}),
|
||
('Scheduling', {
|
||
'fields': [
|
||
'schedule_type', 'schedule_config', 'next_execution'
|
||
]
|
||
}),
|
||
('Distribution', {
|
||
'fields': [
|
||
'recipients', 'distribution_config'
|
||
]
|
||
}),
|
||
('Report Status', {
|
||
'fields': [
|
||
'is_active'
|
||
]
|
||
}),
|
||
('Summary Information', {
|
||
'fields': [
|
||
'execution_count'
|
||
],
|
||
'classes': ['collapse']
|
||
}),
|
||
('Metadata', {
|
||
'fields': [
|
||
'created_at', 'updated_at', 'created_by'
|
||
],
|
||
'classes': ['collapse']
|
||
})
|
||
]
|
||
inlines = [ReportExecutionInline]
|
||
|
||
def execution_count_display(self, obj):
|
||
"""Display execution count."""
|
||
return obj.execution_count
|
||
execution_count_display.short_description = 'Executions'
|
||
|
||
def get_queryset(self, request):
|
||
"""Filter by user's tenant."""
|
||
qs = super().get_queryset(request)
|
||
if hasattr(request.user, 'tenant'):
|
||
qs = qs.filter(tenant=request.user.tenant)
|
||
return qs.select_related('data_source')
|
||
|
||
|
||
@admin.register(ReportExecution)
|
||
class ReportExecutionAdmin(admin.ModelAdmin):
|
||
"""
|
||
Admin interface for report executions.
|
||
"""
|
||
list_display = [
|
||
'report_name', 'execution_type', 'started_at',
|
||
'status', 'duration_display', 'record_count',
|
||
'output_size_display'
|
||
]
|
||
list_filter = [
|
||
'report__tenant', 'execution_type', 'status', 'started_at'
|
||
]
|
||
search_fields = [
|
||
'report__name', 'executed_by__username'
|
||
]
|
||
readonly_fields = [
|
||
'execution_id', 'started_at', 'completed_at',
|
||
'duration_seconds', 'output_size_bytes',
|
||
'record_count'
|
||
]
|
||
fieldsets = [
|
||
('Execution Information', {
|
||
'fields': [
|
||
'execution_id', 'report', 'execution_type'
|
||
]
|
||
}),
|
||
('Timing', {
|
||
'fields': [
|
||
'started_at', 'completed_at', 'duration_seconds'
|
||
]
|
||
}),
|
||
('Status and Results', {
|
||
'fields': [
|
||
'status', 'error_message'
|
||
]
|
||
}),
|
||
('Output', {
|
||
'fields': [
|
||
'output_file_path', 'output_size_bytes', 'record_count'
|
||
]
|
||
}),
|
||
('Parameters', {
|
||
'fields': [
|
||
'execution_parameters'
|
||
]
|
||
}),
|
||
('Metadata', {
|
||
'fields': [
|
||
'executed_by'
|
||
],
|
||
'classes': ['collapse']
|
||
})
|
||
]
|
||
date_hierarchy = 'started_at'
|
||
|
||
def report_name(self, obj):
|
||
"""Display report name."""
|
||
return obj.report.name
|
||
report_name.short_description = 'Report'
|
||
|
||
def duration_display(self, obj):
|
||
"""Display duration."""
|
||
if obj.duration_seconds:
|
||
if obj.duration_seconds >= 3600:
|
||
hours = obj.duration_seconds // 3600
|
||
minutes = (obj.duration_seconds % 3600) // 60
|
||
return f"{hours}h {minutes}m"
|
||
elif obj.duration_seconds >= 60:
|
||
minutes = obj.duration_seconds // 60
|
||
seconds = obj.duration_seconds % 60
|
||
return f"{minutes}m {seconds}s"
|
||
return f"{obj.duration_seconds}s"
|
||
return "-"
|
||
duration_display.short_description = 'Duration'
|
||
|
||
def output_size_display(self, obj):
|
||
"""Display output size."""
|
||
if obj.output_size_bytes:
|
||
if obj.output_size_bytes >= 1024 * 1024:
|
||
mb = obj.output_size_bytes / (1024 * 1024)
|
||
return f"{mb:.1f} MB"
|
||
elif obj.output_size_bytes >= 1024:
|
||
kb = obj.output_size_bytes / 1024
|
||
return f"{kb:.1f} KB"
|
||
return f"{obj.output_size_bytes} B"
|
||
return "-"
|
||
output_size_display.short_description = 'Size'
|
||
|
||
def get_queryset(self, request):
|
||
"""Filter by user's tenant."""
|
||
qs = super().get_queryset(request)
|
||
if hasattr(request.user, 'tenant'):
|
||
qs = qs.filter(report__tenant=request.user.tenant)
|
||
return qs.select_related('report', 'executed_by')
|
||
|
||
|
||
class MetricValueInline(admin.TabularInline):
|
||
"""
|
||
Inline admin for metric values.
|
||
"""
|
||
model = MetricValue
|
||
extra = 0
|
||
fields = [
|
||
'value', 'period_start', 'period_end',
|
||
'data_quality_score', 'confidence_level'
|
||
]
|
||
readonly_fields = [
|
||
'value_id', 'calculation_timestamp',
|
||
'calculation_duration_ms'
|
||
]
|
||
|
||
|
||
@admin.register(MetricDefinition)
|
||
class MetricDefinitionAdmin(admin.ModelAdmin):
|
||
"""
|
||
Admin interface for metric definitions.
|
||
"""
|
||
list_display = [
|
||
'name', 'metric_type', 'aggregation_period',
|
||
'target_value', 'unit_of_measure',
|
||
'value_count_display', 'is_active'
|
||
]
|
||
list_filter = [
|
||
'tenant', 'metric_type', 'aggregation_period', 'is_active'
|
||
]
|
||
search_fields = [
|
||
'name', 'description'
|
||
]
|
||
readonly_fields = [
|
||
'metric_id',
|
||
'created_at', 'updated_at'
|
||
]
|
||
fieldsets = [
|
||
('Metric Information', {
|
||
'fields': [
|
||
'metric_id', 'tenant', 'name', 'description'
|
||
]
|
||
}),
|
||
('Metric Configuration', {
|
||
'fields': [
|
||
'metric_type', 'data_source', 'calculation_config'
|
||
]
|
||
}),
|
||
('Aggregation', {
|
||
'fields': [
|
||
'aggregation_period', 'aggregation_config'
|
||
]
|
||
}),
|
||
('Thresholds and Targets', {
|
||
'fields': [
|
||
'target_value', 'warning_threshold', 'critical_threshold'
|
||
]
|
||
}),
|
||
('Display Configuration', {
|
||
'fields': [
|
||
'unit_of_measure', 'decimal_places', 'display_format'
|
||
]
|
||
}),
|
||
('Metric Status', {
|
||
'fields': [
|
||
'is_active'
|
||
]
|
||
}),
|
||
('Summary Information', {
|
||
'fields': [
|
||
'value_count'
|
||
],
|
||
'classes': ['collapse']
|
||
}),
|
||
('Metadata', {
|
||
'fields': [
|
||
'created_at', 'updated_at', 'created_by'
|
||
],
|
||
'classes': ['collapse']
|
||
})
|
||
]
|
||
inlines = [MetricValueInline]
|
||
|
||
def value_count_display(self, obj):
|
||
"""Display value count."""
|
||
return obj.value_count
|
||
value_count_display.short_description = 'Values'
|
||
|
||
def get_queryset(self, request):
|
||
"""Filter by user's tenant."""
|
||
qs = super().get_queryset(request)
|
||
if hasattr(request.user, 'tenant'):
|
||
qs = qs.filter(tenant=request.user.tenant)
|
||
return qs.select_related('data_source')
|
||
|
||
|
||
@admin.register(MetricValue)
|
||
class MetricValueAdmin(admin.ModelAdmin):
|
||
"""
|
||
Admin interface for metric values.
|
||
"""
|
||
list_display = [
|
||
'metric_name', 'value', 'period_start',
|
||
'period_end', 'threshold_status_display',
|
||
'data_quality_score', 'calculation_timestamp'
|
||
]
|
||
list_filter = [
|
||
'metric_definition__tenant', 'metric_definition__metric_type',
|
||
'period_start', 'calculation_timestamp'
|
||
]
|
||
search_fields = [
|
||
'metric_definition__name'
|
||
]
|
||
readonly_fields = [
|
||
'value_id', 'is_above_target', 'threshold_status',
|
||
'calculation_timestamp', 'calculation_duration_ms'
|
||
]
|
||
fieldsets = [
|
||
('Metric Value Information', {
|
||
'fields': [
|
||
'value_id', 'metric_definition'
|
||
]
|
||
}),
|
||
('Value Details', {
|
||
'fields': [
|
||
'value', 'period_start', 'period_end'
|
||
]
|
||
}),
|
||
('Context', {
|
||
'fields': [
|
||
'dimensions', 'metadata'
|
||
]
|
||
}),
|
||
('Quality Indicators', {
|
||
'fields': [
|
||
'data_quality_score', 'confidence_level'
|
||
]
|
||
}),
|
||
('Calculation Details', {
|
||
'fields': [
|
||
'calculation_timestamp', 'calculation_duration_ms'
|
||
]
|
||
}),
|
||
('Analysis', {
|
||
'fields': [
|
||
'is_above_target', 'threshold_status'
|
||
],
|
||
'classes': ['collapse']
|
||
})
|
||
]
|
||
date_hierarchy = 'period_start'
|
||
|
||
def metric_name(self, obj):
|
||
"""Display metric name."""
|
||
return obj.metric_definition.name
|
||
metric_name.short_description = 'Metric'
|
||
|
||
def threshold_status_display(self, obj):
|
||
"""Display threshold status with color coding."""
|
||
status = obj.threshold_status
|
||
if status == 'CRITICAL':
|
||
return format_html('<span style="color: red;">🔴 Critical</span>')
|
||
elif status == 'WARNING':
|
||
return format_html('<span style="color: orange;">🟡 Warning</span>')
|
||
return format_html('<span style="color: green;">🟢 Normal</span>')
|
||
threshold_status_display.short_description = 'Status'
|
||
|
||
def get_queryset(self, request):
|
||
"""Filter by user's tenant."""
|
||
qs = super().get_queryset(request)
|
||
if hasattr(request.user, 'tenant'):
|
||
qs = qs.filter(metric_definition__tenant=request.user.tenant)
|
||
return qs.select_related('metric_definition')
|
||
|
||
|
||
# Customize admin site
|
||
admin.site.site_header = "Hospital Management System - Analytics"
|
||
admin.site.site_title = "Analytics Admin"
|
||
admin.site.index_title = "Analytics Administration"
|
||
|