Hugging Face's logo

Spaces:
nikeshn
/
researchbee
Sleeping

App Files Community
researchbee
File size: 5,418 Bytes
7c6808b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""
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")}