Spaces:
Running
Running
File size: 2,246 Bytes
2b8ea89 26bd3fa 2b8ea89 26bd3fa 2b8ea89 26bd3fa 2b8ea89 26bd3fa 2b8ea89 | 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 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """Strict-size rolling local cache for certificate PDFs."""
from __future__ import annotations
import re
from pathlib import Path
class PdfCache:
def __init__(self, root: str | Path, max_bytes: int) -> None:
self.root = Path(root)
self.max_bytes = int(max_bytes)
self.root.mkdir(parents=True, exist_ok=True)
# Parallel parse workers share one cache directory, so any file listed here may be gone by
# the time it is stat()ed or unlinked: a vanished file counts as zero bytes, never an error.
@staticmethod
def _stat(path: Path):
try:
return path.stat()
except FileNotFoundError:
return None
def size_bytes(self) -> int:
return sum(
stat.st_size
for stat in (self._stat(p) for p in self.root.rglob("*") if p.is_file())
if stat is not None
)
def _files_oldest_first(self) -> list[Path]:
stamped = [
(stat.st_mtime, p)
for p in self.root.rglob("*")
if p.is_file() and (stat := self._stat(p)) is not None
]
return [p for _mtime, p in sorted(stamped, key=lambda item: item[0])]
def safe_name(self, name: str) -> str:
cleaned = re.sub(r"[^A-Za-z0-9_.-]", "_", str(name or "certificate.pdf"))
return cleaned if cleaned.lower().endswith(".pdf") else f"{cleaned}.pdf"
def write(self, name: str, data: bytes) -> Path:
if len(data) > self.max_bytes:
raise RuntimeError("PDF cache cap exceeded")
for stale in self._files_oldest_first():
if self.size_bytes() + len(data) <= self.max_bytes:
break
stale.unlink(missing_ok=True)
if self.size_bytes() + len(data) > self.max_bytes:
raise RuntimeError("PDF cache cap exceeded")
path = self.root / self.safe_name(name)
path.write_bytes(data)
return path
def delete(self, path: str | Path) -> None:
p = Path(path)
try:
if p.is_file() and p.resolve().is_relative_to(self.root.resolve()):
p.unlink()
except FileNotFoundError:
return
|