Spaces:
Sleeping
Sleeping
| """Local file-based storage fallback for analytics data""" | |
| import json | |
| import os | |
| from datetime import datetime | |
| from typing import Dict, Any, Optional, List | |
| import uuid | |
| # Storage directory - works on HuggingFace Spaces in /tmp | |
| STORAGE_DIR = os.getenv('ANALYTICS_STORAGE_DIR', '/tmp/analytics_data') | |
| class LocalStorage: | |
| """Simple file-based storage that mimics Supabase responses""" | |
| def __init__(self): | |
| self.storage_dir = STORAGE_DIR | |
| os.makedirs(self.storage_dir, exist_ok=True) | |
| print(f"[LocalStorage] Initialized at: {self.storage_dir}") | |
| # Initialize data files | |
| self.students_file = os.path.join(self.storage_dir, 'students.json') | |
| self.personality_file = os.path.join(self.storage_dir, 'personality.json') | |
| self.text_file = os.path.join(self.storage_dir, 'text.json') | |
| self.domain_file = os.path.join(self.storage_dir, 'domain.json') | |
| # Load existing data | |
| self._students = self._load(self.students_file, {}) | |
| self._personality = self._load(self.personality_file, {}) | |
| self._text = self._load(self.text_file, {}) | |
| self._domain = self._load(self.domain_file, {}) | |
| def _load(self, filepath: str, default: Any) -> Any: | |
| """Load JSON file or return default""" | |
| try: | |
| if os.path.exists(filepath): | |
| with open(filepath, 'r') as f: | |
| return json.load(f) | |
| except Exception as e: | |
| print(f"[LocalStorage] Error loading {filepath}: {e}") | |
| return default | |
| def _save(self, filepath: str, data: Any): | |
| """Save data to JSON file""" | |
| try: | |
| with open(filepath, 'w') as f: | |
| json.dump(data, f, indent=2, default=str) | |
| except Exception as e: | |
| print(f"[LocalStorage] Error saving {filepath}: {e}") | |
| def table(self, table_name: str) -> 'LocalTable': | |
| """Return a table-like interface""" | |
| return LocalTable(self, table_name) | |
| def get_data(self, table_name: str) -> Dict: | |
| """Get raw data for a table""" | |
| mapping = { | |
| 'analytics_students': (self._students, self.students_file), | |
| 'analytics_personality_responses': (self._personality, self.personality_file), | |
| 'analytics_text_responses': (self._text, self.text_file), | |
| 'analytics_domain_evidence': (self._domain, self.domain_file), | |
| } | |
| return mapping.get(table_name, ({}, None)) | |
| def set_data(self, table_name: str, key: str, value: Dict): | |
| """Set data for a table""" | |
| data, filepath = self.get_data(table_name) | |
| if data is not None: | |
| data[key] = value | |
| self._save(filepath, data) | |
| class LocalTable: | |
| """Mimics Supabase table interface for local storage""" | |
| def __init__(self, storage: LocalStorage, table_name: str): | |
| self.storage = storage | |
| self.table_name = table_name | |
| self._query = {} | |
| self._select_cols = '*' | |
| def select(self, columns: str = '*') -> 'LocalTable': | |
| self._select_cols = columns | |
| return self | |
| def eq(self, column: str, value: Any) -> 'LocalTable': | |
| self._query[column] = value | |
| return self | |
| def maybe_single(self) -> 'LocalTable': | |
| self._single = True | |
| return self | |
| def single(self) -> 'LocalTable': | |
| self._single = True | |
| return self | |
| def execute(self) -> 'LocalResult': | |
| """Execute the query or write operation""" | |
| # Check if this is a write operation | |
| if hasattr(self, '_insert_data'): | |
| return self._do_insert() | |
| if hasattr(self, '_upsert_data'): | |
| return self._do_upsert() | |
| if hasattr(self, '_update_data'): | |
| return self._do_update() | |
| # Otherwise it's a read/select operation | |
| data, _ = self.storage.get_data(self.table_name) | |
| if self._query: | |
| # Filter by query | |
| results = [] | |
| for key, record in data.items(): | |
| match = all(record.get(k) == v for k, v in self._query.items()) | |
| if match: | |
| results.append(record) | |
| if getattr(self, '_single', False) and results: | |
| return LocalResult(results[0]) | |
| elif getattr(self, '_single', False): | |
| return LocalResult(None) | |
| return LocalResult(results) | |
| else: | |
| # Return all | |
| return LocalResult(list(data.values())) | |
| def insert(self, record: Dict) -> 'LocalTable': | |
| """Insert a record""" | |
| self._insert_data = record | |
| return self | |
| def upsert(self, record: Dict) -> 'LocalTable': | |
| """Upsert a record""" | |
| self._upsert_data = record | |
| return self | |
| def update(self, record: Dict) -> 'LocalTable': | |
| """Update records""" | |
| self._update_data = record | |
| return self | |
| def _do_insert(self) -> 'LocalResult': | |
| """Perform insert""" | |
| record = getattr(self, '_insert_data', {}) | |
| data, filepath = self.storage.get_data(self.table_name) | |
| # Generate ID if needed | |
| if 'id' not in record: | |
| record['id'] = str(uuid.uuid4()) | |
| # Use student_id as key if present | |
| key = record.get('student_id', record.get('id')) | |
| record['created_at'] = datetime.utcnow().isoformat() | |
| data[key] = record | |
| self.storage._save(filepath, data) | |
| print(f"[LocalStorage] Inserted into {self.table_name}: {key}") | |
| return LocalResult([record]) | |
| def _do_upsert(self) -> 'LocalResult': | |
| """Perform upsert""" | |
| record = getattr(self, '_upsert_data', {}) | |
| data, filepath = self.storage.get_data(self.table_name) | |
| # Use student_id as key | |
| key = record.get('student_id', str(uuid.uuid4())) | |
| # Merge with existing if present | |
| if key in data: | |
| existing = data[key] | |
| existing.update(record) | |
| existing['updated_at'] = datetime.utcnow().isoformat() | |
| record = existing | |
| else: | |
| record['id'] = str(uuid.uuid4()) | |
| record['created_at'] = datetime.utcnow().isoformat() | |
| data[key] = record | |
| self.storage._save(filepath, data) | |
| print(f"[LocalStorage] Upserted into {self.table_name}: {key}") | |
| return LocalResult([record]) | |
| def _do_update(self) -> 'LocalResult': | |
| """Perform update on matching records""" | |
| updates = getattr(self, '_update_data', {}) | |
| data, filepath = self.storage.get_data(self.table_name) | |
| updated = [] | |
| for key, record in data.items(): | |
| match = all(record.get(k) == v for k, v in self._query.items()) | |
| if match: | |
| record.update(updates) | |
| record['updated_at'] = datetime.utcnow().isoformat() | |
| updated.append(record) | |
| self.storage._save(filepath, data) | |
| print(f"[LocalStorage] Updated {len(updated)} records in {self.table_name}") | |
| return LocalResult(updated) | |
| class LocalResult: | |
| """Mimics Supabase result object""" | |
| def __init__(self, data: Any): | |
| if data is None: | |
| self.data = None | |
| elif isinstance(data, list): | |
| self.data = data | |
| else: | |
| self.data = data | |
| def execute(self) -> 'LocalResult': | |
| """For chained calls that end with execute()""" | |
| return self | |
| # Singleton instance | |
| _local_storage: LocalStorage = None | |
| def get_local_storage() -> LocalStorage: | |
| """Get or create local storage instance""" | |
| global _local_storage | |
| if _local_storage is None: | |
| _local_storage = LocalStorage() | |
| return _local_storage | |