| import json |
| import os |
| import tempfile |
|
|
|
|
| class StateStore: |
| def __init__(self, file_path: str): |
| self.file_path = os.path.abspath(file_path) |
| self._data = None |
|
|
| def load(self) -> dict: |
| if self._data is not None: |
| return self._data |
| if not os.path.exists(self.file_path): |
| self._data = {} |
| return self._data |
| with open(self.file_path, "r", encoding="utf-8") as f: |
| self._data = json.load(f) |
| if not isinstance(self._data, dict): |
| self._data = {} |
| return self._data |
|
|
| def save(self): |
| data = self.load() |
| os.makedirs(os.path.dirname(self.file_path), exist_ok=True) |
| with tempfile.NamedTemporaryFile("w", delete=False, dir=os.path.dirname(self.file_path), encoding="utf-8") as f: |
| json.dump(data, f, ensure_ascii=False, separators=(",", ":")) |
| tmp = f.name |
| os.replace(tmp, self.file_path) |
|
|
| def get(self, key: str, default=None): |
| data = self.load() |
| return data.get(key, default) |
|
|
| def set(self, key: str, value): |
| data = self.load() |
| data[key] = value |
| self.save() |
|
|
| def get_set(self, key: str) -> set: |
| data = self.load() |
| val = data.get(key, []) |
| if isinstance(val, list): |
| return set(val) |
| if isinstance(val, set): |
| return val |
| return set() |
|
|
| def add_to_set(self, key: str, item): |
| s = self.get_set(key) |
| s.add(item) |
| self.set(key, sorted(list(s))) |
|
|
