74 lines
1.8 KiB
Python
74 lines
1.8 KiB
Python
|
"""TTL-aware cache for mDNS resolutions. JSON-backed for portability."""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import time
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
CACHE_PATH = Path(os.environ.get(
|
||
|
|
"MDNS_CACHE_PATH",
|
||
|
|
Path.home() / ".cache" / "mdns" / "hosts.json",
|
||
|
|
))
|
||
|
|
|
||
|
|
|
||
|
|
def _load() -> dict:
|
||
|
|
if not CACHE_PATH.exists():
|
||
|
|
return {}
|
||
|
|
try:
|
||
|
|
return json.loads(CACHE_PATH.read_text())
|
||
|
|
except Exception:
|
||
|
|
return {}
|
||
|
|
|
||
|
|
|
||
|
|
def _save(data: dict) -> None:
|
||
|
|
CACHE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
tmp = CACHE_PATH.with_suffix(".tmp")
|
||
|
|
tmp.write_text(json.dumps(data, indent=2, sort_keys=True))
|
||
|
|
tmp.replace(CACHE_PATH)
|
||
|
|
|
||
|
|
|
||
|
|
def get(host: str, family: str = "v4") -> str | None:
|
||
|
|
"""Return IP if fresh, None if missing or expired."""
|
||
|
|
entry = _load().get(f"{host}|{family}")
|
||
|
|
if not entry:
|
||
|
|
return None
|
||
|
|
if entry.get("expires", 0) < time.time():
|
||
|
|
return None
|
||
|
|
return entry.get("ip")
|
||
|
|
|
||
|
|
|
||
|
|
def get_stale(host: str, family: str = "v4") -> str | None:
|
||
|
|
"""Return IP even if expired (last-resort fallback)."""
|
||
|
|
entry = _load().get(f"{host}|{family}")
|
||
|
|
return entry.get("ip") if entry else None
|
||
|
|
|
||
|
|
|
||
|
|
def put(host: str, ip: str, family: str = "v4", ttl: int = 120) -> None:
|
||
|
|
data = _load()
|
||
|
|
ttl = max(30, min(ttl, 3600))
|
||
|
|
data[f"{host}|{family}"] = {
|
||
|
|
"ip": ip,
|
||
|
|
"ttl": ttl,
|
||
|
|
"expires": time.time() + ttl,
|
||
|
|
"updated": time.time(),
|
||
|
|
}
|
||
|
|
_save(data)
|
||
|
|
|
||
|
|
|
||
|
|
def clear(host: str | None = None) -> int:
|
||
|
|
"""Clear entries. Returns number of entries removed."""
|
||
|
|
if host is None:
|
||
|
|
n = len(_load())
|
||
|
|
if CACHE_PATH.exists():
|
||
|
|
CACHE_PATH.unlink()
|
||
|
|
return n
|
||
|
|
data = _load()
|
||
|
|
removed = [k for k in list(data) if k.split("|")[0] == host]
|
||
|
|
for k in removed:
|
||
|
|
data.pop(k)
|
||
|
|
_save(data)
|
||
|
|
return len(removed)
|
||
|
|
|
||
|
|
|
||
|
|
def all_entries() -> dict:
|
||
|
|
return _load()
|