Spaces:
Sleeping
Sleeping
File size: 4,790 Bytes
1d4dc07 f0b765c 838cd23 f0b765c 838cd23 1d4dc07 04aa1ba 1d4dc07 04aa1ba 1d4dc07 04aa1ba 1d4dc07 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """
Data models for analytics collections.
"""
import uuid
from datetime import datetime
from typing import Optional, Dict, Any
from pydantic import BaseModel, Field, field_validator
class Session(BaseModel):
"""Session analytics model"""
session_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
start_time: datetime = Field(default_factory=datetime.utcnow)
end_time: Optional[datetime] = None
message_count: int = 0
search_used: bool = False
user_agent: Optional[str] = None
status: str = "active" # active, ended
user_id: Optional[str] = None
@field_validator('user_id')
@classmethod
def validate_user_id(cls, v):
"""Validate user_id format"""
if v is not None:
if not isinstance(v, str):
raise ValueError('user_id must be a string')
if v.strip() == '':
return None # Treat empty string as None (anonymous)
if len(v) > 255:
raise ValueError('user_id must be 255 characters or less')
# Allow ASCII alphanumeric, hyphens, and underscores
if not all((c.isascii() and c.isalnum()) or c in '-_' for c in v):
raise ValueError('user_id can only contain alphanumeric characters, hyphens, and underscores')
return v
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for MongoDB insertion"""
data = self.dict()
data["_id"] = self.session_id
return data
def end_session(self):
"""Mark session as ended"""
self.end_time = datetime.utcnow()
self.status = "ended"
@property
def duration_seconds(self) -> Optional[int]:
"""Calculate session duration in seconds"""
if self.end_time:
return int((self.end_time - self.start_time).total_seconds())
return None
class Message(BaseModel):
"""Message analytics model"""
message_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
session_id: str
timestamp: datetime = Field(default_factory=datetime.utcnow)
prompt_length: int
response_length: int
used_search: bool = False
response_time_ms: int
max_tokens: int = 500
temperature: float = 0.7
success: bool = True
error_message: Optional[str] = None
user_id: Optional[str] = None
@field_validator('user_id')
@classmethod
def validate_user_id(cls, v):
"""Validate user_id format"""
if v is not None:
if not isinstance(v, str):
raise ValueError('user_id must be a string')
if v.strip() == '':
return None # Treat empty string as None (anonymous)
if len(v) > 255:
raise ValueError('user_id must be 255 characters or less')
# Allow ASCII alphanumeric, hyphens, and underscores
if not all((c.isascii() and c.isalnum()) or c in '-_' for c in v):
raise ValueError('user_id can only contain alphanumeric characters, hyphens, and underscores')
return v
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for MongoDB insertion"""
data = self.dict()
data["_id"] = self.message_id
return data
class SearchAnalytics(BaseModel):
"""Search analytics model"""
search_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
message_id: str
timestamp: datetime = Field(default_factory=datetime.utcnow)
search_query: str
search_terms: list[str] = []
brave_results: int = 0
duckduckgo_results: int = 0
total_unique_results: int = 0
brave_response_time_ms: int = 0
duckduckgo_response_time_ms: int = 0
search_engines_used: list[str] = []
search_success: bool = True
fallback_used: bool = False
user_id: Optional[str] = None
@field_validator('user_id')
@classmethod
def validate_user_id(cls, v):
"""Validate user_id format"""
if v is not None:
if not isinstance(v, str):
raise ValueError('user_id must be a string')
if v.strip() == '':
return None # Treat empty string as None (anonymous)
if len(v) > 255:
raise ValueError('user_id must be 255 characters or less')
# Allow ASCII alphanumeric, hyphens, and underscores
if not all((c.isascii() and c.isalnum()) or c in '-_' for c in v):
raise ValueError('user_id can only contain alphanumeric characters, hyphens, and underscores')
return v
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for MongoDB insertion"""
data = self.dict()
data["_id"] = self.search_id
return data |