khazna_loader.py

"""
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  -> lookups return None, and the caller says "not checked"
                      rather than implying the article is absent from Khazna.

Add to requirements.txt:
    huggingface_hub
    apscheduler
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
from pathlib import Path
from typing import Optional

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")


class KhaznaIndex:
    def __init__(self) -> None:
        self._idx: dict[str, dict] = {}
        self._loaded_at: Optional[str] = None

    # ── 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
            )
            return Path(p)
        except Exception as e:
            logger.warning("Khazna index download failed: %s", e)
            return LOCAL_PATH if LOCAL_PATH.exists() else None

    def reload(self) -> bool:
        path = self._download()
        if not path or not path.exists():
            logger.warning("Khazna index unavailable — lookups will return None")
            return False
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
        except json.JSONDecodeError as e:
            logger.error("Khazna index malformed, keeping previous copy: %s", e)
            return False
        # Never replace a good index with an empty one.
        if not data and self._idx:
            logger.error("Downloaded index was empty — keeping previous copy")
            return False
        self._idx = data
        from datetime import datetime, timezone
        self._loaded_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
        logger.info("Khazna index loaded: %d DOIs", len(self._idx))
        return True

    # ── lookup ─────────────────────────────────────────────────────────────
    def get(self, doi: str) -> Optional[dict]:
        d = (doi or "").strip().lower()
        for p in ("https://doi.org/", "http://doi.org/", "https://dx.doi.org/", "doi:"):
            if d.startswith(p):
                d = d[len(p):]
                break
        return self._idx.get(d.strip("/"))

    @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}


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")
#     sched.add_job(lambda: asyncio.create_task(asyncio.to_thread(khazna_index.reload)),
#                   CronTrigger(hour=23, minute=30))   # 30 min after the harvest
#     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")}