Spaces:
Sleeping
Sleeping
File size: 7,873 Bytes
2a317a3 | 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 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 | """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
|