2326 Analytics Engine Cf8bc17a

# Analytics Engine

## Analytics Engine

```python
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from collections import defaultdict
import json

@dataclass
class GenerationEvent:
    timestamp: datetime
    job_id: str
    user_id: str
    project_id: str
    prompt: str
    model: str
    duration: int
    status: str  # completed, failed
    credits_used: int
    generation_time_seconds: float
    video_url: Optional[str] = None
    error: Optional[str] = None

class UsageAnalytics:
    """Analytics engine for Kling AI usage."""

    def __init__(self):
        self.events: List[GenerationEvent] = []

    def record_event(self, event: GenerationEvent):
        """Record a generation event."""
        self.events.append(event)

    def get_summary(
        self,
        start: datetime = None,
        end: datetime = None,
        user_id: str = None,
        project_id: str = None
    ) -> Dict:
        """Get usage summary for period."""
        filtered = self._filter_events(start, end, user_id, project_id)

        if not filtered:
            return {"total_generations": 0}

        completed = [e for e in filtered if e.status == "completed"]
        failed = [e for e in filtered if e.status == "failed"]

        total_credits = sum(e.credits_used for e in filtered)
        total_time = sum(e.generation_time_seconds for e in completed)

        return {
            "period": {
                "start": (start or filtered[0].timestamp).isoformat(),
                "end": (end or filtered[-1].timestamp).isoformat()
            },
            "total_generations": len(filtered),
            "completed": len(completed),
            "failed": len(failed),
            "success_rate": len(completed) / len(filtered) * 100 if filtered else 0,
            "total_credits": total_credits,
            "avg_credits_per_video": total_credits / len(filtered) if filtered else 0,
            "total_generation_time": total_time,
            "avg_generation_time": total_time / len(completed) if completed else 0,
            "by_model": self._group_by_model(filtered),
            "by_duration": self._group_by_duration(filtered),
            "by_user": self._group_by_user(filtered),
            "by_project": self._group_by_project(filtered)
        }

    def get_daily_breakdown(
        self,
        days: int = 30,
        user_id: str = None,
        project_id: str = None
    ) -> List[Dict]:
        """Get day-by-day breakdown."""
        end = datetime.utcnow()
        start = end - timedelta(days=days)

        daily = []
        current = start

        while current < end:
            next_day = current + timedelta(days=1)

            day_events = self._filter_events(current, next_day, user_id, project_id)

            daily.append({
                "date": current.strftime("%Y-%m-%d"),
                "generations": len(day_events),
                "credits": sum(e.credits_used for e in day_events),
                "completed": len([e for e in day_events if e.status == "completed"]),
                "failed": len([e for e in day_events if e.status == "failed"])
            })

            current = next_day

        return daily

    def get_top_users(self, limit: int = 10, days: int = 30) -> List[Dict]:
        """Get top users by usage."""
        start = datetime.utcnow() - timedelta(days=days)
        filtered = self._filter_events(start=start)

        user_stats = defaultdict(lambda: {"generations": 0, "credits": 0})

        for event in filtered:
            user_stats[event.user_id]["generations"] += 1
            user_stats[event.user_id]["credits"] += event.credits_used

        sorted_users = sorted(
            user_stats.items(),
            key=lambda x: x[1]["credits"],
            reverse=True
        )

        return [
            {"user_id": user_id, **stats}
            for user_id, stats in sorted_users[:limit]
        ]

    def get_popular_prompts(self, limit: int = 10, days: int = 30) -> List[Dict]:
        """Analyze popular prompt patterns."""
        start = datetime.utcnow() - timedelta(days=days)
        filtered = self._filter_events(start=start)

        # Simple word frequency analysis
        word_freq = defaultdict(int)
        for event in filtered:
            words = event.prompt.lower().split()
            for word in words:
                if len(word) > 3:  # Skip short words
                    word_freq[word] += 1

        sorted_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)

        return [
            {"term": word, "count": count}
            for word, count in sorted_words[:limit]
        ]

    def get_performance_metrics(self, days: int = 30) -> Dict:
        """Get performance metrics."""
        start = datetime.utcnow() - timedelta(days=days)
        filtered = self._filter_events(start=start)

        completed = [e for e in filtered if e.status == "completed"]

        if not completed:
            return {"no_data": True}

        gen_times = [e.generation_time_seconds for e in completed]
        gen_times.sort()

        return {
            "total_videos": len(completed),
            "generation_time": {
                "min": min(gen_times),
                "max": max(gen_times),
                "avg": sum(gen_times) / len(gen_times),
                "p50": gen_times[len(gen_times) // 2],
                "p95": gen_times[int(len(gen_times) * 0.95)],
                "p99": gen_times[int(len(gen_times) * 0.99)]
            },
            "failure_rate": len([e for e in filtered if e.status == "failed"]) / len(filtered) * 100,
            "avg_retries": 0  # Would need retry tracking
        }

    def detect_anomalies(self, days: int = 7) -> List[Dict]:
        """Detect usage anomalies."""
        daily = self.get_daily_breakdown(days=days * 2)

        if len(daily) < days:
            return []

        # Calculate baseline (first half)
        baseline = daily[:len(daily)//2]
        recent = daily[len(daily)//2:]

        avg_baseline = sum(d["generations"] for d in baseline) / len(baseline)
        std_baseline = (
            sum((d["generations"] - avg_baseline) ** 2 for d in baseline) / len(baseline)
        ) ** 0.5

        anomalies = []
        for day in recent:
            if std_baseline > 0:
                z_score = (day["generations"] - avg_baseline) / std_baseline
                if abs(z_score) > 2:
                    anomalies.append({
                        "date": day["date"],
                        "generations": day["generations"],
                        "expected": round(avg_baseline),
                        "z_score": round(z_score, 2),
                        "type": "spike" if z_score > 0 else "drop"
                    })

        return anomalies

    def _filter_events(
        self,
        start: datetime = None,
        end: datetime = None,
        user_id: str = None,
        project_id: str = None
    ) -> List[GenerationEvent]:
        """Filter events by criteria."""
        filtered = self.events

        if start:
            filtered = [e for e in filtered if e.timestamp >= start]
        if end:
            filtered = [e for e in filtered if e.timestamp < end]
        if user_id:
            filtered = [e for e in filtered if e.user_id == user_id]
        if project_id:
            filtered = [e for e in filtered if e.project_id == project_id]

        return filtered

    def _group_by_model(self, events: List[GenerationEvent]) -> Dict:
        """Group by model."""
        by_model = defaultdict(lambda: {"count": 0, "credits": 0})
        for e in events:
            by_model[e.model]["count"] += 1
            by_model[e.model]["credits"] += e.credits_used
        return dict(by_model)

    def _group_by_duration(self, events: List[GenerationEvent]) -> Dict:
        """Group by duration."""
        by_duration = defaultdict(lambda: {"count": 0, "credits": 0})
        for e in events:
            key = f"{e.duration}s"
            by_duration[key]["count"] += 1
            by_duration[key]["credits"] += e.credits_used
        return dict(by_duration)

    def _group_by_user(self, events: List[GenerationEvent]) -> Dict:
        """Group by user."""
        by_user = defaultdict(lambda: {"count": 0, "credits": 0})
        for e in events:
            by_user[e.user_id]["count"] += 1
            by_user[e.user_id]["credits"] += e.credits_used
        return dict(by_user)

    def _group_by_project(self, events: List[GenerationEvent]) -> Dict:
        """Group by project."""
        by_project = defaultdict(lambda: {"count": 0, "credits": 0})
        for e in events:
            by_project[e.project_id]["count"] += 1
            by_project[e.project_id]["credits"] += e.credits_used
        return dict(by_project)
```