| """Session management — per-user data isolation.""" |
|
|
| import hashlib |
| import os |
| from pathlib import Path |
| from contextvars import ContextVar |
|
|
| import config |
|
|
| USERS_DIR = config.DATA_DIR / "users" |
| _current_user_id: ContextVar[str] = ContextVar("current_user_id", default="default") |
|
|
|
|
| def set_current_user(user_id: str): |
| if not user_id or user_id in ("None", "default", ""): |
| user_id = hashlib.md5(os.urandom(8)).hexdigest()[:12] |
| _current_user_id.set(str(user_id)) |
|
|
|
|
| def get_current_user_id() -> str: |
| return _current_user_id.get() |
|
|
|
|
| def get_user_dir(user_id: str = None) -> Path: |
| uid = user_id or get_current_user_id() |
| safe_id = hashlib.md5(uid.encode()).hexdigest()[:12] |
| user_dir = USERS_DIR / safe_id |
| user_dir.mkdir(parents=True, exist_ok=True) |
| (user_dir / "saves").mkdir(exist_ok=True) |
| return user_dir |
|
|
|
|
| def get_user_world_file(user_id: str = None) -> Path: |
| return get_user_dir(user_id) / "world.json" |
|
|
|
|
| def get_user_saves_dir(user_id: str = None) -> Path: |
| return get_user_dir(user_id) / "saves" |
|
|
|
|
| def get_user_active_slot_file(user_id: str = None) -> Path: |
| return get_user_dir(user_id) / "active_slot.txt" |
|
|
|
|
| def ensure_session_id(sid) -> str: |
| """Ensure we have a valid session ID. Generate one if missing.""" |
| if sid and isinstance(sid, str) and sid != "None" and len(sid) > 4: |
| return sid |
| return hashlib.md5(os.urandom(8)).hexdigest()[:12] |
|
|