59 lines
1.4 KiB
Python
59 lines
1.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import threading
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from config import DATA_DIR
|
||
|
|
|
||
|
|
FEATURES_DIR = DATA_DIR / "features"
|
||
|
|
_LOCK = threading.Lock()
|
||
|
|
_cache: dict = {}
|
||
|
|
|
||
|
|
|
||
|
|
def _ensure():
|
||
|
|
FEATURES_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
|
||
|
|
def _path(guild_id: int) -> Path:
|
||
|
|
return FEATURES_DIR / f"{guild_id}.json"
|
||
|
|
|
||
|
|
|
||
|
|
def get(guild_id: int, key: str = "") -> dict:
|
||
|
|
with _LOCK:
|
||
|
|
_ensure()
|
||
|
|
if guild_id not in _cache:
|
||
|
|
f = _path(guild_id)
|
||
|
|
if f.exists():
|
||
|
|
with open(f, "r", encoding="utf-8") as fh:
|
||
|
|
_cache[guild_id] = json.load(fh)
|
||
|
|
else:
|
||
|
|
_cache[guild_id] = {}
|
||
|
|
data = _cache[guild_id]
|
||
|
|
return data.get(key, {}) if key else data
|
||
|
|
|
||
|
|
|
||
|
|
def set(guild_id: int, key: str, value) -> None:
|
||
|
|
with _LOCK:
|
||
|
|
_ensure()
|
||
|
|
if guild_id not in _cache:
|
||
|
|
f = _path(guild_id)
|
||
|
|
if f.exists():
|
||
|
|
with open(f, "r", encoding="utf-8") as fh:
|
||
|
|
_cache[guild_id] = json.load(fh)
|
||
|
|
else:
|
||
|
|
_cache[guild_id] = {}
|
||
|
|
_cache[guild_id][key] = value
|
||
|
|
_save(guild_id)
|
||
|
|
|
||
|
|
|
||
|
|
def _save(guild_id: int) -> None:
|
||
|
|
_ensure()
|
||
|
|
with open(_path(guild_id), "w", encoding="utf-8") as fh:
|
||
|
|
json.dump(_cache[guild_id], fh, indent=2, ensure_ascii=False)
|
||
|
|
|
||
|
|
|
||
|
|
def section(guild_id: int, name: str) -> dict:
|
||
|
|
data = get(guild_id)
|
||
|
|
return data.get(name, {})
|