Spaces:
Running
Running
File size: 5,280 Bytes
74626f2 8f2ee4f 74626f2 8f2ee4f 74626f2 | 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 | import json
import os
from datetime import datetime
# HF Native Persistence: Check if /data volume is mounted
def get_db_file_path():
# Primary choice: HF Persistent Storage Mount
if os.path.exists('/data'):
# Use a subdirectory to avoid permission issues at the root mount point
test_path = '/data/db/db.json'
try:
os.makedirs(os.path.dirname(test_path), exist_ok=True)
# Verify write access
with open(os.path.join(os.path.dirname(test_path), '.write_test'), 'w') as f:
f.write('test')
os.remove(os.path.join(os.path.dirname(test_path), '.write_test'))
print(f"📡 HF Native Persistence: Verified. Storing data in {test_path}")
return test_path
except Exception as e:
print(f"⚠️ HF Native Persistence: (/data) exists but is not writable: {e}")
# Fallback: Local Storage
local_path = os.path.join(os.path.dirname(__file__), 'data', 'db.json')
os.makedirs(os.path.dirname(local_path), exist_ok=True)
print(f"💻 Local Persistence: Active. Storing data in {local_path}")
return local_path
DB_FILE = get_db_file_path()
# Ensure final data directory exists
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
def init_db():
if not os.path.exists(DB_FILE):
data = {
"users": [],
"learning_sessions": {},
"polylines": {},
"summaries": [],
"bookmarks": {}, # session_id -> list of resource_ids
"notes": {}, # session_id -> list of note objects
"lectures": [] # list of lecture objects
}
save_db(data)
def reset_db():
if os.path.exists(DB_FILE):
os.remove(DB_FILE)
init_db()
def load_db():
if not os.path.exists(DB_FILE):
init_db()
try:
with open(DB_FILE, 'r') as f:
content = f.read().strip()
if not content:
init_db()
with open(DB_FILE, 'r') as f2:
db = json.load(f2)
else:
db = json.loads(content)
if "bookmarks" not in db:
db["bookmarks"] = {}
save_db(db)
return db
except (json.JSONDecodeError, FileNotFoundError):
init_db()
with open(DB_FILE, 'r') as f:
return json.load(f)
def save_db(data):
# Ensure directory exists (in case /data was just mounted)
os.makedirs(os.path.dirname(DB_FILE), exist_ok=True)
with open(DB_FILE, 'w') as f:
json.dump(data, f, indent=4)
def get_session(session_id):
db = load_db()
if session_id not in db["learning_sessions"]:
db["learning_sessions"][session_id] = {
'position': {'x': 10, 'y': 10},
'level': 0,
'totalReward': 0,
'visitedResources': [],
'notifications': [
{
'id': 'initial',
'type': 'info',
'message': 'Welcome back to the Intelligence Hub. Neural Sync complete.',
'timestamp': int(datetime.now().timestamp() * 1000),
'read': False
}
]
}
save_db(db)
return db["learning_sessions"][session_id]
def update_session(session_id, session_data):
db = load_db()
db["learning_sessions"][session_id] = session_data
save_db(db)
def save_summary(summary_data):
db = load_db()
if "summaries" not in db:
db["summaries"] = []
db["summaries"].append(summary_data)
save_db(db)
def save_polyline(polyline_id, polyline_data):
db = load_db()
db["polylines"][polyline_id] = polyline_data
save_db(db)
def get_polylines():
db = load_db()
return db["polylines"]
def get_bookmarks(session_id):
db = load_db()
return db["bookmarks"].get(session_id, [])
def add_bookmark(session_id, resource_id):
db = load_db()
if session_id not in db["bookmarks"]:
db["bookmarks"][session_id] = []
if resource_id not in db["bookmarks"][session_id]:
db["bookmarks"][session_id].append(resource_id)
save_db(db)
def remove_bookmark(session_id, resource_id):
db = load_db()
if session_id in db["bookmarks"] and resource_id in db["bookmarks"][session_id]:
db["bookmarks"][session_id].remove(resource_id)
save_db(db)
def get_notes(session_id):
db = load_db()
if "notes" not in db or isinstance(db["notes"], list):
db["notes"] = {}
save_db(db)
return db["notes"].get(session_id, [])
def add_note(session_id, note_data):
db = load_db()
if "notes" not in db or isinstance(db["notes"], list):
db["notes"] = {}
if session_id not in db["notes"]:
db["notes"][session_id] = []
# Simple ID generation if not provided
if "id" not in note_data:
note_data["id"] = f"note_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
if "createdAt" not in note_data:
note_data["createdAt"] = datetime.now().isoformat()
db["notes"][session_id].append(note_data)
save_db(db)
return note_data
def get_lectures():
db = load_db()
return db.get("lectures", [])
|