493 lines
20 KiB
Python
493 lines
20 KiB
Python
"""
|
|
Analytics app models.
|
|
"""
|
|
|
|
import uuid
|
|
import json
|
|
from django.db import models
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.validators import MinValueValidator, MaxValueValidator
|
|
from django.utils import timezone
|
|
from core.models import Tenant
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class Dashboard(models.Model):
|
|
"""
|
|
Dashboard model for organizing widgets and analytics views.
|
|
"""
|
|
|
|
class DashboardType(models.TextChoices):
|
|
EXECUTIVE = 'EXECUTIVE', 'Executive Dashboard'
|
|
CLINICAL = 'CLINICAL', 'Clinical Dashboard'
|
|
OPERATIONAL = 'OPERATIONAL', 'Operational Dashboard'
|
|
FINANCIAL = 'FINANCIAL', 'Financial Dashboard'
|
|
QUALITY = 'QUALITY', 'Quality Dashboard'
|
|
PATIENT = 'PATIENT', 'Patient Dashboard'
|
|
PROVIDER = 'PROVIDER', 'Provider Dashboard'
|
|
DEPARTMENT = 'DEPARTMENT', 'Department Dashboard'
|
|
CUSTOM = 'CUSTOM', 'Custom Dashboard'
|
|
|
|
dashboard_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='dashboards')
|
|
name = models.CharField(max_length=200)
|
|
description = models.TextField(blank=True)
|
|
dashboard_type = models.CharField(max_length=20, choices=DashboardType.choices)
|
|
|
|
# Layout and configuration
|
|
layout_config = models.JSONField(default=dict, help_text="Dashboard layout configuration")
|
|
refresh_interval = models.PositiveIntegerField(default=300, help_text="Refresh interval in seconds")
|
|
|
|
# Access control
|
|
is_public = models.BooleanField(default=False)
|
|
allowed_users = models.ManyToManyField(User, blank=True, related_name='accessible_dashboards')
|
|
allowed_roles = models.JSONField(default=list, help_text="List of allowed user roles")
|
|
|
|
# Status
|
|
is_active = models.BooleanField(default=True)
|
|
is_default = models.BooleanField(default=False)
|
|
|
|
# Metadata
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
|
|
|
|
class Meta:
|
|
db_table = 'analytics_dashboard'
|
|
indexes = [
|
|
models.Index(fields=['tenant', 'dashboard_type']),
|
|
models.Index(fields=['tenant', 'is_active']),
|
|
models.Index(fields=['tenant', 'is_default']),
|
|
]
|
|
unique_together = [['tenant', 'name']]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.get_dashboard_type_display()})"
|
|
|
|
|
|
class DashboardWidget(models.Model):
|
|
"""
|
|
Dashboard widget model for individual analytics components.
|
|
"""
|
|
|
|
class WidgetType(models.TextChoices):
|
|
CHART = 'CHART', 'Chart Widget'
|
|
TABLE = 'TABLE', 'Table Widget'
|
|
METRIC = 'METRIC', 'Metric Widget'
|
|
GAUGE = 'GAUGE', 'Gauge Widget'
|
|
MAP = 'MAP', 'Map Widget'
|
|
TEXT = 'TEXT', 'Text Widget'
|
|
IMAGE = 'IMAGE', 'Image Widget'
|
|
IFRAME = 'IFRAME', 'IFrame Widget'
|
|
CUSTOM = 'CUSTOM', 'Custom Widget'
|
|
|
|
class ChartType(models.TextChoices):
|
|
LINE = 'LINE', 'Line Chart'
|
|
BAR = 'BAR', 'Bar Chart'
|
|
PIE = 'PIE', 'Pie Chart'
|
|
DOUGHNUT = 'DOUGHNUT', 'Doughnut Chart'
|
|
AREA = 'AREA', 'Area Chart'
|
|
SCATTER = 'SCATTER', 'Scatter Plot'
|
|
HISTOGRAM = 'HISTOGRAM', 'Histogram'
|
|
HEATMAP = 'HEATMAP', 'Heat Map'
|
|
TREEMAP = 'TREEMAP', 'Tree Map'
|
|
FUNNEL = 'FUNNEL', 'Funnel Chart'
|
|
|
|
widget_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
dashboard = models.ForeignKey(Dashboard, on_delete=models.CASCADE, related_name='widgets')
|
|
|
|
# Widget configuration
|
|
name = models.CharField(max_length=200)
|
|
description = models.TextField(blank=True)
|
|
widget_type = models.CharField(max_length=20, choices=WidgetType.choices)
|
|
chart_type = models.CharField(max_length=20, choices=ChartType.choices, blank=True)
|
|
|
|
# Data source
|
|
data_source = models.ForeignKey('DataSource', on_delete=models.CASCADE, related_name='widgets')
|
|
query_config = models.JSONField(default=dict, help_text="Query configuration for data source")
|
|
|
|
# Layout
|
|
position_x = models.PositiveIntegerField(default=0)
|
|
position_y = models.PositiveIntegerField(default=0)
|
|
width = models.PositiveIntegerField(default=4, validators=[MinValueValidator(1), MaxValueValidator(12)])
|
|
height = models.PositiveIntegerField(default=4, validators=[MinValueValidator(1), MaxValueValidator(12)])
|
|
|
|
# Display configuration
|
|
display_config = models.JSONField(default=dict, help_text="Widget display configuration")
|
|
color_scheme = models.CharField(max_length=50, default='default')
|
|
|
|
# Refresh settings
|
|
auto_refresh = models.BooleanField(default=True)
|
|
refresh_interval = models.PositiveIntegerField(default=300, help_text="Refresh interval in seconds")
|
|
|
|
# Status
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
# Metadata
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
db_table = 'analytics_dashboard_widget'
|
|
indexes = [
|
|
models.Index(fields=['dashboard', 'is_active']),
|
|
models.Index(fields=['dashboard', 'position_x', 'position_y']),
|
|
]
|
|
ordering = ['position_y', 'position_x']
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.get_widget_type_display()})"
|
|
|
|
|
|
class DataSource(models.Model):
|
|
"""
|
|
Data source model for analytics data connections.
|
|
"""
|
|
|
|
class SourceType(models.TextChoices):
|
|
DATABASE = 'DATABASE', 'Database Query'
|
|
API = 'API', 'API Endpoint'
|
|
FILE = 'FILE', 'File Upload'
|
|
STREAM = 'STREAM', 'Real-time Stream'
|
|
WEBHOOK = 'WEBHOOK', 'Webhook'
|
|
CUSTOM = 'CUSTOM', 'Custom Source'
|
|
|
|
class ConnectionType(models.TextChoices):
|
|
POSTGRESQL = 'POSTGRESQL', 'PostgreSQL'
|
|
MYSQL = 'MYSQL', 'MySQL'
|
|
SQLITE = 'SQLITE', 'SQLite'
|
|
MONGODB = 'MONGODB', 'MongoDB'
|
|
REDIS = 'REDIS', 'Redis'
|
|
REST_API = 'REST_API', 'REST API'
|
|
GRAPHQL = 'GRAPHQL', 'GraphQL'
|
|
WEBSOCKET = 'WEBSOCKET', 'WebSocket'
|
|
CSV = 'CSV', 'CSV File'
|
|
JSON = 'JSON', 'JSON File'
|
|
XML = 'XML', 'XML File'
|
|
|
|
class TestStatus(models.TextChoices):
|
|
PENDING = 'PENDING', 'Pending'
|
|
RUNNING = 'RUNNING', 'Running'
|
|
SUCCESS = 'SUCCESS', 'Success'
|
|
FAILURE = 'FAILURE', 'Failure'
|
|
|
|
source_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='data_sources')
|
|
|
|
# Source configuration
|
|
name = models.CharField(max_length=200)
|
|
description = models.TextField(blank=True)
|
|
source_type = models.CharField(max_length=20, choices=SourceType.choices)
|
|
connection_type = models.CharField(max_length=20, choices=ConnectionType.choices)
|
|
|
|
# Connection details
|
|
connection_config = models.JSONField(default=dict, help_text="Connection configuration")
|
|
authentication_config = models.JSONField(default=dict, help_text="Authentication configuration")
|
|
|
|
# Query/endpoint details
|
|
query_template = models.TextField(blank=True, help_text="SQL query or API endpoint template")
|
|
parameters = models.JSONField(default=dict, help_text="Query parameters")
|
|
|
|
# Data processing
|
|
data_transformation = models.JSONField(default=dict, help_text="Data transformation rules")
|
|
cache_duration = models.PositiveIntegerField(default=300, help_text="Cache duration in seconds")
|
|
|
|
# Health monitoring
|
|
is_healthy = models.BooleanField(default=True)
|
|
last_health_check = models.DateTimeField(null=True, blank=True)
|
|
health_check_interval = models.PositiveIntegerField(default=300, help_text="Health check interval in seconds")
|
|
|
|
# Status
|
|
is_active = models.BooleanField(default=True)
|
|
last_test_status = models.CharField(max_length=20, choices=TestStatus.choices, default=TestStatus.PENDING)
|
|
last_test_start_at = models.DateTimeField(null=True, blank=True)
|
|
last_test_end_at = models.DateTimeField(null=True, blank=True)
|
|
last_test_duration_seconds = models.PositiveIntegerField(null=True, blank=True)
|
|
last_test_error_message = models.TextField(blank=True)
|
|
|
|
# Metadata
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
|
|
|
|
class Meta:
|
|
db_table = 'analytics_data_source'
|
|
indexes = [
|
|
models.Index(fields=['tenant', 'source_type']),
|
|
models.Index(fields=['tenant', 'is_active']),
|
|
models.Index(fields=['tenant', 'is_healthy']),
|
|
]
|
|
unique_together = [['tenant', 'name']]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.get_source_type_display()})"
|
|
|
|
|
|
class Report(models.Model):
|
|
"""
|
|
Report model for scheduled and ad-hoc reporting.
|
|
"""
|
|
|
|
class ReportType(models.TextChoices):
|
|
OPERATIONAL = 'OPERATIONAL', 'Operational Report'
|
|
FINANCIAL = 'FINANCIAL', 'Financial Report'
|
|
CLINICAL = 'CLINICAL', 'Clinical Report'
|
|
QUALITY = 'QUALITY', 'Quality Report'
|
|
COMPLIANCE = 'COMPLIANCE', 'Compliance Report'
|
|
PERFORMANCE = 'PERFORMANCE', 'Performance Report'
|
|
CUSTOM = 'CUSTOM', 'Custom Report'
|
|
|
|
class OutputFormat(models.TextChoices):
|
|
PDF = 'PDF', 'PDF Document'
|
|
EXCEL = 'EXCEL', 'Excel Spreadsheet'
|
|
CSV = 'CSV', 'CSV File'
|
|
JSON = 'JSON', 'JSON Data'
|
|
HTML = 'HTML', 'HTML Page'
|
|
EMAIL = 'EMAIL', 'Email Report'
|
|
|
|
class ScheduleType(models.TextChoices):
|
|
MANUAL = 'MANUAL', 'Manual Execution'
|
|
DAILY = 'DAILY', 'Daily'
|
|
WEEKLY = 'WEEKLY', 'Weekly'
|
|
MONTHLY = 'MONTHLY', 'Monthly'
|
|
QUARTERLY = 'QUARTERLY', 'Quarterly'
|
|
YEARLY = 'YEARLY', 'Yearly'
|
|
CUSTOM = 'CUSTOM', 'Custom Schedule'
|
|
|
|
report_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='reports')
|
|
|
|
# Report configuration
|
|
name = models.CharField(max_length=200)
|
|
description = models.TextField(blank=True)
|
|
report_type = models.CharField(max_length=20, choices=ReportType.choices)
|
|
|
|
# Data source
|
|
data_source = models.ForeignKey(DataSource, on_delete=models.CASCADE, related_name='reports')
|
|
query_config = models.JSONField(default=dict, help_text="Query configuration for report")
|
|
|
|
# Output configuration
|
|
output_format = models.CharField(max_length=20, choices=OutputFormat.choices)
|
|
template_config = models.JSONField(default=dict, help_text="Report template configuration")
|
|
|
|
# Scheduling
|
|
schedule_type = models.CharField(max_length=20, choices=ScheduleType.choices, default=ScheduleType.MANUAL)
|
|
schedule_config = models.JSONField(default=dict, help_text="Schedule configuration")
|
|
next_execution = models.DateTimeField(null=True, blank=True)
|
|
|
|
# Distribution
|
|
recipients = models.JSONField(default=list, help_text="Report recipients")
|
|
distribution_config = models.JSONField(default=dict, help_text="Distribution configuration")
|
|
|
|
# Status
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
# Metadata
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
|
|
|
|
class Meta:
|
|
db_table = 'analytics_report'
|
|
indexes = [
|
|
models.Index(fields=['tenant', 'report_type']),
|
|
models.Index(fields=['tenant', 'schedule_type']),
|
|
models.Index(fields=['tenant', 'next_execution']),
|
|
models.Index(fields=['tenant', 'is_active']),
|
|
]
|
|
unique_together = [['tenant', 'name']]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.get_report_type_display()})"
|
|
|
|
|
|
class ReportExecution(models.Model):
|
|
"""
|
|
Report execution model for tracking report runs.
|
|
"""
|
|
|
|
class ExecutionStatus(models.TextChoices):
|
|
PENDING = 'PENDING', 'Pending'
|
|
RUNNING = 'RUNNING', 'Running'
|
|
COMPLETED = 'COMPLETED', 'Completed'
|
|
FAILED = 'FAILED', 'Failed'
|
|
CANCELLED = 'CANCELLED', 'Cancelled'
|
|
|
|
class ExecutionType(models.TextChoices):
|
|
MANUAL = 'MANUAL', 'Manual'
|
|
SCHEDULED = 'SCHEDULED', 'Scheduled'
|
|
API = 'API', 'API Triggered'
|
|
|
|
execution_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
report = models.ForeignKey(Report, on_delete=models.CASCADE, related_name='executions')
|
|
|
|
# Execution details
|
|
execution_type = models.CharField(max_length=20, choices=ExecutionType.choices, default=ExecutionType.MANUAL)
|
|
|
|
# Timing
|
|
started_at = models.DateTimeField(auto_now_add=True)
|
|
completed_at = models.DateTimeField(null=True, blank=True)
|
|
duration_seconds = models.PositiveIntegerField(null=True, blank=True)
|
|
|
|
# Status and results
|
|
status = models.CharField(max_length=20, choices=ExecutionStatus.choices, default=ExecutionStatus.PENDING)
|
|
error_message = models.TextField(blank=True)
|
|
|
|
# Output
|
|
output_file_path = models.CharField(max_length=500, blank=True)
|
|
output_size_bytes = models.PositiveBigIntegerField(null=True, blank=True)
|
|
record_count = models.PositiveIntegerField(null=True, blank=True)
|
|
|
|
# Parameters
|
|
execution_parameters = models.JSONField(default=dict, help_text="Execution parameters")
|
|
|
|
# Metadata
|
|
executed_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
|
|
|
|
class Meta:
|
|
db_table = 'analytics_report_execution'
|
|
indexes = [
|
|
models.Index(fields=['report', 'status']),
|
|
models.Index(fields=['report', 'started_at']),
|
|
models.Index(fields=['status', 'started_at']),
|
|
]
|
|
ordering = ['-started_at']
|
|
|
|
def __str__(self):
|
|
return f"{self.report.name} - {self.started_at.strftime('%Y-%m-%d %H:%M')}"
|
|
|
|
@property
|
|
def is_completed(self):
|
|
"""Check if execution is completed."""
|
|
return self.status in ['COMPLETED', 'FAILED', 'CANCELLED']
|
|
|
|
|
|
class MetricDefinition(models.Model):
|
|
"""
|
|
Metric definition model for KPI and performance metrics.
|
|
"""
|
|
|
|
class MetricType(models.TextChoices):
|
|
COUNT = 'COUNT', 'Count'
|
|
SUM = 'SUM', 'Sum'
|
|
AVERAGE = 'AVERAGE', 'Average'
|
|
PERCENTAGE = 'PERCENTAGE', 'Percentage'
|
|
RATIO = 'RATIO', 'Ratio'
|
|
RATE = 'RATE', 'Rate'
|
|
DURATION = 'DURATION', 'Duration'
|
|
CUSTOM = 'CUSTOM', 'Custom Calculation'
|
|
|
|
class AggregationPeriod(models.TextChoices):
|
|
REAL_TIME = 'REAL_TIME', 'Real-time'
|
|
HOURLY = 'HOURLY', 'Hourly'
|
|
DAILY = 'DAILY', 'Daily'
|
|
WEEKLY = 'WEEKLY', 'Weekly'
|
|
MONTHLY = 'MONTHLY', 'Monthly'
|
|
QUARTERLY = 'QUARTERLY', 'Quarterly'
|
|
YEARLY = 'YEARLY', 'Yearly'
|
|
|
|
metric_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='metric_definitions')
|
|
|
|
# Metric configuration
|
|
name = models.CharField(max_length=200)
|
|
description = models.TextField(blank=True)
|
|
metric_type = models.CharField(max_length=20, choices=MetricType.choices)
|
|
|
|
# Data source
|
|
data_source = models.ForeignKey(DataSource, on_delete=models.CASCADE, related_name='metrics')
|
|
calculation_config = models.JSONField(default=dict, help_text="Metric calculation configuration")
|
|
|
|
# Aggregation
|
|
aggregation_period = models.CharField(max_length=20, choices=AggregationPeriod.choices)
|
|
aggregation_config = models.JSONField(default=dict, help_text="Aggregation configuration")
|
|
|
|
# Thresholds and targets
|
|
target_value = models.DecimalField(max_digits=15, decimal_places=4, null=True, blank=True)
|
|
warning_threshold = models.DecimalField(max_digits=15, decimal_places=4, null=True, blank=True)
|
|
critical_threshold = models.DecimalField(max_digits=15, decimal_places=4, null=True, blank=True)
|
|
|
|
# Display configuration
|
|
unit_of_measure = models.CharField(max_length=50, blank=True)
|
|
decimal_places = models.PositiveIntegerField(default=2, validators=[MaxValueValidator(10)])
|
|
display_format = models.CharField(max_length=50, default='number')
|
|
|
|
# Status
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
# Metadata
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
created_by = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True)
|
|
|
|
class Meta:
|
|
db_table = 'analytics_metric_definition'
|
|
indexes = [
|
|
models.Index(fields=['tenant', 'metric_type']),
|
|
models.Index(fields=['tenant', 'aggregation_period']),
|
|
models.Index(fields=['tenant', 'is_active']),
|
|
]
|
|
unique_together = [['tenant', 'name']]
|
|
|
|
def __str__(self):
|
|
return f"{self.name} ({self.get_metric_type_display()})"
|
|
|
|
|
|
class MetricValue(models.Model):
|
|
"""
|
|
Metric value model for storing calculated metric values.
|
|
"""
|
|
value_id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
|
|
metric_definition = models.ForeignKey(MetricDefinition, on_delete=models.CASCADE, related_name='values')
|
|
|
|
# Value details
|
|
value = models.DecimalField(max_digits=15, decimal_places=4)
|
|
period_start = models.DateTimeField()
|
|
period_end = models.DateTimeField()
|
|
|
|
# Context
|
|
dimensions = models.JSONField(default=dict, help_text="Metric dimensions (e.g., department, provider)")
|
|
metadata = models.JSONField(default=dict, help_text="Additional metadata")
|
|
|
|
# Quality indicators
|
|
data_quality_score = models.DecimalField(max_digits=5, decimal_places=2, null=True, blank=True,
|
|
validators=[MinValueValidator(0), MaxValueValidator(100)])
|
|
confidence_level = models.DecimalField(max_digits=5, decimal_places=2, null=True, blank=True,
|
|
validators=[MinValueValidator(0), MaxValueValidator(100)])
|
|
|
|
# Calculation details
|
|
calculation_timestamp = models.DateTimeField(auto_now_add=True)
|
|
calculation_duration_ms = models.PositiveIntegerField(null=True, blank=True)
|
|
|
|
class Meta:
|
|
db_table = 'analytics_metric_value'
|
|
indexes = [
|
|
models.Index(fields=['metric_definition', 'period_start']),
|
|
models.Index(fields=['metric_definition', 'period_end']),
|
|
models.Index(fields=['period_start', 'period_end']),
|
|
models.Index(fields=['calculation_timestamp']),
|
|
]
|
|
unique_together = [['metric_definition', 'period_start', 'period_end']]
|
|
ordering = ['-period_start']
|
|
|
|
def __str__(self):
|
|
return f"{self.metric_definition.name}: {self.value} ({self.period_start.date()})"
|
|
|
|
@property
|
|
def is_above_target(self):
|
|
"""Check if value is above target."""
|
|
if self.metric_definition.target_value:
|
|
return self.value >= self.metric_definition.target_value
|
|
return None
|
|
|
|
@property
|
|
def threshold_status(self):
|
|
"""Get threshold status."""
|
|
if self.metric_definition.critical_threshold and self.value >= self.metric_definition.critical_threshold:
|
|
return 'CRITICAL'
|
|
elif self.metric_definition.warning_threshold and self.value >= self.metric_definition.warning_threshold:
|
|
return 'WARNING'
|
|
return 'NORMAL'
|
|
|