Spaces:
Sleeping
Sleeping
File size: 6,337 Bytes
ed6391b | 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 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 | """
Score Persistence Service - Save and retrieve analytics scores from Supabase
"""
import logging
from typing import Dict, Any, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class ScorePersistenceService:
"""Service to persist computed analytics scores to Supabase"""
def __init__(self, supabase_client):
self.supabase = supabase_client
self.table_name = 'analytics_scores'
def save_score(self, student_id: str, score_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Save or update a student's analytics score
Args:
student_id: Student identifier
score_data: Dictionary containing score data with keys:
- final_score
- component_scores (universal, personality, text)
- domain, domain_confidence
- fidelity_score
- strengths, improvement_areas, career_suggestions, etc.
Returns:
Saved record or error
"""
try:
# Determine tier based on final score
final_score = score_data.get('final_score', 0)
tier = self._get_tier(final_score)
# Extract component scores
component_scores = score_data.get('component_scores', {})
# Prepare record
record = {
'student_id': student_id,
'final_score': final_score,
'tier': tier,
'universal_score': component_scores.get('universal', 0),
'personality_score': component_scores.get('personality', 0),
'text_score': component_scores.get('text', 0),
'fidelity_score': score_data.get('fidelity_score'),
'detected_domain': score_data.get('domain'),
'domain_confidence': score_data.get('domain_confidence'),
'full_report': score_data, # Store complete report as JSON
'updated_at': datetime.utcnow().isoformat()
}
# Upsert (insert or update)
result = self.supabase.table(self.table_name).upsert(
record,
on_conflict='student_id'
).execute()
logger.info(f"Saved score for student {student_id}: {final_score}")
return {'success': True, 'data': result.data}
except Exception as e:
logger.error(f"Failed to save score for {student_id}: {e}")
return {'success': False, 'error': str(e)}
def get_score(self, student_id: str) -> Optional[Dict[str, Any]]:
"""
Get a student's saved analytics score
Args:
student_id: Student identifier
Returns:
Score record or None
"""
try:
result = self.supabase.table(self.table_name).select('*').eq(
'student_id', student_id
).single().execute()
if result.data:
return result.data
return None
except Exception as e:
logger.error(f"Failed to get score for {student_id}: {e}")
return None
def get_scores_by_tier(self, tier: str, limit: int = 100) -> list:
"""Get all scores in a specific tier"""
try:
result = self.supabase.table(self.table_name).select('*').eq(
'tier', tier
).limit(limit).execute()
return result.data or []
except Exception as e:
logger.error(f"Failed to get scores by tier {tier}: {e}")
return []
def get_scores_by_domain(self, domain: str, limit: int = 100) -> list:
"""Get all scores in a specific domain"""
try:
result = self.supabase.table(self.table_name).select('*').eq(
'detected_domain', domain
).limit(limit).execute()
return result.data or []
except Exception as e:
logger.error(f"Failed to get scores by domain {domain}: {e}")
return []
def get_batch_stats(self) -> Dict[str, Any]:
"""Get aggregate statistics for all scores"""
try:
# Get all scores
result = self.supabase.table(self.table_name).select('*').execute()
scores = result.data or []
if not scores:
return {'total': 0}
# Calculate stats
final_scores = [s['final_score'] for s in scores if s['final_score']]
tier_counts = {}
domain_counts = {}
for s in scores:
tier = s.get('tier', 'Unknown')
domain = s.get('detected_domain', 'Unknown')
tier_counts[tier] = tier_counts.get(tier, 0) + 1
domain_counts[domain] = domain_counts.get(domain, 0) + 1
return {
'total': len(scores),
'average_score': sum(final_scores) / len(final_scores) if final_scores else 0,
'min_score': min(final_scores) if final_scores else 0,
'max_score': max(final_scores) if final_scores else 0,
'tier_distribution': tier_counts,
'domain_distribution': domain_counts
}
except Exception as e:
logger.error(f"Failed to get batch stats: {e}")
return {'error': str(e)}
def _get_tier(self, score: float) -> str:
"""Determine tier based on score"""
if score >= 80:
return 'Excellent'
elif score >= 60:
return 'Good'
elif score >= 40:
return 'Average'
return 'Needs Work'
# Singleton instance
_persistence_service = None
def get_score_persistence_service(supabase_client=None):
"""Get or create singleton ScorePersistenceService"""
global _persistence_service
if _persistence_service is None:
if supabase_client is None:
from database.db import get_supabase
supabase_client = get_supabase()
_persistence_service = ScorePersistenceService(supabase_client)
return _persistence_service
|