Spaces:
Sleeping
Sleeping
| """ | |
| Report Storage Service - Store analytics reports in Supabase Storage (S3 bucket) | |
| """ | |
| import json | |
| import logging | |
| from typing import Dict, Any, Optional | |
| from datetime import datetime | |
| import io | |
| logger = logging.getLogger(__name__) | |
| # Bucket name for analytics reports | |
| REPORTS_BUCKET = 'analytics-reports' | |
| class ReportStorageService: | |
| """Service to store/retrieve analytics reports from Supabase Storage""" | |
| def __init__(self, supabase_client): | |
| self.supabase = supabase_client | |
| self.bucket = REPORTS_BUCKET | |
| self._ensure_bucket_exists() | |
| def _ensure_bucket_exists(self): | |
| """Create bucket if it doesn't exist""" | |
| try: | |
| # List buckets to check if ours exists | |
| buckets = self.supabase.storage.list_buckets() | |
| bucket_names = [b.name for b in buckets] | |
| if self.bucket not in bucket_names: | |
| # Create the bucket (public=False for private access) | |
| self.supabase.storage.create_bucket( | |
| self.bucket, | |
| options={'public': False} | |
| ) | |
| logger.info(f"Created storage bucket: {self.bucket}") | |
| except Exception as e: | |
| logger.warning(f"Could not check/create bucket: {e}") | |
| def save_report(self, student_id: str, report_data: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Save a student's analytics report as JSON file in Supabase Storage | |
| Args: | |
| student_id: Student identifier | |
| report_data: Complete report data to save | |
| Returns: | |
| Success status with file path or error | |
| """ | |
| try: | |
| # File path: reports/{student_id}/report_{timestamp}.json | |
| timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S') | |
| file_path = f"reports/{student_id}/report_{timestamp}.json" | |
| # Also save a "latest" copy for easy retrieval | |
| latest_path = f"reports/{student_id}/latest.json" | |
| # Add metadata to the report | |
| report_with_meta = { | |
| **report_data, | |
| 'meta': { | |
| 'student_id': student_id, | |
| 'generated_at': datetime.utcnow().isoformat(), | |
| 'version': '1.0' | |
| } | |
| } | |
| # Convert to JSON bytes | |
| json_bytes = json.dumps(report_with_meta, indent=2).encode('utf-8') | |
| # Upload timestamped version | |
| self.supabase.storage.from_(self.bucket).upload( | |
| file_path, | |
| json_bytes, | |
| file_options={'content-type': 'application/json'} | |
| ) | |
| # Upload/overwrite latest version | |
| try: | |
| self.supabase.storage.from_(self.bucket).remove([latest_path]) | |
| except: | |
| pass # File might not exist yet | |
| self.supabase.storage.from_(self.bucket).upload( | |
| latest_path, | |
| json_bytes, | |
| file_options={'content-type': 'application/json'} | |
| ) | |
| logger.info(f"Saved report for {student_id} at {file_path}") | |
| return { | |
| 'success': True, | |
| 'file_path': file_path, | |
| 'latest_path': latest_path, | |
| 'timestamp': timestamp | |
| } | |
| except Exception as e: | |
| logger.error(f"Failed to save report for {student_id}: {e}") | |
| return {'success': False, 'error': str(e)} | |
| def get_latest_report(self, student_id: str) -> Optional[Dict[str, Any]]: | |
| """ | |
| Get the latest report for a student | |
| Args: | |
| student_id: Student identifier | |
| Returns: | |
| Report data or None | |
| """ | |
| try: | |
| latest_path = f"reports/{student_id}/latest.json" | |
| # Download file | |
| response = self.supabase.storage.from_(self.bucket).download(latest_path) | |
| if response: | |
| return json.loads(response.decode('utf-8')) | |
| return None | |
| except Exception as e: | |
| logger.error(f"Failed to get report for {student_id}: {e}") | |
| return None | |
| def get_report_by_path(self, file_path: str) -> Optional[Dict[str, Any]]: | |
| """Get a specific report by its file path""" | |
| try: | |
| response = self.supabase.storage.from_(self.bucket).download(file_path) | |
| if response: | |
| return json.loads(response.decode('utf-8')) | |
| return None | |
| except Exception as e: | |
| logger.error(f"Failed to get report at {file_path}: {e}") | |
| return None | |
| def list_student_reports(self, student_id: str) -> list: | |
| """List all reports for a student""" | |
| try: | |
| folder_path = f"reports/{student_id}" | |
| files = self.supabase.storage.from_(self.bucket).list(folder_path) | |
| return [ | |
| { | |
| 'name': f['name'], | |
| 'path': f"{folder_path}/{f['name']}", | |
| 'created_at': f.get('created_at'), | |
| 'size': f.get('metadata', {}).get('size') | |
| } | |
| for f in files if f['name'] != 'latest.json' | |
| ] | |
| except Exception as e: | |
| logger.error(f"Failed to list reports for {student_id}: {e}") | |
| return [] | |
| def get_signed_url(self, file_path: str, expires_in: int = 3600) -> Optional[str]: | |
| """ | |
| Get a signed URL for downloading a report | |
| Args: | |
| file_path: Path to the file in storage | |
| expires_in: URL expiry time in seconds (default 1 hour) | |
| Returns: | |
| Signed URL or None | |
| """ | |
| try: | |
| result = self.supabase.storage.from_(self.bucket).create_signed_url( | |
| file_path, | |
| expires_in | |
| ) | |
| return result.get('signedURL') | |
| except Exception as e: | |
| logger.error(f"Failed to create signed URL: {e}") | |
| return None | |
| def delete_report(self, file_path: str) -> bool: | |
| """Delete a specific report""" | |
| try: | |
| self.supabase.storage.from_(self.bucket).remove([file_path]) | |
| return True | |
| except Exception as e: | |
| logger.error(f"Failed to delete report: {e}") | |
| return False | |
| # Singleton instance | |
| _storage_service = None | |
| def get_report_storage_service(supabase_client=None): | |
| """Get or create singleton ReportStorageService""" | |
| global _storage_service | |
| if _storage_service is None: | |
| if supabase_client is None: | |
| from database.db import get_supabase | |
| supabase_client = get_supabase() | |
| _storage_service = ReportStorageService(supabase_client) | |
| return _storage_service | |