""" Analytics models — risk scores, anomalies, metrics, caching, usage logs. Maps to AGENT.md DBML: content_embeddings (545-555), risk_scores (557-563), anomalies (565-572), metrics (574-580), read_models_cache (582-587), feature_usage_logs (589-595), api_request_logs (597-608). """ from django.conf import settings from django.db import models from core.models import BaseModel class ContentEmbedding(BaseModel): """ Vector embedding for content (products, help articles, etc.). Uses a text field to store the vector since pgvector may not be available. """ content_type = models.CharField(max_length=100) content_id = models.UUIDField() embedding = models.TextField() # JSON serialized vector (pgvector in production) class Meta: db_table = "content_embeddings" def __str__(self) -> str: return f"Embedding for {self.content_type}:{self.content_id}" class RiskScore(BaseModel): """User risk score for security intelligence.""" user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="risk_scores", ) score = models.FloatField() # 0.0 - 1.0 factors = models.JSONField(default=dict, blank=True) calculated_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = "risk_scores" ordering = ["-calculated_at"] def __str__(self) -> str: return f"Risk {self.score:.2f} for {self.user_id}" class Anomaly(BaseModel): """Detected anomalous behavior.""" class Severity(models.TextChoices): LOW = "low", "Low" MEDIUM = "medium", "Medium" HIGH = "high", "High" CRITICAL = "critical", "Critical" user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="anomalies", ) anomaly_type = models.CharField(max_length=100) severity = models.CharField(max_length=20, choices=Severity.choices, default=Severity.LOW) metadata = models.JSONField(default=dict, blank=True) class Meta: db_table = "anomalies" ordering = ["-created_at"] def __str__(self) -> str: return f"{self.anomaly_type} ({self.get_severity_display()})" class Metric(BaseModel): """Time-series metric data point.""" metric_key = models.CharField(max_length=100, db_index=True) value = models.FloatField() tags = models.JSONField(default=dict, blank=True) recorded_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = "metrics" ordering = ["-recorded_at"] def __str__(self) -> str: return f"{self.metric_key}={self.value}" class ReadModelCache(models.Model): """ CQRS read model cache for denormalized queries. """ cache_key = models.CharField(max_length=255, primary_key=True) org = models.ForeignKey( "organizations.Organization", on_delete=models.CASCADE, null=True, blank=True, related_name="read_caches", ) data_json = models.JSONField(default=dict) refreshed_at = models.DateTimeField(auto_now=True) class Meta: db_table = "read_models_cache" def __str__(self) -> str: return self.cache_key class FeatureUsageLog(BaseModel): """Tracks feature usage for analytics.""" org = models.ForeignKey( "organizations.Organization", on_delete=models.SET_NULL, null=True, blank=True, related_name="feature_usage_logs", ) user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="feature_usage_logs", ) feature_key = models.CharField(max_length=100, db_index=True) used_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = "feature_usage_logs" ordering = ["-used_at"] def __str__(self) -> str: return f"{self.feature_key} by {self.user_id}" class APIRequestLog(BaseModel): """Logs API requests for monitoring and analytics.""" request_id = models.CharField(max_length=100, unique=True) user = models.ForeignKey( settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name="api_request_logs", ) org = models.ForeignKey( "organizations.Organization", on_delete=models.SET_NULL, null=True, blank=True, related_name="api_request_logs", ) endpoint = models.CharField(max_length=500, blank=True, default="") method = models.CharField(max_length=10, blank=True, default="") status_code = models.PositiveIntegerField(default=0) duration_ms = models.PositiveIntegerField(default=0) ip_address = models.GenericIPAddressField(null=True, blank=True) class Meta: db_table = "api_request_logs" ordering = ["-created_at"] def __str__(self) -> str: return f"{self.method} {self.endpoint} → {self.status_code}"