Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |