""" Analytics service — risk scoring, anomaly detection, metrics, usage logs. """ import uuid from django.utils import timezone from apps.intelligence.models import ( Anomaly, APIRequestLog, ContentEmbedding, FeatureUsageLog, Metric, ReadModelCache, RiskScore, ) # ─── Risk Scores ───────────────────────────────────────────────────────────── def calculate_risk_score(user, score: float, factors: dict = None) -> RiskScore: """Record a risk score for a user.""" return RiskScore.objects.create( user=user, score=max(0.0, min(1.0, score)), factors=factors or {}, ) def get_latest_risk_score(user): """Get the most recent risk score for a user.""" return RiskScore.objects.filter(user=user).first() def process_security_event_for_risk(user, event_type: str): """Feed a security event into the risk intelligence system.""" latest = get_latest_risk_score(user) current_score = latest.score if latest else 0.1 # Default baseline risk # Simple risk heuristics based on event type risk_deltas = { "login_failed": 0.15, "password_reset_requested": 0.05, "mfa_disabled": 0.2, "login_success": -0.05, "mfa_enabled": -0.2, "password_changed": -0.1, } delta = risk_deltas.get(event_type, 0.0) if delta == 0.0: return None # No risk impact new_score = max(0.0, min(1.0, current_score + delta)) factors = {"event": event_type, "delta": delta} if latest and latest.factors: # Keep a short history of factors factors["previous"] = latest.factors.get("event") return calculate_risk_score(user, new_score, factors=factors) # ─── Anomalies ─────────────────────────────────────────────────────────────── def report_anomaly(user, anomaly_type: str, severity: str = "low", metadata: dict = None) -> Anomaly: """Record a detected anomaly.""" return Anomaly.objects.create( user=user, anomaly_type=anomaly_type, severity=severity, metadata=metadata or {}, ) def get_anomalies(user=None, severity: str = None, limit: int = 50): """Query anomalies with optional filters.""" qs = Anomaly.objects.all() if user: qs = qs.filter(user=user) if severity: qs = qs.filter(severity=severity) return qs[:limit] # ─── Metrics ───────────────────────────────────────────────────────────────── def record_metric(metric_key: str, value: float, tags: dict = None) -> Metric: """Record a time-series metric data point.""" return Metric.objects.create( metric_key=metric_key, value=value, tags=tags or {}, ) def get_metrics(metric_key: str, limit: int = 100): """Query metrics by key.""" return Metric.objects.filter(metric_key=metric_key)[:limit] # ─── Read Model Cache ─────────────────────────────────────────────────────── def set_cache(cache_key: str, data: dict, org=None) -> ReadModelCache: """Set or update a read model cache entry.""" cache, _ = ReadModelCache.objects.update_or_create( cache_key=cache_key, defaults={"data_json": data, "org": org}, ) return cache def get_cache(cache_key: str) -> dict: """Get a cached read model.""" try: return ReadModelCache.objects.get(cache_key=cache_key).data_json except ReadModelCache.DoesNotExist: return None # ─── Feature Usage Logs ───────────────────────────────────────────────────── def log_feature_usage(feature_key: str, user=None, org=None) -> FeatureUsageLog: """Log a feature usage event.""" return FeatureUsageLog.objects.create( feature_key=feature_key, user=user, org=org, ) # ─── API Request Logs ─────────────────────────────────────────────────────── def log_api_request( endpoint: str, method: str, status_code: int, duration_ms: int = 0, user=None, org=None, ip_address: str = None, ) -> APIRequestLog: """Log an API request.""" return APIRequestLog.objects.create( request_id=str(uuid.uuid4()), user=user, org=org, endpoint=endpoint, method=method, status_code=status_code, duration_ms=duration_ms, ip_address=ip_address, ) # ─── Content Embeddings & Semantic Search ──────────────────────────────────── def store_embedding( content_type: str, content_id, embedding: list, ) -> ContentEmbedding: """ Store or update a vector embedding for a piece of content. Args: content_type: Category label (e.g. "help_article", "product", "faq") content_id: UUID of the source content embedding: List of floats — the embedding vector """ import json obj, _ = ContentEmbedding.objects.update_or_create( content_type=content_type, content_id=content_id, defaults={"embedding": json.dumps(embedding)}, ) return obj def semantic_search( query_embedding: list, content_type: str = None, limit: int = 10, ) -> list: """ Search for content by vector similarity (cosine distance). Args: query_embedding: The embedding vector for the search query. content_type: Optional filter — restrict to a specific content type. limit: Maximum results to return. Returns: List of dicts: [{"content_type", "content_id", "score"}, ...] Sorted by descending similarity score. Note: In production with pgvector, replace the Python cosine calculation with: ContentEmbedding.objects.order_by( CosineDistance('embedding', query_embedding) )[:limit] """ import json import math qs = ContentEmbedding.objects.all() if content_type: qs = qs.filter(content_type=content_type) results = [] for entry in qs.iterator(): try: stored = json.loads(entry.embedding) except (json.JSONDecodeError, TypeError): continue score = _cosine_similarity(query_embedding, stored) results.append({ "content_type": entry.content_type, "content_id": str(entry.content_id), "score": score, }) # Sort by similarity (highest first) and return top N results.sort(key=lambda r: r["score"], reverse=True) return results[:limit] def _cosine_similarity(a: list, b: list) -> float: """Compute cosine similarity between two vectors.""" import math if len(a) != len(b) or not a: return 0.0 dot = sum(x * y for x, y in zip(a, b)) mag_a = math.sqrt(sum(x * x for x in a)) mag_b = math.sqrt(sum(x * x for x in b)) if mag_a == 0 or mag_b == 0: return 0.0 return dot / (mag_a * mag_b)