diff --git a/data_manager.py b/data_manager.py deleted file mode 100644 index 453d7e4..0000000 --- a/data_manager.py +++ /dev/null @@ -1,90 +0,0 @@ -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': [], - 'nick_verified': [], - 'nick_claim_pending': [], - 'nick_reviews': [], - 'rr_msg_channels': [], - } - self._save(default) - return default - - def _safe_write(self, data: dict): - # ensure parent dir exists - os.makedirs(os.path.dirname(self.json_path) or ".", exist_ok=True) - 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