Spaces:
Running
Running
| """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. | |
| 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 | |