Spaces:
Runtime error
Runtime error
| """Tiny JSON-file registry of uploaded documents (status, progress, page counts).""" | |
| import json | |
| import os | |
| import threading | |
| import time | |
| from functools import lru_cache | |
| from pathlib import Path | |
| from .config import settings | |
| class Registry: | |
| def __init__(self, path: Path): | |
| self.path = path | |
| self.path.parent.mkdir(parents=True, exist_ok=True) | |
| self.lock = threading.RLock() | |
| self.docs: dict[str, dict] = {} | |
| if self.path.exists(): | |
| try: | |
| for d in json.loads(self.path.read_text(encoding="utf-8")): | |
| self.docs[d["doc_id"]] = d | |
| except Exception: | |
| self.docs = {} | |
| def _save(self) -> None: | |
| data = json.dumps(list(self.docs.values()), indent=2) | |
| tmp = self.path.with_suffix(".tmp") | |
| try: | |
| tmp.write_text(data, encoding="utf-8") | |
| os.replace(tmp, self.path) | |
| except OSError: # e.g. file briefly locked by OneDrive/antivirus on Windows | |
| self.path.write_text(data, encoding="utf-8") | |
| def all(self) -> list[dict]: | |
| with self.lock: | |
| return sorted((dict(d) for d in self.docs.values()), key=lambda d: d["uploaded_at"]) | |
| def ready(self) -> list[dict]: | |
| return [d for d in self.all() if d["status"] == "ready"] | |
| def get(self, doc_id: str) -> dict | None: | |
| with self.lock: | |
| d = self.docs.get(doc_id) | |
| return dict(d) if d else None | |
| def find_by_hash(self, sha: str) -> dict | None: | |
| with self.lock: | |
| for d in self.docs.values(): | |
| if d.get("sha256") == sha and d["status"] in ("ready", "processing"): | |
| return dict(d) | |
| return None | |
| def create(self, entry: dict) -> dict: | |
| with self.lock: | |
| entry.setdefault("uploaded_at", time.time()) | |
| self.docs[entry["doc_id"]] = entry | |
| self._save() | |
| return dict(entry) | |
| def update(self, doc_id: str, **fields) -> None: | |
| with self.lock: | |
| if doc_id in self.docs: | |
| self.docs[doc_id].update(fields) | |
| self._save() | |
| def delete(self, doc_id: str) -> None: | |
| with self.lock: | |
| self.docs.pop(doc_id, None) | |
| self._save() | |
| def clear(self) -> None: | |
| with self.lock: | |
| self.docs.clear() | |
| self._save() | |
| def get_registry() -> Registry: | |
| return Registry(Path(settings.data_dir) / "registry.json") | |
| def public(entry: dict) -> dict: | |
| """What the frontend is allowed to see.""" | |
| return { | |
| "doc_id": entry["doc_id"], | |
| "filename": entry["filename"], | |
| "status": entry["status"], | |
| "stage": entry.get("stage", ""), | |
| "progress": entry.get("progress", 0), | |
| "pages": entry.get("pages", 0), | |
| "chunks": entry.get("chunks", 0), | |
| "empty_pages": len(entry.get("empty_pages", [])), | |
| "warning": entry.get("warning"), | |
| "error": entry.get("error"), | |
| } | |