Spaces:
Running
Running
| """Small process-local cache for static character and prompt files only.""" | |
| from collections import OrderedDict | |
| from copy import deepcopy | |
| import json | |
| from pathlib import Path | |
| from threading import RLock | |
| _MAX_ENTRIES = 64 | |
| _CACHE = OrderedDict() | |
| _LOCK = RLock() | |
| def _signature(path): | |
| stat = path.stat() | |
| return (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns) | |
| def _load(path, as_json): | |
| path = Path(path).resolve() | |
| key = (path, as_json) | |
| with _LOCK: | |
| # Always stat: changed or deleted files must never silently use stale data. | |
| signature = _signature(path) | |
| cached = _CACHE.get(key) | |
| if cached is not None and cached[0] == signature: | |
| _CACHE.move_to_end(key) | |
| value = cached[1] | |
| else: | |
| _CACHE.pop(key, None) | |
| text = path.read_text(encoding='utf-8') | |
| value = json.loads(text) if as_json else text | |
| # A concurrent file edit must not be cached under its older signature. | |
| if _signature(path) == signature: | |
| _CACHE[key] = (signature, value) | |
| while len(_CACHE) > _MAX_ENTRIES: | |
| _CACHE.popitem(last=False) | |
| return deepcopy(value) if as_json else value | |
| def load_static_json(path): | |
| """Return a private copy of parsed JSON, refreshing when file metadata changes.""" | |
| return _load(path, as_json=True) | |
| def load_static_text(path): | |
| """Return cached UTF-8 text without altering whitespace.""" | |
| return _load(path, as_json=False) | |