app/services/paper_metadata.py

"""Canonical scholarly metadata with provenance and an atomic local cache."""
from __future__ import annotations

import difflib
import hashlib
import html
import json
import os
import re
import tempfile
import threading
import time
from collections.abc import Mapping
from typing import Any
from urllib.parse import quote, quote_plus, urlparse

from pydantic import BaseModel, ConfigDict, Field

_DOI_RE = re.compile(r"(?:https?://(?:dx\.)?doi\.org/|doi:\s*)?(10\.\d{4,9}/[-._;()/:a-z0-9]+)", re.I)
_ARXIV_ID_PATTERN = r"(?:\d{4}\.\d{4,5}|[a-z-]+(?:\.[a-z-]+)?/\d{7})(?:v\d+)?"
_ARXIV_RE = re.compile(
    rf"(?i)(?:https?://(?:www\.)?arxiv\.org/(?:abs|pdf)/|10\.48550/arxiv\.|arxiv[:./]\s*)({_ARXIV_ID_PATTERN})"
)
_RAW_ARXIV_RE = re.compile(rf"(?i)^({_ARXIV_ID_PATTERN})$")
_DEFAULT_CACHE_ROOT = os.path.expanduser("~/.studybuddy/paper_metadata")
_CACHE_VERSION = 2


class PaperAuthor(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    name: str
    openalex_id: str = ""
    semantic_scholar_id: str = ""
    orcid: str = ""
    profile_url: str = ""


class PaperDestination(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    label: str
    url: str
    provider: str


class PaperIdentifiers(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    doi: str = ""
    arxiv_id: str = ""
    semantic_scholar_id: str = ""
    openalex_id: str = ""
    pmid: str = ""
    openreview_id: str = ""


class PaperMetadata(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    identifiers: PaperIdentifiers = Field(default_factory=PaperIdentifiers)
    candidate_id: str = ""
    title: str = ""
    authors: str = ""
    author_details: list[PaperAuthor] = Field(default_factory=list)
    year: int | None = None
    venue: str = ""
    abstract: str = ""
    canonical_url: str = ""
    cited_by: int = 0
    reference_count: int = 0
    is_open_access: bool = False
    pdf_url: str = ""
    provider: str = "uploaded_front_matter"
    pdf_provider: str = ""
    field_providers: dict[str, str] = Field(default_factory=dict)
    resolution_confidence: float = 0.0
    looked_up_at: float = Field(default_factory=time.time)
    error: str = ""
    destinations: list[PaperDestination] = Field(default_factory=list)
    metadata_quality: str = "sparse"
    abstract_status: str = ""

    def candidate_dict(self) -> dict[str, Any]:
        candidate_id = self.candidate_id or PaperMetadataService.stable_candidate_id(self)
        return {
            "candidate_id": candidate_id,
            "title": self.title, "authors": self.authors, "year": self.year,
            "author_details": [author.model_dump() for author in self.author_details],
            "venue": self.venue, "abstract": self.abstract,
            "url": self.canonical_url, "cited_by": self.cited_by,
            "doi": self.identifiers.doi, "arxiv_id": self.identifiers.arxiv_id,
            "semantic_scholar_id": self.identifiers.semantic_scholar_id,
            "openalex_id": self.identifiers.openalex_id,
            "pmid": self.identifiers.pmid, "openreview_id": self.identifiers.openreview_id,
            "reference_count": self.reference_count,
            "is_open_access": self.is_open_access, "pdf_url": self.pdf_url,
            "provider": self.provider, "pdf_provider": self.pdf_provider,
            "field_providers": self.field_providers,
            "resolution_confidence": self.resolution_confidence,
            "looked_up_at": self.looked_up_at, "error": self.error,
            "destinations": [destination.model_dump() for destination in self.destinations],
            "metadata_quality": self.metadata_quality, "abstract_status": self.abstract_status,
        }


class PaperMetadataService:
    _cache_lock = threading.Lock()

    def __init__(self, cache_root: str | None = None) -> None:
        self.cache_root = cache_root or _DEFAULT_CACHE_ROOT
        self.success_ttl = int(os.getenv("PAPER_METADATA_CACHE_TTL", str(7 * 86400)))
        self.negative_ttl = int(os.getenv("PAPER_METADATA_NEGATIVE_TTL", str(6 * 3600)))
        self.error_ttl = int(os.getenv("PAPER_METADATA_ERROR_TTL", str(15 * 60)))
        os.makedirs(self.cache_root, exist_ok=True)

    def remember_file_metadata(self, file_id: str, metadata: PaperMetadata) -> None:
        """Persist the verified identity used to acquire an owned PDF.

        Evidence parsing intentionally does not rely on vendor metadata.  The
        graph, however, must retain the canonical metadata that was verified
        immediately before this exact file was downloaded, rather than trying
        to rediscover identity from PDF layout text on every refresh.
        """
        if not re.fullmatch(r"[0-9a-f]{64}", file_id or ""):
            return
        directory = os.path.join(self.cache_root, "files")
        os.makedirs(directory, exist_ok=True)
        payload = {"cache_version": _CACHE_VERSION, "file_id": file_id, "metadata": metadata.candidate_dict()}
        with self._cache_lock:
            descriptor, temporary = tempfile.mkstemp(prefix=f".{file_id}.", suffix=".tmp", dir=directory, text=True)
            try:
                with os.fdopen(descriptor, "w", encoding="utf-8") as file:
                    json.dump(payload, file, indent=2)
                    file.flush()
                    os.fsync(file.fileno())
                os.replace(temporary, os.path.join(directory, f"{file_id}.json"))
            finally:
                if os.path.exists(temporary):
                    os.remove(temporary)

    def file_metadata(self, file_id: str) -> dict[str, Any] | None:
        """Return acquisition-time canonical metadata for one PDF content hash."""
        if not re.fullmatch(r"[0-9a-f]{64}", file_id or ""):
            return None
        path = os.path.join(self.cache_root, "files", f"{file_id}.json")
        try:
            with open(path, encoding="utf-8") as file:
                row = json.load(file)
            if int(row.get("cache_version", 0)) != _CACHE_VERSION or row.get("file_id") != file_id:
                return None
            candidate = row.get("metadata")
            return dict(candidate) if isinstance(candidate, dict) else None
        except Exception:
            return None

    def forget_file_metadata(self, file_id: str) -> None:
        if not re.fullmatch(r"[0-9a-f]{64}", file_id or ""):
            return
        try:
            os.remove(os.path.join(self.cache_root, "files", f"{file_id}.json"))
        except FileNotFoundError:
            pass

    def is_fresh(self, looked_up_at: Any, error: str = "") -> bool:
        try:
            age = time.time() - float(looked_up_at or 0)
        except (TypeError, ValueError):
            return False
        if error == "No confident provider match found.":
            ttl = self.negative_ttl
        elif error:
            ttl = self.error_ttl
        else:
            ttl = self.success_ttl
        return 0 <= age < ttl

    async def resolve(self, candidate: Mapping[str, Any]) -> PaperMetadata:
        base = self.from_candidate(candidate)
        cache_key = self._cache_key(base)
        cached = self._load_cache(cache_key)
        if cached is not None:
            return cached

        errors: list[str] = []
        resolved_any = False
        # Provider failures are isolated so one outage does not suppress another.
        if base.identifiers.arxiv_id:
            try:
                resolved = await self._from_arxiv(base.identifiers.arxiv_id)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"arXiv: {exc}")
        if base.identifiers.doi:
            try:
                resolved = await self._from_openalex_doi(base.identifiers.doi)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"OpenAlex DOI: {exc}")
        if base.identifiers.openalex_id:
            try:
                resolved = await self._from_openalex_id(base.identifiers.openalex_id)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"OpenAlex ID: {exc}")
        if base.identifiers.semantic_scholar_id:
            try:
                resolved = await self._from_semantic_scholar(base.identifiers.semantic_scholar_id)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"Semantic Scholar: {exc}")
        elif base.identifiers.doi or base.identifiers.arxiv_id:
            stable_id = f"DOI:{base.identifiers.doi}" if base.identifiers.doi else f"ARXIV:{base.identifiers.arxiv_id}"
            try:
                resolved = await self._from_semantic_scholar(stable_id)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"Semantic Scholar stable ID: {exc}")
        if base.identifiers.pmid:
            try:
                resolved = await self._from_pubmed(base.identifiers.pmid)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"PubMed: {exc}")
        if base.identifiers.doi and (not base.abstract or not base.canonical_url):
            try:
                resolved = await self._from_crossref(base.identifiers.doi)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"Crossref: {exc}")
        if not base.identifiers.openalex_id and base.title:
            try:
                resolved = await self._from_openalex_title(base)
                if resolved:
                    base = self._merge(base, resolved)
                    resolved_any = True
            except Exception as exc:
                errors.append(f"OpenAlex title: {exc}")

        kind = "success" if resolved_any else ("error" if errors else "negative")
        error = "; ".join(errors)
        if not resolved_any and not error:
            error = "No confident provider match found."
        base = self._with_destinations(base).model_copy(update={
            "candidate_id": base.candidate_id or self.stable_candidate_id(base),
            "looked_up_at": time.time(), "error": error,
            "abstract_status": "available" if base.abstract else "unavailable",
        })
        base = base.model_copy(update={"metadata_quality": self._metadata_quality(base)})
        try:
            self._save_cache(cache_key, base, kind)
        except Exception as exc:
            cache_error = f"metadata cache: {exc}"
            base = base.model_copy(update={"error": "; ".join(filter(None, (base.error, cache_error)))})
        return base

    @classmethod
    def from_candidate(cls, candidate: Mapping[str, Any]) -> PaperMetadata:
        nested = candidate.get("identifiers") if isinstance(candidate.get("identifiers"), Mapping) else {}
        identifiers = PaperIdentifiers(
            doi=cls.normalize_doi(candidate.get("doi") or nested.get("doi") or ""),
            arxiv_id=cls.normalize_arxiv(candidate.get("arxiv_id") or candidate.get("arxivId") or nested.get("arxiv_id") or ""),
            semantic_scholar_id=str(candidate.get("semantic_scholar_id") or candidate.get("paperId") or nested.get("semantic_scholar_id") or "").strip(),
            openalex_id=str(candidate.get("openalex_id") or nested.get("openalex_id") or "").strip(),
            pmid=str(candidate.get("pmid") or nested.get("pmid") or "").strip().rsplit("/", 1)[-1],
            openreview_id=str(
                candidate.get("openreview_id")
                or nested.get("openreview_id")
                or cls._openreview_id(str(candidate.get("canonical_url") or candidate.get("url") or ""))
            ).strip(),
        )
        oa = candidate.get("openAccessPdf")
        oa_url = oa.get("url") if isinstance(oa, Mapping) else ""
        pdf_url = str(candidate.get("pdf_url") or oa_url or "").strip()
        provider = str(candidate.get("provider") or "uploaded_front_matter")
        fields = dict(candidate.get("field_providers") or {})
        for field in ("title", "authors", "year", "venue", "abstract", "canonical_url", "cited_by"):
            source_key = "url" if field == "canonical_url" else field
            if candidate.get(source_key) not in (None, ""):
                fields.setdefault(field, provider)
        metadata = PaperMetadata(
            identifiers=identifiers, candidate_id=str(candidate.get("candidate_id") or ""),
            title=str(candidate.get("title") or "").strip(), authors=cls._authors(candidate.get("authors")),
            author_details=cls._author_details(candidate.get("author_details") or candidate.get("authors")),
            year=cls._year(candidate.get("year")), venue=str(candidate.get("venue") or candidate.get("journal") or "").strip(),
            abstract=str(candidate.get("abstract") or "").strip(),
            canonical_url=str(candidate.get("canonical_url") or candidate.get("url") or "").strip(),
            cited_by=cls._int(candidate.get("cited_by") or candidate.get("citationCount")),
            reference_count=cls._int(candidate.get("reference_count") or candidate.get("referenceCount")),
            is_open_access=bool(candidate.get("is_open_access") or oa_url or pdf_url), pdf_url=pdf_url,
            provider=provider, pdf_provider=str(candidate.get("pdf_provider") or (provider if pdf_url else "")),
            field_providers=fields, resolution_confidence=float(candidate.get("resolution_confidence") or 0.0),
            looked_up_at=float(candidate.get("looked_up_at") or time.time()), error=str(candidate.get("error") or ""),
            destinations=[PaperDestination(**row) for row in (candidate.get("destinations") or []) if isinstance(row, Mapping)],
            metadata_quality=str(candidate.get("metadata_quality") or "sparse"),
            abstract_status=str(candidate.get("abstract_status") or ""),
        )
        metadata = cls._with_destinations(metadata)
        return metadata.model_copy(update={
            "candidate_id": metadata.candidate_id or cls.stable_candidate_id(metadata),
            "metadata_quality": cls._metadata_quality(metadata),
        })

    @classmethod
    def from_openalex_work(cls, work: Mapping[str, Any], confidence: float = 1.0) -> PaperMetadata:
        authorships = [row for row in work.get("authorships") or [] if isinstance(row, Mapping)]
        author_details = [
            PaperAuthor(
                name=str((row.get("author") or {}).get("display_name") or "").strip(),
                openalex_id=str((row.get("author") or {}).get("id") or ""),
                orcid=str((row.get("author") or {}).get("orcid") or ""),
                profile_url=str((row.get("author") or {}).get("id") or ""),
            )
            for row in authorships if (row.get("author") or {}).get("display_name")
        ]
        authors = [author.name for author in author_details]
        best_oa, primary, ids = work.get("best_oa_location") or {}, work.get("primary_location") or {}, work.get("ids") or {}
        provider = "OpenAlex"
        pdf_url = str(best_oa.get("pdf_url") or primary.get("pdf_url") or "").strip()
        arxiv_id = cls._arxiv_id_from_openalex(work)
        publication_year = cls._year(work.get("publication_year"))
        if arxiv_id and re.match(r"^\d{4}\.", arxiv_id):
            short_year = int(arxiv_id[:2])
            arxiv_year = 2000 + short_year if short_year < 91 else 1900 + short_year
            publication_year = min(filter(None, (publication_year, arxiv_year)), default=publication_year)
        values = {
            "title": " ".join(str(work.get("display_name") or "").split()), "authors": ", ".join(filter(None, authors)),
            "year": publication_year,
            "venue": " ".join(str((primary.get("source") or {}).get("display_name") or "").split()),
            "abstract": cls.reconstruct_openalex_abstract(work.get("abstract_inverted_index")),
            "canonical_url": str(work.get("doi") or primary.get("landing_page_url") or work.get("id") or ""),
            "cited_by": cls._int(work.get("cited_by_count")),
        }
        fields = {key: provider for key, value in values.items() if value not in (None, "")}
        metadata = PaperMetadata(
            identifiers=PaperIdentifiers(
                doi=cls.normalize_doi(work.get("doi") or ids.get("doi") or ""),
                arxiv_id=arxiv_id,
                openalex_id=str(work.get("id") or ""),
                pmid=str(ids.get("pmid") or "").rstrip("/").rsplit("/", 1)[-1],
                openreview_id=cls._openreview_id(str(primary.get("landing_page_url") or "")),
            ),
            **values, author_details=author_details,
            reference_count=cls._int(work.get("referenced_works_count")),
            is_open_access=bool((work.get("open_access") or {}).get("is_oa") or pdf_url),
            pdf_url=pdf_url, provider=provider, pdf_provider=provider if pdf_url else "",
            field_providers=fields, resolution_confidence=confidence,
        )
        return cls._with_destinations(metadata)

    @classmethod
    def from_semantic_scholar(cls, paper: Mapping[str, Any]) -> PaperMetadata:
        external, oa = paper.get("externalIds") or {}, paper.get("openAccessPdf") or {}
        venue = paper.get("venue") or ((paper.get("publicationVenue") or {}).get("name") if isinstance(paper.get("publicationVenue"), Mapping) else "")
        provider = "Semantic Scholar"
        pdf_url = str(oa.get("url") or "").strip()
        author_details = [
            PaperAuthor(
                name=str(row.get("name") or "").strip(),
                semantic_scholar_id=str(row.get("authorId") or ""),
                profile_url=f"https://www.semanticscholar.org/author/{row.get('authorId')}" if row.get("authorId") else "",
            )
            for row in (paper.get("authors") or []) if isinstance(row, Mapping) and row.get("name")
        ]
        values = {
            "title": str(paper.get("title") or "").strip(), "authors": cls._authors(paper.get("authors")),
            "year": cls._year(paper.get("year")), "venue": str(venue or "").strip(),
            "abstract": str(paper.get("abstract") or "").strip(), "canonical_url": str(paper.get("url") or "").strip(),
            "cited_by": cls._int(paper.get("citationCount")),
        }
        return PaperMetadata(
            identifiers=PaperIdentifiers(
                doi=cls.normalize_doi(external.get("DOI") or paper.get("doi") or ""),
                arxiv_id=cls.normalize_arxiv(external.get("ArXiv") or paper.get("arxiv_id") or ""),
                semantic_scholar_id=str(paper.get("paperId") or paper.get("semantic_scholar_id") or ""),
                pmid=str(external.get("PubMed") or external.get("PubMedCentral") or ""),
            ),
            **values, author_details=author_details,
            reference_count=cls._int(paper.get("referenceCount")),
            is_open_access=bool(pdf_url), pdf_url=pdf_url, provider=provider,
            pdf_provider=provider if pdf_url else "", field_providers={key: provider for key, value in values.items() if value not in (None, "")}, resolution_confidence=1.0,
        )

    @staticmethod
    def stable_candidate_id(metadata: PaperMetadata) -> str:
        ids = metadata.identifiers
        identity = ids.doi or ids.arxiv_id or ids.semantic_scholar_id or ids.openalex_id
        if not identity:
            identity = "|".join((PaperMetadataService._norm(metadata.title), PaperMetadataService._norm(metadata.authors), str(metadata.year or "")))
        return hashlib.sha256(identity.lower().encode("utf-8")).hexdigest()

    @staticmethod
    def normalize_doi(value: Any) -> str:
        match = _DOI_RE.search(str(value or "").strip())
        return match.group(1).rstrip(".,;)]}").lower() if match else ""

    @staticmethod
    def normalize_arxiv(value: Any) -> str:
        clean = str(value or "").strip().rstrip(".,;)]}")
        if clean.lower().endswith(".pdf"):
            clean = clean[:-4]
        match = _ARXIV_RE.search(clean) or _RAW_ARXIV_RE.fullmatch(clean)
        return match.group(1).lower() if match else ""

    @classmethod
    def _arxiv_id_from_openalex(cls, work: Mapping[str, Any]) -> str:
        ids = work.get("ids") if isinstance(work.get("ids"), Mapping) else {}
        direct_values = [ids.get("arxiv"), work.get("doi"), ids.get("doi")]
        locations: list[Mapping[str, Any]] = []
        for key in ("best_oa_location", "primary_location"):
            value = work.get(key)
            if isinstance(value, Mapping):
                locations.append(value)
        locations.extend(row for row in work.get("locations") or [] if isinstance(row, Mapping))
        for location in locations:
            direct_values.extend((location.get("landing_page_url"), location.get("pdf_url"), location.get("id")))
        for value in direct_values:
            arxiv_id = cls.normalize_arxiv(value)
            if arxiv_id:
                return arxiv_id
        return ""

    @staticmethod
    def reconstruct_openalex_abstract(index: Any) -> str:
        words = [(pos, str(word)) for word, positions in index.items() for pos in positions if isinstance(pos, int)] if isinstance(index, Mapping) else []
        return " ".join(word for _, word in sorted(words))[:12000]

    async def _from_openalex_doi(self, doi: str) -> PaperMetadata | None:
        rows = await self._openalex({"filter": f"doi:https://doi.org/{doi}", "per-page": 1})
        return self.from_openalex_work(rows[0]) if rows else None

    async def _from_openalex_id(self, openalex_id: str) -> PaperMetadata | None:
        import httpx
        clean_id = openalex_id.rstrip("/").rsplit("/", 1)[-1]
        params: dict[str, Any] = {}
        if os.getenv("OPENALEX_API_KEY"): params["api_key"] = os.environ["OPENALEX_API_KEY"]
        if os.getenv("OPENALEX_MAILTO"): params["mailto"] = os.environ["OPENALEX_MAILTO"]
        async with httpx.AsyncClient(timeout=8.0) as client:
            response = await client.get(f"https://api.openalex.org/works/{clean_id}", params=params)
        if response.status_code == 404: return None
        if response.status_code >= 400: raise RuntimeError(f"HTTP {response.status_code}")
        return self.from_openalex_work(response.json())

    async def _from_openalex_title(self, base: PaperMetadata) -> PaperMetadata | None:
        rows = await self._openalex({"search": base.title, "per-page": 5})
        scored: list[tuple[float, PaperMetadata]] = []
        for work in rows:
            candidate = self.from_openalex_work(work)
            confidence = self._match_confidence(base, candidate)
            if confidence is not None:
                scored.append((confidence, candidate.model_copy(update={"resolution_confidence": confidence})))
        return max(scored, key=lambda row: row[0])[1] if scored else None

    async def _openalex(self, params: dict[str, Any]) -> list[Mapping[str, Any]]:
        import httpx
        params = dict(params)
        if os.getenv("OPENALEX_API_KEY"): params["api_key"] = os.environ["OPENALEX_API_KEY"]
        if os.getenv("OPENALEX_MAILTO"): params["mailto"] = os.environ["OPENALEX_MAILTO"]
        async with httpx.AsyncClient(timeout=8.0) as client:
            response = await client.get("https://api.openalex.org/works", params=params)
        if response.status_code >= 400: raise RuntimeError(f"HTTP {response.status_code}")
        return list(response.json().get("results") or [])

    async def _from_arxiv(self, arxiv_id: str) -> PaperMetadata | None:
        import httpx, xml.etree.ElementTree as et
        async with httpx.AsyncClient(timeout=8.0, follow_redirects=True) as client:
            response = await client.get("https://export.arxiv.org/api/query", params={"id_list": arxiv_id})
        if response.status_code >= 400: raise RuntimeError(f"HTTP {response.status_code}")
        ns = {"atom": "http://www.w3.org/2005/Atom", "arxiv": "http://arxiv.org/schemas/atom"}
        entry = et.fromstring(response.text).find("atom:entry", ns)
        if entry is None: return None
        authors = [" ".join((a.findtext("atom:name", default="", namespaces=ns) or "").split()) for a in entry.findall("atom:author", ns)]
        published = entry.findtext("atom:published", default="", namespaces=ns) or ""
        provider = "arXiv"
        values = {"title": " ".join((entry.findtext("atom:title", default="", namespaces=ns) or "").split()), "authors": ", ".join(filter(None, authors)), "year": self._year(published[:4]), "abstract": " ".join((entry.findtext("atom:summary", default="", namespaces=ns) or "").split()), "canonical_url": f"https://arxiv.org/abs/{arxiv_id}"}
        return PaperMetadata(
            identifiers=PaperIdentifiers(
                doi=self.normalize_doi(entry.findtext("arxiv:doi", default="", namespaces=ns) or ""),
                arxiv_id=self.normalize_arxiv(arxiv_id),
            ),
            **values,
            author_details=[PaperAuthor(name=name) for name in authors if name],
            is_open_access=True,
            pdf_url=f"https://arxiv.org/pdf/{arxiv_id}.pdf",
            provider=provider,
            pdf_provider=provider,
            field_providers={key: provider for key, value in values.items() if value},
            resolution_confidence=1.0,
        )

    async def _from_semantic_scholar(self, paper_id: str) -> PaperMetadata | None:
        import httpx
        headers = {"x-api-key": os.environ["S2_API_KEY"]} if os.getenv("S2_API_KEY") else {}
        fields = "title,year,abstract,authors,externalIds,openAccessPdf,citationCount,referenceCount,venue,publicationVenue,url"
        async with httpx.AsyncClient(timeout=8.0) as client:
            response = await client.get(f"https://api.semanticscholar.org/graph/v1/paper/{paper_id}", params={"fields": fields}, headers=headers)
        if response.status_code >= 400: raise RuntimeError(f"HTTP {response.status_code}")
        return self.from_semantic_scholar(response.json())

    async def _from_pubmed(self, pmid: str) -> PaperMetadata | None:
        import httpx
        import xml.etree.ElementTree as et

        async with httpx.AsyncClient(timeout=8.0) as client:
            response = await client.get(
                "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi",
                params={"db": "pubmed", "id": pmid, "retmode": "xml"},
            )
        if response.status_code >= 400:
            raise RuntimeError(f"HTTP {response.status_code}")
        article = et.fromstring(response.text).find(".//PubmedArticle")
        if article is None:
            return None

        title_node = article.find(".//ArticleTitle")
        title = " ".join("".join(title_node.itertext()).split()) if title_node is not None else ""
        abstract_parts = []
        for node in article.findall(".//Abstract/AbstractText"):
            text = " ".join("".join(node.itertext()).split())
            label = str(node.attrib.get("Label") or "").strip()
            if text:
                abstract_parts.append(f"{label}: {text}" if label else text)
        authors: list[PaperAuthor] = []
        for node in article.findall(".//AuthorList/Author"):
            collective = node.findtext("CollectiveName", default="") or ""
            name = collective or " ".join(filter(None, (
                node.findtext("ForeName", default="") or "",
                node.findtext("LastName", default="") or "",
            )))
            orcid = ""
            for identifier in node.findall("Identifier"):
                if str(identifier.attrib.get("Source") or "").upper() == "ORCID":
                    orcid = str(identifier.text or "").strip()
                    break
            if name.strip():
                authors.append(PaperAuthor(name=" ".join(name.split()), orcid=orcid))

        doi = ""
        for identifier in article.findall(".//ArticleIdList/ArticleId"):
            if str(identifier.attrib.get("IdType") or "").lower() == "doi":
                doi = self.normalize_doi(identifier.text or "")
                break
        year = None
        for xpath in (".//ArticleDate/Year", ".//PubDate/Year", ".//PubDate/MedlineDate"):
            raw_year = article.findtext(xpath, default="") or ""
            match = re.search(r"\b(1\d{3}|20\d{2}|21\d{2})\b", raw_year)
            if match:
                year = self._year(match.group(1))
                break
        venue = article.findtext(".//Journal/Title", default="") or ""
        provider = "PubMed"
        values = {
            "title": title,
            "authors": ", ".join(author.name for author in authors),
            "year": year,
            "venue": " ".join(venue.split()),
            "abstract": "\n\n".join(abstract_parts),
            "canonical_url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/",
        }
        return PaperMetadata(
            identifiers=PaperIdentifiers(doi=doi, pmid=pmid),
            **values,
            author_details=authors,
            provider=provider,
            field_providers={key: provider for key, value in values.items() if value not in (None, "")},
            resolution_confidence=1.0,
        )

    async def _from_crossref(self, doi: str) -> PaperMetadata | None:
        import httpx

        async with httpx.AsyncClient(timeout=8.0) as client:
            response = await client.get(f"https://api.crossref.org/works/{quote(doi, safe='')}")
        if response.status_code == 404:
            return None
        if response.status_code >= 400:
            raise RuntimeError(f"HTTP {response.status_code}")
        message = response.json().get("message") or {}
        raw_abstract = str(message.get("abstract") or "")
        abstract = " ".join(html.unescape(re.sub(r"<[^>]+>", " ", raw_abstract)).split())
        authors = []
        for row in message.get("author") or []:
            if not isinstance(row, Mapping):
                continue
            name = " ".join(filter(None, (str(row.get("given") or "").strip(), str(row.get("family") or "").strip())))
            if name:
                authors.append(PaperAuthor(name=name, orcid=str(row.get("ORCID") or "")))
        dates = message.get("published-print") or message.get("published-online") or message.get("issued") or {}
        date_parts = dates.get("date-parts") or []
        year = self._year(date_parts[0][0]) if date_parts and date_parts[0] else None
        provider = "Crossref"
        values = {
            "title": " ".join(str((message.get("title") or [""])[0]).split()),
            "authors": ", ".join(author.name for author in authors),
            "year": year,
            "venue": " ".join(str((message.get("container-title") or [""])[0]).split()),
            "abstract": abstract,
            "canonical_url": str(message.get("URL") or f"https://doi.org/{doi}"),
        }
        return PaperMetadata(
            identifiers=PaperIdentifiers(doi=self.normalize_doi(message.get("DOI") or doi)),
            **values,
            author_details=authors,
            reference_count=self._int(message.get("reference-count")),
            provider=provider,
            field_providers={key: provider for key, value in values.items() if value not in (None, "")},
            resolution_confidence=1.0,
        )

    @classmethod
    def _match_confidence(cls, base: PaperMetadata, candidate: PaperMetadata) -> float | None:
        left, right = cls._norm(base.title), cls._norm(candidate.title)
        if not left or not right: return None
        left_terms, right_terms = set(left.split()), set(right.split())
        jaccard = len(left_terms & right_terms) / max(1, len(left_terms | right_terms))
        title_score = 0.65 * difflib.SequenceMatcher(None, left, right).ratio() + 0.35 * jaccard
        if title_score < 0.86: return None
        scores, weights = [title_score], [0.75]
        author_score: float | None = None
        if base.authors:
            author_left, author_right = set(cls._norm(base.authors).split()), set(cls._norm(candidate.authors).split())
            author_score = len(author_left & author_right) / max(1, len(author_left))
            if author_score < 0.25: return None
            scores.append(author_score); weights.append(0.15)
        if base.year:
            delta = abs((candidate.year or 0) - base.year)
            if delta > 1:
                # Providers occasionally re-date exact preprints after a formal
                # publication. Exact title + strong author identity is safer
                # than rejecting the otherwise unique work.
                if title_score < 0.98 or author_score is None or author_score < 0.6:
                    return None
            else:
                scores.append(1.0 if delta == 0 else 0.5); weights.append(0.10)
        confidence = sum(score * weight for score, weight in zip(scores, weights)) / sum(weights)
        return confidence if confidence >= 0.85 else None

    @staticmethod
    def _merge(base: PaperMetadata, resolved: PaperMetadata) -> PaperMetadata:
        base_ids, resolved_ids = base.identifiers.model_dump(), resolved.identifiers.model_dump()
        ids = PaperIdentifiers(**{key: resolved_ids.get(key) or base_ids.get(key) or "" for key in base_ids})
        updates: dict[str, Any] = {"identifiers": ids, "candidate_id": base.candidate_id or resolved.candidate_id}
        providers = dict(base.field_providers)
        for field in ("title", "authors", "venue", "abstract", "canonical_url"):
            value = getattr(resolved, field)
            updates[field] = value if value not in (None, "") else getattr(base, field)
            if value not in (None, ""): providers[field] = resolved.field_providers.get(field, resolved.provider)
        if base.year and resolved.year and abs(base.year - resolved.year) > 1:
            updates["year"] = base.year
        else:
            updates["year"] = resolved.year or base.year
        updates["cited_by"] = max(base.cited_by, resolved.cited_by)
        updates["reference_count"] = max(base.reference_count, resolved.reference_count)
        updates.update({
            "field_providers": providers,
            # PDF layout extraction can mistake affiliations and markup for
            # authors.  A provider's structured author list is authoritative
            # in that situation, especially because these names are clickable
            # UI targets.
            "author_details": (
                resolved.author_details
                if resolved.author_details and PaperMetadataService._has_noisy_authors(base.author_details)
                else PaperMetadataService._merge_authors(base.author_details, resolved.author_details)
            ),
            "is_open_access": resolved.is_open_access or base.is_open_access,
            "pdf_url": resolved.pdf_url or base.pdf_url,
            "pdf_provider": resolved.pdf_provider if resolved.pdf_url else base.pdf_provider,
            "provider": resolved.provider,
            "resolution_confidence": max(base.resolution_confidence, resolved.resolution_confidence),
            "abstract_status": "available" if (resolved.abstract or base.abstract) else (resolved.abstract_status or base.abstract_status),
        })
        return PaperMetadataService._with_destinations(resolved.model_copy(update=updates))

    @staticmethod
    def _has_noisy_authors(authors: list[PaperAuthor]) -> bool:
        noise_markers = ("<", ">", "_", "university of", "institute for", "paper:", "section:", "element:")
        return any(any(marker in author.name.lower() for marker in noise_markers) for author in authors)

    @staticmethod
    def _merge_authors(base: list[PaperAuthor], resolved: list[PaperAuthor]) -> list[PaperAuthor]:
        merged: dict[str, PaperAuthor] = {}
        order: list[str] = []
        for author in [*base, *resolved]:
            key = PaperMetadataService._norm(author.name)
            if not key:
                continue
            if key not in merged:
                merged[key] = author
                order.append(key)
                continue
            current = merged[key]
            merged[key] = PaperAuthor(
                name=current.name or author.name,
                openalex_id=current.openalex_id or author.openalex_id,
                semantic_scholar_id=current.semantic_scholar_id or author.semantic_scholar_id,
                orcid=current.orcid or author.orcid,
                profile_url=current.profile_url or author.profile_url,
            )
        return [merged[key] for key in order]

    def _cache_key(self, metadata: PaperMetadata) -> str:
        raw = "|".join((
            metadata.identifiers.doi,
            metadata.identifiers.arxiv_id,
            metadata.identifiers.semantic_scholar_id,
            metadata.identifiers.openalex_id,
            metadata.identifiers.pmid,
            metadata.identifiers.openreview_id,
            self._norm(metadata.title),
            self._norm(metadata.authors),
            str(metadata.year or ""),
        ))
        return hashlib.sha256(raw.encode()).hexdigest()

    def _load_cache(self, key: str) -> PaperMetadata | None:
        path = os.path.join(self.cache_root, f"{key}.json")
        try:
            with open(path, encoding="utf-8") as file: row = json.load(file)
            if int(row.get("cache_version", 0)) != _CACHE_VERSION:
                return None
            return PaperMetadata(**row["metadata"]) if float(row.get("expires_at", 0)) > time.time() else None
        except Exception: return None

    def _save_cache(self, key: str, metadata: PaperMetadata, kind: str) -> None:
        ttl = self.success_ttl if kind == "success" and metadata.metadata_quality == "rich" else self.error_ttl if kind == "error" else self.negative_ttl
        payload = {"cache_version": _CACHE_VERSION, "expires_at": time.time() + ttl, "kind": kind, "metadata": metadata.model_dump()}
        with self._cache_lock:
            descriptor, temporary = tempfile.mkstemp(prefix=f".{key}.", suffix=".tmp", dir=self.cache_root, text=True)
            try:
                with os.fdopen(descriptor, "w", encoding="utf-8") as file:
                    json.dump(payload, file, indent=2); file.flush(); os.fsync(file.fileno())
                os.replace(temporary, os.path.join(self.cache_root, f"{key}.json"))
            finally:
                if os.path.exists(temporary): os.remove(temporary)

    @staticmethod
    def _metadata_quality(metadata: PaperMetadata) -> str:
        has_identity = bool(metadata.title and (metadata.author_details or metadata.authors))
        has_open_path = bool(metadata.destinations)
        abstract_resolved = bool(metadata.abstract or metadata.abstract_status == "unavailable")
        return "rich" if has_identity and has_open_path and abstract_resolved else "sparse"

    @staticmethod
    def _openreview_id(value: str) -> str:
        if not value or "openreview.net" not in value.lower():
            return ""
        try:
            parsed = urlparse(value)
            query = parsed.query.split("&")
            for part in query:
                key, _, candidate = part.partition("=")
                if key == "id" and candidate:
                    return candidate
        except ValueError:
            pass
        return ""

    @classmethod
    def _author_details(cls, value: Any) -> list[PaperAuthor]:
        if not isinstance(value, list):
            return [PaperAuthor(name=name.strip()) for name in str(value or "").split(",") if name.strip()]
        authors: list[PaperAuthor] = []
        for row in value:
            if isinstance(row, PaperAuthor):
                authors.append(row)
                continue
            if isinstance(row, Mapping):
                nested = row.get("author") if isinstance(row.get("author"), Mapping) else row
                name = str(nested.get("display_name") or nested.get("name") or "").strip()
                if not name:
                    continue
                openalex_id = str(nested.get("openalex_id") or nested.get("id") or "") if "openalex.org" in str(nested.get("id") or "") else str(nested.get("openalex_id") or "")
                semantic_id = str(nested.get("semantic_scholar_id") or nested.get("authorId") or "")
                authors.append(PaperAuthor(
                    name=name,
                    openalex_id=openalex_id,
                    semantic_scholar_id=semantic_id,
                    orcid=str(nested.get("orcid") or nested.get("ORCID") or ""),
                    profile_url=str(nested.get("profile_url") or nested.get("id") or ""),
                ))
                continue
            name = str(row or "").strip()
            if name:
                authors.append(PaperAuthor(name=name))
        return authors

    @classmethod
    def _with_destinations(cls, metadata: PaperMetadata) -> PaperMetadata:
        ids = metadata.identifiers
        rows: list[PaperDestination] = []
        if ids.arxiv_id:
            rows.append(PaperDestination(label="arXiv", url=f"https://arxiv.org/abs/{ids.arxiv_id}", provider="arXiv"))
        if ids.openreview_id:
            rows.append(PaperDestination(label="OpenReview", url=f"https://openreview.net/forum?id={ids.openreview_id}", provider="OpenReview"))
        if ids.doi:
            rows.append(PaperDestination(label="DOI", url=f"https://doi.org/{ids.doi}", provider="Crossref"))
        if ids.pmid:
            rows.append(PaperDestination(label="PubMed", url=f"https://pubmed.ncbi.nlm.nih.gov/{ids.pmid}/", provider="PubMed"))
        if ids.semantic_scholar_id:
            rows.append(PaperDestination(label="Semantic Scholar", url=f"https://www.semanticscholar.org/paper/{ids.semantic_scholar_id}", provider="Semantic Scholar"))
        if metadata.canonical_url:
            rows.append(PaperDestination(label="Source", url=metadata.canonical_url, provider=metadata.provider))
        if ids.openalex_id:
            rows.append(PaperDestination(label="OpenAlex", url=ids.openalex_id if ids.openalex_id.startswith("https://") else f"https://openalex.org/{ids.openalex_id}", provider="OpenAlex"))
        if metadata.title:
            rows.append(PaperDestination(label="Google Scholar", url=f"https://scholar.google.com/scholar?q={quote_plus(metadata.title)}", provider="Google Scholar"))

        seen: set[str] = set()
        destinations: list[PaperDestination] = []
        for row in [*metadata.destinations, *rows]:
            url = str(row.url or "").strip()
            if not url.startswith("https://") or url.lower() in seen:
                continue
            seen.add(url.lower())
            destinations.append(row.model_copy(update={"url": url}))
        updates: dict[str, Any] = {"destinations": destinations}
        if ids.arxiv_id:
            updates.update({
                "pdf_url": f"https://arxiv.org/pdf/{ids.arxiv_id}.pdf",
                "is_open_access": True,
                "pdf_provider": "arXiv",
            })
        return metadata.model_copy(update=updates)

    @staticmethod
    def _norm(value: str) -> str: return re.sub(r"\s+", " ", re.sub(r"[^a-z0-9]+", " ", value.lower())).strip()
    @staticmethod
    def _authors(value: Any) -> str:
        return ", ".join(str(item.get("name") if isinstance(item, Mapping) else item).strip() for item in value if item) if isinstance(value, list) else str(value or "").strip()
    @staticmethod
    def _year(value: Any) -> int | None:
        try: year = int(value); return year if 1000 <= year <= 9999 else None
        except (TypeError, ValueError): return None
    @staticmethod
    def _int(value: Any) -> int:
        try: return max(0, int(value or 0))
        except (TypeError, ValueError): return 0