""" khazna_loader.py — Space-side loading of the Khazna index. The harvest runs in GitHub Actions (nightly, 23:00 Dubai) and publishes khazna_index.json to an HF Dataset repo. The Space downloads it — it never harvests, because HF Spaces have an ephemeral filesystem and a 20-minute harvest would block startup. Behaviour: * startup -> download once, load into memory (non-blocking) * daily 23:30 -> re-download (a few seconds) * unavailable -> keep the last valid snapshot, or say "not checked" when none is loaded; a load time is not a source harvest date. Add to requirements.txt: huggingface_hub apscheduler """ from __future__ import annotations import json import logging import os import re import threading from datetime import datetime, timezone from pathlib import Path from typing import Optional from urllib.parse import urlsplit logger = logging.getLogger("khazna") HF_DATASET = os.getenv("KHAZNA_DATASET", "nikeshn/researchbee-data") INDEX_FILE = "khazna_index.json" LOCAL_PATH = Path("/tmp/khazna_index.json") _DOI_RE = re.compile(r"^10\.\d{4,9}/\S+$", re.ASCII) _DEPOSIT_STATES = {"open", "embargoed", "restricted", "closed", "metadata_only", "undetermined"} def _normalise_doi(doi: str) -> str: d = doi.strip().lower() for prefix in ("https://doi.org/", "http://doi.org/", "https://dx.doi.org/", "http://dx.doi.org/", "doi:"): if d.startswith(prefix): d = d[len(prefix):] break return d.strip().strip("/") def _validate_snapshot(data: object) -> dict[str, dict]: """Validate the existing flat source format before replacing a good index.""" if not isinstance(data, dict) or not data: raise ValueError("empty_or_invalid_map") validated: dict[str, dict] = {} for key, value in data.items(): if not isinstance(key, str) or not isinstance(value, dict): raise ValueError("invalid_record") doi = _normalise_doi(key) if not _DOI_RE.fullmatch(doi) or doi in validated: raise ValueError("invalid_or_duplicate_doi") record = dict(value) for field in ("deposit_state", "embargo_end", "portal_url"): if record.get(field) is not None and not isinstance(record[field], str): raise ValueError("invalid_record_field") state = (record.get("deposit_state") or "undetermined").strip().lower() record["deposit_state"] = state if state in _DEPOSIT_STATES else "undetermined" portal = (record.get("portal_url") or "").strip() if portal: parsed = urlsplit(portal) if parsed.scheme not in {"https", "http"} or not parsed.hostname or parsed.username or parsed.password: raise ValueError("invalid_portal_url") record["portal_url"] = portal validated[doi] = record return validated class KhaznaIndex: def __init__(self) -> None: self._idx: dict[str, dict] = {} self._loaded_at: Optional[str] = None self._last_reload_at: Optional[str] = None self._last_reload_success: Optional[bool] = None self._last_reload_status = "not_attempted" self._load_source: Optional[str] = None self._download_origin = "unknown" self._reload_lock = threading.Lock() # ── loading ──────────────────────────────────────────────────────────── def _download(self) -> Optional[Path]: try: from huggingface_hub import hf_hub_download p = hf_hub_download( repo_id=HF_DATASET, filename=INDEX_FILE, repo_type="dataset", token=os.getenv("HF_TOKEN"), # omit if the dataset is public ) self._download_origin = "huggingface_download_or_cache" return Path(p) except Exception: # SDK/network exception messages can contain tokens, signed URLs # or internal paths. Keep logs and public status categorical. logger.warning("Khazna index download unavailable; trying local fallback") self._download_origin = "local_fallback" return LOCAL_PATH if LOCAL_PATH.exists() else None def reload(self) -> bool: with self._reload_lock: self._last_reload_at = datetime.now(timezone.utc).isoformat(timespec="seconds") self._last_reload_success = False self._download_origin = "unknown" try: path = self._download() if not path or not path.exists(): self._last_reload_status = "unavailable" logger.warning("Khazna index unavailable; retaining any previously loaded snapshot") return False data = json.loads(path.read_text(encoding="utf-8")) validated = _validate_snapshot(data) except (ValueError, UnicodeError): self._last_reload_status = "invalid_snapshot" logger.error("Khazna index invalid; retaining any previously loaded snapshot") return False except Exception: self._last_reload_status = "load_error" logger.error("Khazna index could not be loaded; retaining any previously loaded snapshot") return False self._idx = validated self._loaded_at = datetime.now(timezone.utc).isoformat(timespec="seconds") self._last_reload_success = True self._last_reload_status = "loaded" self._load_source = self._download_origin logger.info("Khazna index loaded: %d DOIs; source harvest date unknown", len(self._idx)) return True # ── lookup ───────────────────────────────────────────────────────────── def get(self, doi: str) -> Optional[dict]: if not isinstance(doi, str): return None return self._idx.get(_normalise_doi(doi)) @property def available(self) -> bool: return bool(self._idx) @property def status(self) -> dict: return {"available": self.available, "records": len(self._idx), "loaded_at": self._loaded_at, "source": HF_DATASET, "last_reload_at": self._last_reload_at, "last_reload_success": self._last_reload_success, "last_reload_status": self._last_reload_status, "load_source": self._load_source, "retaining_previous_snapshot": self.available and self._last_reload_success is False, # The existing DOI map contains no harvest timestamp. Neither # loading a cache nor downloading establishes source freshness. "source_harvested_at": None, "snapshot_freshness": "unknown", "freshness_note": "Source harvest date is not supplied. loaded_at records loading only; an absent DOI is not proof of current repository absence."} khazna_index = KhaznaIndex() # ── FastAPI wiring (app.py) ──────────────────────────────────────────────── # # from khazna_loader import khazna_index # from apscheduler.schedulers.asyncio import AsyncIOScheduler # from apscheduler.triggers.cron import CronTrigger # # @app.on_event("startup") # async def _startup(): # # Non-blocking: the Space serves traffic immediately; Khazna lookups # # report "not checked" for the few seconds before the index lands. # asyncio.create_task(asyncio.to_thread(khazna_index.reload)) # # sched = AsyncIOScheduler(timezone="Asia/Dubai") # # AsyncIOScheduler runs this synchronous job in its executor; do not call # # asyncio.create_task from a scheduler worker with no running event loop. # sched.add_job(khazna_index.reload, # CronTrigger(hour=23, minute=30, timezone="Asia/Dubai")) # sched.start() # # @app.get("/api/health/khazna") # async def khazna_health(): # return khazna_index.status # # # ── Using it in the license flow ─────────────────────────────────────────── # # def khazna_status_for(doi: str) -> dict: # """Three-valued, never guesses absence when the index is missing.""" # if not khazna_index.available: # return {"checked": False, # "note": "Khazna could not be checked right now."} # rec = khazna_index.get(doi) # if rec is None: # return {"checked": True, "in_khazna": False} # return {"checked": True, "in_khazna": True, # "deposit_state": rec["deposit_state"], # open|embargoed|...|metadata_only # "embargo_end": rec.get("embargo_end"), # "portal_url": rec.get("portal_url"), # "needs_deposit": rec["deposit_state"] in ("metadata_only", "closed")}