85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
import json
|
|
import threading
|
|
import shutil
|
|
import os
|
|
from typing import Callable, Any
|
|
|
|
class DataManager:
|
|
def __init__(self, json_path: str):
|
|
self.json_path = json_path
|
|
self.lock = threading.Lock()
|
|
self._data = self._load()
|
|
|
|
def _load(self):
|
|
try:
|
|
with open(self.json_path, 'r') as f:
|
|
return json.load(f)
|
|
except FileNotFoundError:
|
|
default = {
|
|
'agreed_rules': [],
|
|
'agreed_engagement': [],
|
|
'agreed_nickname': [],
|
|
'nick_same_confirmed': [],
|
|
'nick_nudged': [],
|
|
'nick_dm_map': [],
|
|
'pirates': [],
|
|
'modlog': [],
|
|
'reports': [],
|
|
'encounters': [],
|
|
'vc_channels': [],
|
|
'user_cards': [],
|
|
'pirates_list_posts': [],
|
|
'spicepay_prefs': [],
|
|
}
|
|
self._save(default)
|
|
return default
|
|
|
|
def _safe_write(self, data: dict):
|
|
tmp = self.json_path + ".tmp"
|
|
with open(tmp, 'w') as f:
|
|
json.dump(data, f, indent=4)
|
|
# backup current file (best-effort)
|
|
if os.path.exists(self.json_path):
|
|
try:
|
|
shutil.copy2(self.json_path, self.json_path + ".bak")
|
|
except Exception:
|
|
pass
|
|
os.replace(tmp, self.json_path)
|
|
|
|
def _save(self, data: dict):
|
|
# single place to write (atomic replace + rolling .bak)
|
|
self._safe_write(data)
|
|
|
|
def get(self, category: str):
|
|
with self.lock:
|
|
return list(self._data.get(category, []))
|
|
|
|
def add(self, category: str, item: Any):
|
|
with self.lock:
|
|
self._data.setdefault(category, []).append(item)
|
|
self._save(self._data)
|
|
|
|
def remove(self, category: str, predicate: Callable[[Any], bool]):
|
|
with self.lock:
|
|
arr = self._data.get(category, [])
|
|
self._data[category] = [i for i in arr if not predicate(i)]
|
|
self._save(self._data)
|
|
|
|
def update(self, category: str, predicate: Callable[[Any], bool], updater: Callable[[dict], dict]) -> bool:
|
|
"""
|
|
Atomically find one item in `category` matching predicate and update it with `updater`.
|
|
Returns True if an item was updated, else False.
|
|
"""
|
|
with self.lock:
|
|
arr = self._data.get(category, [])
|
|
for idx, item in enumerate(arr):
|
|
if predicate(item):
|
|
# Copy → mutate → save back
|
|
new_item = dict(item)
|
|
new_item = updater(new_item) or new_item
|
|
arr[idx] = new_item
|
|
self._data[category] = arr
|
|
self._save(self._data)
|
|
return True
|
|
return False
|