Blame for re3data.py

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