Spaces:
Running
Running
| """ | |
| 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)) | |
| def available(self) -> bool: | |
| return bool(self._idx) | |
| 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")} | |