Spaces:
Running
Running
File size: 1,549 Bytes
1d3bf6d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 | """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)
|