Spaces:
Running
Running
| """ | |
| re3data.py — supplementary repository discovery from the re3data.org registry. | |
| WHY THIS EXISTS | |
| --------------- | |
| The curated KU Library registry (repositories.py) stays the primary source: | |
| the model selects from it by id and every displayed fact comes from a | |
| reviewed record. That registry is intentionally small, so a dataset in a | |
| narrower field may only receive generalist suggestions. | |
| This module adds a clearly separated "more options" list from re3data.org, | |
| the Registry of Research Data Repositories (DataCite). It never changes the | |
| curated recommendations, the deposit strategy, or Khazna's position. | |
| EVIDENCE RULES | |
| -------------- | |
| * Facts shown for these entries come only from the re3data record, labelled | |
| as such, with the record's own last-update date. They are not reviewed by | |
| KU Library and are never merged into curated entries. | |
| * Repositories already in the curated registry, repositories with an end | |
| date (closed), repositories whose data upload is closed, and purely | |
| institutional repositories (usually restricted to their own staff) are | |
| excluded. | |
| * Only the researcher-entered *discipline* is sent to re3data, never the | |
| dataset title, description or notes, which may be unpublished. | |
| FAILURE MODEL | |
| ------------- | |
| Every network, parsing or validation failure returns an "unavailable" | |
| status and an empty list. Nothing here may raise into the route. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import logging | |
| import os | |
| import re | |
| import time | |
| from collections import OrderedDict | |
| from datetime import datetime, timezone | |
| from typing import Any, Dict, Iterable, List, Optional, Tuple | |
| from urllib.parse import urlparse | |
| import xml.etree.ElementTree as ET | |
| import httpx | |
| logger = logging.getLogger(__name__) | |
| API_HOST = "www.re3data.org" | |
| SEARCH_URL = f"https://{API_HOST}/api/beta/repositories" | |
| DETAIL_URL = f"https://{API_HOST}/api/v1/repository/{{rid}}" | |
| RECORD_URL = f"https://{API_HOST}/repository/{{rid}}" | |
| UA = "ResearchBee/1.8.3 (mailto:library@ku.ac.ae)" | |
| ID_RE = re.compile(r"^r3d\d{9}$") | |
| MAX_RESPONSE_BYTES = 2_000_000 | |
| MAX_DETAIL_FETCHES = 20 | |
| DETAIL_CONCURRENCY = 6 | |
| SEARCH_TTL = 24 * 3600 | |
| FAILURE_TTL = 10 * 60 | |
| DETAIL_TTL = 7 * 24 * 3600 | |
| CACHE_MAX = 256 | |
| DETAIL_CACHE_MAX = 1024 | |
| _STOPWORDS = { | |
| "and", "of", "the", "in", "for", "or", "a", "an", "to", "with", "on", "data", | |
| "dataset", "datasets", "research", "science", "sciences", "studies", "study", | |
| "general", "applied", "other", "misc", "miscellaneous", | |
| } | |
| _http_client: Optional[httpx.AsyncClient] = None | |
| _search_cache: "OrderedDict[str, Tuple[float, Any]]" = OrderedDict() | |
| _detail_cache: "OrderedDict[str, Tuple[float, Optional[dict]]]" = OrderedDict() | |
| _inflight: Dict[str, asyncio.Future] = {} | |
| _last_success: Optional[str] = None | |
| _last_error: Optional[dict] = None | |
| def enabled() -> bool: | |
| return (os.getenv("RE3DATA_ENABLED", "true").strip().lower() not in {"0", "false", "no", "off"}) | |
| def _now_iso() -> str: | |
| return datetime.now(timezone.utc).isoformat(timespec="seconds") | |
| def status() -> dict: | |
| return { | |
| "enabled": enabled(), | |
| "last_success": _last_success, | |
| "last_error": _last_error, | |
| "availability": ( | |
| "disabled" if not enabled() | |
| else "not_checked" if _last_success is None and _last_error is None | |
| else "degraded" if _last_error else "available" | |
| ), | |
| } | |
| # ── HTTP client lifecycle ───────────────────────────────────────────────── | |
| async def _enforce_origin(request: httpx.Request) -> None: | |
| url = request.url | |
| if url.scheme != "https" or url.host != API_HOST or url.port not in (None, 443): | |
| raise ValueError("re3data client refuses non-provider origins") | |
| def init_client() -> None: | |
| global _http_client | |
| if _http_client is None: | |
| _http_client = httpx.AsyncClient( | |
| timeout=httpx.Timeout(6.0, connect=4.0), | |
| headers={"User-Agent": UA, "Accept": "application/xml, text/xml"}, | |
| follow_redirects=False, | |
| event_hooks={"request": [_enforce_origin]}, | |
| ) | |
| async def close_client() -> None: | |
| global _http_client | |
| if _http_client is not None: | |
| await _http_client.aclose() | |
| _http_client = None | |
| def _client() -> httpx.AsyncClient: | |
| if _http_client is None: | |
| init_client() | |
| return _http_client # type: ignore[return-value] | |
| # ── Bounded caches ──────────────────────────────────────────────────────── | |
| def _cache_get(cache: OrderedDict, key: str): | |
| item = cache.get(key) | |
| if not item: | |
| return None | |
| expires, value = item | |
| if expires < time.monotonic(): | |
| cache.pop(key, None) | |
| return None | |
| cache.move_to_end(key) | |
| return item | |
| def _cache_put(cache: OrderedDict, key: str, value: Any, ttl: float, limit: int) -> None: | |
| cache[key] = (time.monotonic() + ttl, value) | |
| cache.move_to_end(key) | |
| while len(cache) > limit: | |
| cache.popitem(last=False) | |
| def clear_caches() -> None: | |
| _search_cache.clear() | |
| _detail_cache.clear() | |
| _inflight.clear() | |
| # ── XML helpers (namespace-agnostic; the schema namespace has changed) ──── | |
| def _local(tag: Any) -> str: | |
| return tag.rsplit("}", 1)[-1] if isinstance(tag, str) else "" | |
| def _safe_parse(content: bytes) -> Optional[ET.Element]: | |
| if not content or len(content) > MAX_RESPONSE_BYTES: | |
| return None | |
| head = content[:4096].upper() | |
| # re3data never needs a DTD; refusing one blocks entity-expansion tricks. | |
| if b"<!DOCTYPE" in head or b"<!ENTITY" in content.upper(): | |
| return None | |
| try: | |
| return ET.fromstring(content) | |
| except ET.ParseError: | |
| return None | |
| def _children(node: ET.Element, name: str) -> List[ET.Element]: | |
| return [child for child in list(node) if _local(child.tag) == name] | |
| def _text(node: Optional[ET.Element]) -> str: | |
| if node is None: | |
| return "" | |
| return " ".join("".join(node.itertext()).split()) | |
| def _first_text(node: ET.Element, name: str) -> str: | |
| for child in _children(node, name): | |
| value = _text(child) | |
| if value: | |
| return value | |
| return "" | |
| def _all_text(node: ET.Element, name: str, limit: int = 20, width: int = 200) -> List[str]: | |
| out: List[str] = [] | |
| for child in node.iter(): | |
| if _local(child.tag) == name: | |
| value = _text(child)[:width] | |
| if value and value not in out: | |
| out.append(value) | |
| if len(out) >= limit: | |
| break | |
| return out | |
| def _https_url(value: str) -> str: | |
| value = (value or "").strip() | |
| try: | |
| parsed = urlparse(value) | |
| except ValueError: | |
| return "" | |
| if parsed.scheme != "https" or not parsed.hostname or "." not in parsed.hostname: | |
| return "" | |
| if parsed.username or parsed.password or (parsed.port not in (None, 443)): | |
| return "" | |
| return value[:500] | |
| def _host(value: str) -> str: | |
| try: | |
| host = (urlparse(value).hostname or "").lower() | |
| except ValueError: | |
| return "" | |
| return host[4:] if host.startswith("www.") else host | |
| def parse_search(content: bytes) -> List[dict]: | |
| """Parse a search/list response into [{"id", "name", "doi"}].""" | |
| root = _safe_parse(content) | |
| if root is None: | |
| return [] | |
| out: List[dict] = [] | |
| seen = set() | |
| for node in root.iter(): | |
| if _local(node.tag) != "repository": | |
| continue | |
| rid = _first_text(node, "id") | |
| if not ID_RE.match(rid) or rid in seen: | |
| continue | |
| seen.add(rid) | |
| out.append({ | |
| "id": rid, | |
| "name": _first_text(node, "name")[:300], | |
| "doi": _first_text(node, "doi")[:200], | |
| }) | |
| return out | |
| def _yes_no_unknown(value: str) -> str: | |
| value = (value or "").strip().lower() | |
| if value == "yes": | |
| return "Yes" | |
| if value == "no": | |
| return "No" | |
| return "Unknown" | |
| _SUBJECT_CODE = re.compile(r"^\s*\d+\s+") | |
| def parse_detail(content: bytes) -> Optional[dict]: | |
| """Parse one re3data repository record. Returns None if unusable.""" | |
| root = _safe_parse(content) | |
| if root is None: | |
| return None | |
| repo = root if _local(root.tag) == "repository" else next( | |
| (node for node in root.iter() if _local(node.tag) == "repository"), None | |
| ) | |
| if repo is None: | |
| return None | |
| rid = _first_text(repo, "re3data.orgIdentifier") | |
| name = _first_text(repo, "repositoryName") | |
| if not ID_RE.match(rid) or not name: | |
| return None | |
| subjects: List[str] = [] | |
| for value in _all_text(repo, "subject", limit=30): | |
| clean = _SUBJECT_CODE.sub("", value).strip() | |
| if clean and clean not in subjects: | |
| subjects.append(clean) | |
| pid_systems = [p for p in _all_text(repo, "pidSystem", limit=8, width=40) if p.lower() != "none"] | |
| upload_types = [v.lower() for v in _all_text(repo, "dataUploadType", limit=5, width=30)] | |
| access_types = [v.lower() for v in _all_text(repo, "dataAccessType", limit=5, width=30)] | |
| description = _first_text(repo, "description") | |
| if len(description) > 600: | |
| description = description[:597].rsplit(" ", 1)[0] + "…" | |
| return { | |
| "id": rid, | |
| "name": name[:300], | |
| "url": _https_url(_first_text(repo, "repositoryURL")), | |
| "description": description, | |
| "types": [t.lower() for t in _all_text(repo, "type", limit=5, width=40)], | |
| "subjects": subjects[:10], | |
| "keywords": _all_text(repo, "keyword", limit=15, width=80), | |
| "content_types": [_SUBJECT_CODE.sub("", c) for c in _all_text(repo, "contentType", limit=10, width=120)], | |
| "pid_systems": pid_systems, | |
| "versioning": _yes_no_unknown(_first_text(repo, "versioning")), | |
| "data_upload": upload_types, | |
| "data_access": access_types, | |
| "certificates": _all_text(repo, "certificate", limit=5, width=80), | |
| "countries": _all_text(repo, "institutionCountry", limit=5, width=10), | |
| "end_date": _first_text(repo, "endDate")[:30], | |
| "last_update": _first_text(repo, "lastUpdate")[:30], | |
| } | |
| # ── Network operations ──────────────────────────────────────────────────── | |
| class Re3dataUnavailable(RuntimeError): | |
| pass | |
| async def _get(url: str, params: Optional[dict] = None) -> bytes: | |
| global _last_success, _last_error | |
| try: | |
| response = await _client().get(url, params=params) | |
| except (httpx.HTTPError, ValueError) as exc: | |
| _last_error = {"type": type(exc).__name__, "at": _now_iso()} | |
| raise Re3dataUnavailable() from None | |
| if response.status_code != 200: | |
| _last_error = {"status": response.status_code, "at": _now_iso()} | |
| raise Re3dataUnavailable() | |
| content = response.content | |
| if len(content) > MAX_RESPONSE_BYTES: | |
| _last_error = {"type": "ResponseTooLarge", "at": _now_iso()} | |
| raise Re3dataUnavailable() | |
| _last_success = _now_iso() | |
| _last_error = None | |
| return content | |
| async def search(query: str) -> List[dict]: | |
| """Search re3data; cached. Raises Re3dataUnavailable on failure.""" | |
| key = " ".join(query.lower().split())[:120] | |
| cached = _cache_get(_search_cache, key) | |
| if cached is not None: | |
| value = cached[1] | |
| if isinstance(value, Exception): | |
| raise Re3dataUnavailable() | |
| return value | |
| pending = _inflight.get("s:" + key) | |
| if pending is not None: | |
| return await asyncio.shield(pending) | |
| loop = asyncio.get_running_loop() | |
| future: asyncio.Future = loop.create_future() | |
| _inflight["s:" + key] = future | |
| try: | |
| content = await _get(SEARCH_URL, params={"query": key}) | |
| hits = parse_search(content) | |
| _cache_put(_search_cache, key, hits, SEARCH_TTL, CACHE_MAX) | |
| future.set_result(hits) | |
| return hits | |
| except Exception as exc: | |
| _cache_put(_search_cache, key, Re3dataUnavailable(), FAILURE_TTL, CACHE_MAX) | |
| future.set_exception(Re3dataUnavailable()) | |
| future.exception() # mark retrieved; concurrent waiters still see it | |
| raise Re3dataUnavailable() from None | |
| finally: | |
| _inflight.pop("s:" + key, None) | |
| async def detail(rid: str) -> Optional[dict]: | |
| """Fetch one record; cached. Returns None when unavailable.""" | |
| if not ID_RE.match(rid or ""): | |
| return None | |
| cached = _cache_get(_detail_cache, rid) | |
| if cached is not None: | |
| return cached[1] | |
| try: | |
| content = await _get(DETAIL_URL.format(rid=rid)) | |
| except Re3dataUnavailable: | |
| _cache_put(_detail_cache, rid, None, FAILURE_TTL, DETAIL_CACHE_MAX) | |
| return None | |
| record = parse_detail(content) | |
| _cache_put(_detail_cache, rid, record, DETAIL_TTL if record else FAILURE_TTL, DETAIL_CACHE_MAX) | |
| return record | |
| # ── Selection ───────────────────────────────────────────────────────────── | |
| def _tokens(text: str) -> List[str]: | |
| out: List[str] = [] | |
| for token in re.findall(r"[a-z0-9]+", (text or "").lower()): | |
| if len(token) > 2 and token not in _STOPWORDS and token not in out: | |
| out.append(token) | |
| return out | |
| def _stem(token: str) -> str: | |
| for suffix in ("ies", "es", "s"): | |
| if token.endswith(suffix) and len(token) - len(suffix) >= 4: | |
| return token[: -len(suffix)] + ("y" if suffix == "ies" else "") | |
| return token | |
| def _matches(tokens: Iterable[str], text: str) -> List[str]: | |
| words = {_stem(w) for w in re.findall(r"[a-z0-9]+", (text or "").lower())} | |
| return [t for t in tokens if _stem(t) in words] | |
| def build_query(discipline: str) -> str: | |
| """Only the user's discipline field is sent; bounded and normalised.""" | |
| text = " ".join(str(discipline or "").split()) | |
| text = re.sub(r"[^\w\s\-&,/()]", " ", text, flags=re.UNICODE) | |
| return " ".join(text.split())[:100] | |
| def _is_sensitive(constraints: dict) -> bool: | |
| sensitivity = str(constraints.get("sensitivity") or "").strip().lower() | |
| personal = str(constraints.get("contains_personal_data") or "").strip().lower() | |
| return (bool(sensitivity) and sensitivity not in {"open", "none", "non-sensitive", "not sensitive"}) or ( | |
| bool(personal) and personal != "no" | |
| ) | |
| def _excluded_by_registry(record: dict, curated: Iterable[dict]) -> bool: | |
| name = " ".join(record.get("name", "").lower().split()) | |
| host = _host(record.get("url", "")) | |
| for entry in curated: | |
| entry_name = " ".join(str(entry.get("name", "")).lower().split()) | |
| if name and (name == entry_name or (len(entry_name) > 4 and entry_name in name)): | |
| return True | |
| entry_host = _host(str(entry.get("url", ""))) | |
| if host and entry_host and (host == entry_host or host.endswith("." + entry_host)): | |
| return True | |
| return False | |
| def score_record(record: dict, tokens: List[str]) -> Tuple[int, List[str]]: | |
| """Relevance from subject/keyword/name evidence plus deposit-readiness.""" | |
| subject_text = " ".join(record.get("subjects", []) + record.get("keywords", []) + [record.get("name", "")]) | |
| strong = _matches(tokens, subject_text) | |
| weak = [t for t in _matches(tokens, record.get("description", "")) if t not in strong] | |
| if not strong: | |
| return 0, [] | |
| score = 10 * len(strong) + 3 * len(weak) | |
| pids = {p.lower() for p in record.get("pid_systems", [])} | |
| if "doi" in pids: | |
| score += 4 | |
| if record.get("certificates"): | |
| score += 3 | |
| if "open" in record.get("data_upload", []): | |
| score += 2 | |
| if record.get("versioning") == "Yes": | |
| score += 1 | |
| return score, strong + weak | |
| def _public_entry(record: dict, matched: List[str], sensitive: bool) -> dict: | |
| pids = record.get("pid_systems", []) | |
| upload = record.get("data_upload", []) | |
| access = record.get("data_access", []) | |
| notes = [ | |
| "Listed in re3data.org; not reviewed by KU Library. Confirm deposit eligibility, costs, " | |
| "licences and file limits on the repository's own site before uploading." | |
| ] | |
| if upload and "open" not in upload: | |
| notes.append("re3data lists data upload as restricted; depositor eligibility must be confirmed.") | |
| if sensitive: | |
| notes.append( | |
| "Your dataset is marked sensitive or personal: KU Library and the responsible ethics/data-protection " | |
| "team must approve any repository and access conditions before upload." | |
| ) | |
| last_update = record.get("last_update", "") | |
| return { | |
| "id": record["id"], | |
| "name": record["name"], | |
| "url": record.get("url", ""), | |
| "re3data_url": RECORD_URL.format(rid=record["id"]), | |
| "description": record.get("description", ""), | |
| "repository_types": record.get("types", []), | |
| "subjects": record.get("subjects", [])[:6], | |
| "content_types": record.get("content_types", [])[:6], | |
| "persistent_identifier": ", ".join(pids) if pids else "Not listed", | |
| "doi_assignment": "Yes" if any(p.lower() == "doi" for p in pids) else "Not listed", | |
| "versioning": record.get("versioning", "Unknown"), | |
| "data_upload": ", ".join(u.capitalize() for u in upload) if upload else "Not listed", | |
| "data_access": ", ".join(a.capitalize() for a in access) if access else "Not listed", | |
| "certificates": record.get("certificates", []), | |
| "countries": record.get("countries", []), | |
| "last_update": last_update, | |
| "matched_terms": matched[:6], | |
| "source": "re3data", | |
| "source_note": ( | |
| "Details as recorded by re3data.org" | |
| + (f" (record last updated {last_update[:10]})" if last_update else "") | |
| + "." | |
| ), | |
| "review_notes": notes, | |
| "file_deposit_recommended": False, | |
| } | |
| async def find_supplementary( | |
| discipline: str, | |
| constraints: Optional[dict] = None, | |
| curated: Iterable[dict] = (), | |
| limit: int = 5, | |
| ) -> dict: | |
| """Return {"status", "query", "repositories", "message", "retrieved_at"}. | |
| Never raises. ``curated`` is the KU registry, used only for de-duplication. | |
| """ | |
| constraints = constraints if isinstance(constraints, dict) else {} | |
| query = build_query(discipline) | |
| base = {"status": "skipped", "query": query, "repositories": [], "message": "", | |
| "source": "re3data.org", "source_url": "https://www.re3data.org", | |
| "retrieved_at": _now_iso()} | |
| if not enabled(): | |
| return {**base, "status": "disabled", | |
| "message": "re3data search is switched off for this service."} | |
| tokens = _tokens(query) | |
| if not tokens: | |
| return {**base, "message": "Enter a discipline to search re3data for additional repositories."} | |
| curated = list(curated) | |
| try: | |
| hits = await search(query) | |
| if len(hits) < 3 and len(tokens) > 1: | |
| # A multi-word discipline can be too narrow for the registry's | |
| # search; retry once with the most distinctive (longest) word. | |
| extra = await search(max(tokens, key=len)) | |
| seen = {h["id"] for h in hits} | |
| hits += [h for h in extra if h["id"] not in seen] | |
| except Re3dataUnavailable: | |
| return {**base, "status": "unavailable", | |
| "message": "re3data.org could not be reached. The curated KU Library recommendations are unaffected."} | |
| except Exception as exc: # defensive: never break the repository finder | |
| logger.warning("[re3data] search failed (%s)", type(exc).__name__) | |
| return {**base, "status": "unavailable", | |
| "message": "re3data.org could not be searched. The curated KU Library recommendations are unaffected."} | |
| if not hits: | |
| return {**base, "status": "no_matches", | |
| "message": f'No additional re3data repositories were found for "{query}".'} | |
| # Fetch the hits most likely to be relevant first: name matches lead. | |
| ordered = sorted(enumerate(hits), key=lambda item: (-len(_matches(tokens, item[1].get("name", ""))), item[0])) | |
| candidates = [hit for _i, hit in ordered[:MAX_DETAIL_FETCHES]] | |
| semaphore = asyncio.Semaphore(DETAIL_CONCURRENCY) | |
| async def fetch(rid: str) -> Optional[dict]: | |
| async with semaphore: | |
| try: | |
| return await detail(rid) | |
| except Exception as exc: # defensive | |
| logger.warning("[re3data] detail failed (%s)", type(exc).__name__) | |
| return None | |
| records = await asyncio.gather(*(fetch(hit["id"]) for hit in candidates)) | |
| sensitive = _is_sensitive(constraints) | |
| scored: List[Tuple[int, str, dict, List[str]]] = [] | |
| for record in records: | |
| if not record: | |
| continue | |
| if record.get("end_date"): | |
| continue # closed repository | |
| upload = record.get("data_upload", []) | |
| if upload and all(u == "closed" for u in upload): | |
| continue # does not accept deposits | |
| types = set(record.get("types", [])) | |
| if types and types <= {"institutional"}: | |
| continue # usually limited to that institution's own members | |
| if _excluded_by_registry(record, curated): | |
| continue | |
| if sensitive: | |
| access = set(record.get("data_access", [])) | |
| if not access & {"restricted", "embargoed", "closed"}: | |
| continue # no controlled-access route recorded | |
| score, matched = score_record(record, tokens) | |
| if score <= 0: | |
| continue | |
| scored.append((score, record["name"].lower(), record, matched)) | |
| scored.sort(key=lambda item: (-item[0], item[1])) | |
| picked = [_public_entry(record, matched, sensitive) for _s, _n, record, matched in scored[: max(0, limit)]] | |
| if not picked: | |
| return {**base, "status": "no_matches", | |
| "message": f're3data returned results for "{query}", but none passed the subject and deposit checks.'} | |
| return {**base, "status": "ok", "repositories": picked, | |
| "message": ( | |
| f'{len(picked)} additional repositor{"y" if len(picked) == 1 else "ies"} from re3data.org matched "{query}". ' | |
| "These are registry listings, not KU Library recommendations." | |
| )} | |