""" AI Usage data model for tracking AI generation costs and usage. This module defines the AIUsageLog dataclass which represents a single AI usage event for tracking costs, tokens used, and generating usage analytics. """ from dataclasses import dataclass, field from datetime import datetime, timezone, date from typing import Dict, Any, Optional from enum import Enum class TaskType(str, Enum): """Types of AI tasks that can be tracked.""" STORY_PROGRESSION = "story_progression" COMBAT_NARRATION = "combat_narration" QUEST_SELECTION = "quest_selection" NPC_DIALOGUE = "npc_dialogue" GENERAL = "general" @dataclass class AIUsageLog: """ Represents a single AI usage event for cost and usage tracking. This dataclass captures all relevant information about an AI API call including the user, model used, tokens consumed, and estimated cost. Used for: - Cost monitoring and budgeting - Usage analytics per user/tier - Rate limiting enforcement - Billing and invoicing (future) Attributes: log_id: Unique identifier for this usage log entry user_id: User who made the request timestamp: When the request was made model: Model identifier (e.g., "meta/meta-llama-3-8b-instruct") tokens_input: Number of input tokens (prompt) tokens_output: Number of output tokens (response) tokens_total: Total tokens used (input + output) estimated_cost: Estimated cost in USD task_type: Type of task (story, combat, quest, npc) session_id: Optional game session ID for context character_id: Optional character ID for context request_duration_ms: How long the request took in milliseconds success: Whether the request completed successfully error_message: Error message if the request failed """ log_id: str user_id: str timestamp: datetime model: str tokens_input: int tokens_output: int tokens_total: int estimated_cost: float task_type: TaskType session_id: Optional[str] = None character_id: Optional[str] = None request_duration_ms: int = 0 success: bool = True error_message: Optional[str] = None def to_dict(self) -> Dict[str, Any]: """ Convert usage log to dictionary for storage. Returns: Dictionary representation suitable for Appwrite storage """ return { "user_id": self.user_id, "timestamp": self.timestamp.isoformat() if isinstance(self.timestamp, datetime) else self.timestamp, "model": self.model, "tokens_input": self.tokens_input, "tokens_output": self.tokens_output, "tokens_total": self.tokens_total, "estimated_cost": self.estimated_cost, "task_type": self.task_type.value if isinstance(self.task_type, TaskType) else self.task_type, "session_id": self.session_id, "character_id": self.character_id, "request_duration_ms": self.request_duration_ms, "success": self.success, "error_message": self.error_message, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "AIUsageLog": """ Create AIUsageLog from dictionary. Args: data: Dictionary with usage log data Returns: AIUsageLog instance """ # Parse timestamp timestamp = data.get("timestamp") if isinstance(timestamp, str): timestamp = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) elif timestamp is None: timestamp = datetime.now(timezone.utc) # Parse task type task_type = data.get("task_type", "general") if isinstance(task_type, str): try: task_type = TaskType(task_type) except ValueError: task_type = TaskType.GENERAL return cls( log_id=data.get("log_id", ""), user_id=data.get("user_id", ""), timestamp=timestamp, model=data.get("model", ""), tokens_input=data.get("tokens_input", 0), tokens_output=data.get("tokens_output", 0), tokens_total=data.get("tokens_total", 0), estimated_cost=data.get("estimated_cost", 0.0), task_type=task_type, session_id=data.get("session_id"), character_id=data.get("character_id"), request_duration_ms=data.get("request_duration_ms", 0), success=data.get("success", True), error_message=data.get("error_message"), ) @dataclass class DailyUsageSummary: """ Summary of AI usage for a specific day. Used for reporting and rate limiting checks. Attributes: date: The date of this summary user_id: User ID total_requests: Number of AI requests made total_tokens: Total tokens consumed total_input_tokens: Total input tokens total_output_tokens: Total output tokens estimated_cost: Total estimated cost in USD requests_by_task: Breakdown of requests by task type """ date: date user_id: str total_requests: int total_tokens: int total_input_tokens: int total_output_tokens: int estimated_cost: float requests_by_task: Dict[str, int] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert summary to dictionary.""" return { "date": self.date.isoformat() if isinstance(self.date, date) else self.date, "user_id": self.user_id, "total_requests": self.total_requests, "total_tokens": self.total_tokens, "total_input_tokens": self.total_input_tokens, "total_output_tokens": self.total_output_tokens, "estimated_cost": self.estimated_cost, "requests_by_task": self.requests_by_task, } @dataclass class MonthlyUsageSummary: """ Summary of AI usage for a specific month. Used for billing and cost projections. Attributes: year: Year month: Month (1-12) user_id: User ID total_requests: Number of AI requests made total_tokens: Total tokens consumed estimated_cost: Total estimated cost in USD daily_breakdown: List of daily summaries """ year: int month: int user_id: str total_requests: int total_tokens: int estimated_cost: float daily_breakdown: list = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: """Convert summary to dictionary.""" return { "year": self.year, "month": self.month, "user_id": self.user_id, "total_requests": self.total_requests, "total_tokens": self.total_tokens, "estimated_cost": self.estimated_cost, "daily_breakdown": self.daily_breakdown, }