app/services/citation_graph.py

from __future__ import annotations

import inspect
import io
import json
import os
import re
import asyncio
import hashlib
import tempfile
import threading
import time
import xml.etree.ElementTree as ET
from typing import Any, Callable, Literal

from pydantic import BaseModel, Field

from app.services.project_service import ProjectService
from app.services.paper_metadata import PaperMetadataService
from app.services.paper_acquisition import sanitize_acquisition_candidate

CitationNodeType = Literal[
    "paper",
    "dataset",
    "method",
    "task",
    "model",
    "benchmark",
    "metric",
    "formula",
    "research_question",
    "concept",
]
CitationStatus = Literal["owned", "connector"]
CitationRelation = Literal["cites"]

_DEFAULT_ROOT = os.path.expanduser("~/.studybuddy/citation_graphs")
_ARXIV_RE = re.compile(r"(?:arXiv\s*:\s*)?(\d{4}\.\d{4,5}(?:v\d+)?)", re.I)
_DOI_RE = re.compile(r"\b10\.\d{4,9}/[-._;()/:A-Z0-9]+\b", re.I)
_S2_URL_RE = re.compile(r"semanticscholar\.org/(?:paper/)?(?:[^/\s]+/)?([0-9a-f]{40})", re.I)
_ABSTRACT_RE = re.compile(
    r"\bAbstract\b\s*(.*?)(?=\n\s*(?:1\.?\s+)?(?:Introduction|Background|Related Work)\b|\n\s*I\.?\s+Introduction\b)",
    re.I | re.S,
)
_REFERENCES_RE = re.compile(r"\n\s*(?:References|Bibliography)\s*\n(.*)$", re.I | re.S)

class PaperExtraction(BaseModel):
    file_id: str
    filename: str
    text: str
    title: str
    authors: str = ""
    abstract: str = ""
    year: int | None = None
    arxiv_id: str = ""
    doi: str = ""
    semantic_scholar_id: str = ""
    openalex_id: str = ""
    venue: str = ""
    canonical_url: str = ""
    cited_by: int = 0
    is_open_access: bool = False
    pdf_url: str = ""
    metadata_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 = 0.0
    metadata_error: str = ""
    document_metadata: dict[str, str] = Field(default_factory=dict)
    references: list[str] = Field(default_factory=list)
    unmatched_references: list[str] = Field(default_factory=list)
    evidence: dict[str, list[str]] = Field(default_factory=dict)


class CitationGraphNode(BaseModel):
    id: str
    type: CitationNodeType
    label: str
    title: str = ""
    filename: str = ""
    abstract: str = ""
    url: str = ""
    year: int | None = None
    authors: str = ""
    cited_by: int = 0
    status: CitationStatus = "connector"
    source_file_ids: list[str] = Field(default_factory=list)
    confidence: float = 0.5
    arxiv_id: str = ""
    doi: str = ""
    semantic_scholar_id: str = ""
    evidence_by_file: dict[str, list[str]] = Field(default_factory=dict)
    unmatched_references: list[str] = Field(default_factory=list)
    metadata: dict[str, Any] = Field(default_factory=dict)


class CitationGraphEdge(BaseModel):
    id: str
    source: str
    target: str
    relation: CitationRelation
    evidence: str
    source_file_id: str = ""
    confidence: float = 0.5


class CitationGraph(BaseModel):
    project_id: str
    nodes: list[CitationGraphNode] = Field(default_factory=list)
    edges: list[CitationGraphEdge] = Field(default_factory=list)
    updated_at: float = 0
    file_fingerprint: str = ""
    warnings: list[str] = Field(default_factory=list)

    @property
    def nodes_by_id(self) -> dict[str, CitationGraphNode]:
        return {node.id: node for node in self.nodes}


class CitationGraphService:
    _refresh_locks: dict[str, threading.Lock] = {}
    _refresh_locks_guard = threading.Lock()
    _PROJECT_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}")

    def __init__(
        self,
        root: str | None = None,
        project_service: ProjectService | None = None,
        text_loader: Callable[[str], list[dict[str, str]]] | None = None,
        paper_resolver: Callable[[str], Any] | None = None,
        citation_fetcher: Callable[[PaperExtraction], Any] | None = None,
        metadata_resolver: Callable[[PaperExtraction], Any] | None = None,
        metadata_service: PaperMetadataService | None = None,
    ) -> None:
        self.root = root or _DEFAULT_ROOT
        self.project_service = project_service or ProjectService()
        self.text_loader = text_loader or self._load_project_texts
        self.paper_resolver = paper_resolver or self._resolve_paper
        self.citation_fetcher = citation_fetcher or self._fetch_semantic_scholar_references
        self.paper_metadata_service = metadata_service or PaperMetadataService()
        self.metadata_resolver = metadata_resolver or self._resolve_paper_metadata
        os.makedirs(self.root, exist_ok=True)

    @classmethod
    def is_valid_project_id(cls, project_id: str) -> bool:
        return bool(isinstance(project_id, str) and cls._PROJECT_ID_RE.fullmatch(project_id))

    @classmethod
    def _refresh_lock(cls, project_id: str) -> threading.Lock:
        with cls._refresh_locks_guard:
            return cls._refresh_locks.setdefault(project_id, threading.Lock())

    def _require_project_id(self, project_id: str) -> None:
        if not self.is_valid_project_id(project_id):
            raise ValueError("Invalid project id.")

    def _path(self, project_id: str) -> str:
        self._require_project_id(project_id)
        return os.path.join(self.root, f"{project_id}.json")

    def load(self, project_id: str) -> CitationGraph:
        path = self._path(project_id)
        if not os.path.exists(path):
            return CitationGraph(project_id=project_id)
        try:
            with open(path, encoding="utf-8") as f:
                return CitationGraph(**json.load(f))
        except Exception as exc:
            return CitationGraph(project_id=project_id, warnings=[f"Could not read citation graph: {exc}"])

    def save(self, graph: CitationGraph) -> None:
        os.makedirs(self.root, exist_ok=True)
        path = self._path(graph.project_id)
        descriptor, temporary = tempfile.mkstemp(prefix=f".{graph.project_id}.", suffix=".tmp", dir=self.root, text=True)
        try:
            with os.fdopen(descriptor, "w", encoding="utf-8") as file:
                json.dump(graph.model_dump(), file, indent=2)
                file.flush()
                os.fsync(file.fileno())
            os.replace(temporary, path)
        finally:
            if os.path.exists(temporary):
                os.remove(temporary)

    async def refresh(self, project_id: str) -> CitationGraph:
        self._require_project_id(project_id)
        lock = self._refresh_lock(project_id)
        while not lock.acquire(blocking=False):
            await asyncio.sleep(0.02)
        try:
            return await self._refresh_unlocked(project_id)
        finally:
            lock.release()

    async def _refresh_unlocked(self, project_id: str) -> CitationGraph:
        graph = CitationGraph(project_id=project_id, updated_at=time.time())
        project = self.project_service.load(project_id)
        if project is None:
            graph.warnings.append("Project not found.")
            self.save(graph)
            return graph

        nodes: dict[str, CitationGraphNode] = {}
        edges: dict[str, CitationGraphEdge] = {}
        items = self.text_loader(project_id)
        by_file_id = {item.get("file_id", ""): item for item in items}
        papers: dict[str, PaperExtraction] = {}

        for file in project.files:
            item = by_file_id.get(file.file_id, {"file_id": file.file_id, "filename": file.filename, "text": ""})
            paper = self._extract_paper_metadata(
                file_id=file.file_id,
                filename=file.filename,
                text=item.get("text", ""),
                document_metadata=item.get("document_metadata") or {},
                front_matter_text=item.get("front_matter_text") or "",
                reference_evidence_ids=list(item.get("reference_evidence_ids") or []),
            )
            # A downloaded paper has already passed provider resolution before
            # it entered the project.  Prefer that exact, durable association
            # over lossy PDF-layout inference or another network lookup.
            metadata = self.paper_metadata_service.file_metadata(file.file_id)
            if metadata is None:
                metadata = await self._maybe_resolve_metadata(paper)
            if metadata:
                paper = self._with_resolved_metadata(paper, metadata)
            papers[file.file_id] = paper
            node_id = f"paper:{file.file_id}"
            nodes[node_id] = CitationGraphNode(
                id=node_id,
                type="paper",
                label=paper.title or file.filename,
                title=paper.title or file.filename,
                filename=file.filename,
                abstract=paper.abstract,
                url=paper.canonical_url,
                authors=paper.authors,
                year=paper.year,
                cited_by=paper.cited_by,
                arxiv_id=paper.arxiv_id,
                doi=paper.doi,
                semantic_scholar_id=paper.semantic_scholar_id,
                status="owned",
                source_file_ids=[file.file_id],
                evidence_by_file={
                    file.file_id: list(paper.evidence.get("reference_ids") or [])
                },
                confidence=1.0,
                unmatched_references=paper.unmatched_references,
                metadata={
                    "references": paper.references[:20],
                    "cites": [],
                    "cited_by_project": [],
                    "verified_unmatched_references": [],
                    "venue": paper.venue,
                    "is_open_access": paper.is_open_access,
                    "pdf_url": paper.pdf_url,
                    "provider": paper.metadata_provider,
                    "pdf_provider": paper.pdf_provider,
                    "field_providers": paper.field_providers,
                    "resolution_confidence": paper.resolution_confidence,
                    "openalex_id": paper.openalex_id,
                    "looked_up_at": paper.looked_up_at,
                    "error": paper.metadata_error,
                    "external_candidate_id": str((metadata or {}).get("candidate_id") or ""),
                    "author_details": list((metadata or {}).get("author_details") or []),
                    "destinations": list((metadata or {}).get("destinations") or []),
                    "metadata_quality": str((metadata or {}).get("metadata_quality") or "sparse"),
                    "abstract_status": str((metadata or {}).get("abstract_status") or ("available" if paper.abstract else "")),
                    "reference_count": int((metadata or {}).get("reference_count") or 0),
                },
            )

        await self._add_verified_citation_edges(nodes, edges, papers, graph.warnings)

        for paper in papers.values():
            node = nodes.get(f"paper:{paper.file_id}")
            if node is not None:
                node.unmatched_references = paper.unmatched_references[:20]
                node.metadata["references"] = paper.references[:20]

        graph.nodes = list(nodes.values())
        graph.edges = list(edges.values())
        graph.file_fingerprint = self._fingerprint_files(project.files)
        self.save(graph)
        return graph

    @staticmethod
    def _with_resolved_metadata(paper: PaperExtraction, metadata: dict[str, Any]) -> PaperExtraction:
        return paper.model_copy(update={
            "title": metadata.get("title") or paper.title,
            "authors": metadata.get("authors") or paper.authors,
            "abstract": metadata.get("abstract") or paper.abstract,
            "year": metadata.get("year") or paper.year,
            "doi": metadata.get("doi") or paper.doi,
            "arxiv_id": metadata.get("arxiv_id") or paper.arxiv_id,
            "semantic_scholar_id": metadata.get("semantic_scholar_id") or paper.semantic_scholar_id,
            "openalex_id": metadata.get("openalex_id") or paper.openalex_id,
            "venue": metadata.get("venue") or paper.venue,
            "canonical_url": metadata.get("url") or paper.canonical_url,
            "cited_by": metadata.get("cited_by") or paper.cited_by,
            "is_open_access": bool(metadata.get("is_open_access") or paper.is_open_access),
            "pdf_url": metadata.get("pdf_url") or paper.pdf_url,
            "metadata_provider": metadata.get("provider") or paper.metadata_provider,
            "pdf_provider": metadata.get("pdf_provider") or paper.pdf_provider,
            "field_providers": metadata.get("field_providers") or paper.field_providers,
            "resolution_confidence": metadata.get("resolution_confidence") or paper.resolution_confidence,
            "looked_up_at": metadata.get("looked_up_at") or paper.looked_up_at,
            "metadata_error": metadata.get("error") or paper.metadata_error,
        })

    def file_fingerprint(self, project_id: str) -> str:
        self._require_project_id(project_id)
        project = self.project_service.load(project_id)
        return self._fingerprint_files(project.files if project is not None else [])

    def metadata_is_fresh(self, graph: CitationGraph) -> bool:
        if not graph.nodes:
            return False
        for node in graph.nodes:
            metadata = node.metadata or {}
            if not self.paper_metadata_service.is_fresh(metadata.get("looked_up_at"), str(metadata.get("error") or "")):
                return False
            for candidate in metadata.get("verified_unmatched_references") or []:
                if isinstance(candidate, dict) and not self.paper_metadata_service.is_fresh(candidate.get("looked_up_at"), str(candidate.get("error") or "")):
                    return False
        return True

    @staticmethod
    def _fingerprint_files(files: Any) -> str:
        file_ids = sorted(file.file_id for file in files)
        return hashlib.sha256("\n".join(file_ids).encode("utf-8")).hexdigest()

    def _load_project_texts(self, project_id: str) -> list[dict[str, Any]]:
        from app.rag.retrieval_service import PaperEvidenceService

        return PaperEvidenceService().project_citation_material(project_id)  # type: ignore[return-value]

    async def _resolve_paper(self, title: str) -> dict[str, Any] | None:
        from app.services.scholar_service import fetch_top_papers

        papers = await fetch_top_papers(title, n=1)
        if not papers:
            return None
        return papers[0]

    async def _maybe_resolve(self, title: str) -> dict[str, Any] | None:
        try:
            result = self.paper_resolver(title)
            if inspect.isawaitable(result):
                result = await result
            return result
        except Exception:
            return None

    async def _maybe_resolve_metadata(self, paper: PaperExtraction) -> dict[str, Any] | None:
        try:
            result = self.metadata_resolver(paper)
            if inspect.isawaitable(result):
                result = await result
            return result or None
        except Exception:
            return None

    def _extract_paper_metadata(
        self,
        file_id: str,
        filename: str,
        text: str,
        document_metadata: dict[str, str] | None = None,
        front_matter_text: str = "",
        reference_evidence_ids: list[str] | None = None,
    ) -> PaperExtraction:
        document_metadata = document_metadata or {}
        document_title = self._document_metadata_value(document_metadata, "Title")
        if document_title.lower() in {"untitled", "unknown"} or self._is_filename_like_title(document_title, filename):
            document_title = ""
        document_authors = self._document_metadata_value(document_metadata, "Author")
        creation_date = self._document_metadata_value(document_metadata, "CreationDate")
        creation_year_match = re.search(r"(?:D:)?(19\d{2}|20\d{2})", creation_date)
        creation_year = int(creation_year_match.group(1)) if creation_year_match else None
        # Canonical evidence starts with a document label for retrieval.  The
        # first-page source text is the only safe heuristic identity surface.
        identity_text = front_matter_text or self._before_references(text)
        heuristic_title = self._extract_title(identity_text)
        title = document_title or heuristic_title or filename
        authors = document_authors or self._extract_authors(identity_text, heuristic_title or title)
        # Bibliographies may contain decades of publication years.  They must
        # not be used as the publication year of the uploaded paper.
        year = creation_year or self._extract_year(front_matter_text or self._before_references(text))
        field_providers: dict[str, str] = {}
        if title:
            field_providers["title"] = "pdf_document_metadata" if document_title else "pdf_first_page"
        if authors:
            field_providers["authors"] = "pdf_document_metadata" if document_authors else "pdf_first_page"
        if year:
            field_providers["year"] = "pdf_document_metadata" if creation_year else "pdf_first_page"
        references = self._extract_references(text)
        # Identity extraction is limited to PDF metadata plus the front matter;
        # identifiers in the bibliography belong to cited works, not this upload.
        arxiv_id = self._extract_arxiv_id(front_matter_text, filename, document_metadata)
        doi = self._extract_doi(front_matter_text, document_metadata)
        semantic_scholar_id = self._extract_semantic_scholar_id(front_matter_text, document_metadata)
        return PaperExtraction(
            file_id=file_id,
            filename=filename,
            text=text,
            title=title,
            authors=authors,
            abstract=self._extract_abstract(identity_text) or self._extract_abstract(text),
            year=year,
            arxiv_id=arxiv_id,
            doi=doi,
            semantic_scholar_id=semantic_scholar_id,
            field_providers=field_providers,
            document_metadata=document_metadata,
            references=references,
            unmatched_references=references[:],
            evidence={"reference_ids": list(reference_evidence_ids or [])},
        )

    @staticmethod
    def _document_metadata_value(document_metadata: dict[str, str], key: str) -> str:
        for candidate_key, value in document_metadata.items():
            if candidate_key.lstrip("/").lower() == key.lower():
                return re.sub(r"\s+", " ", str(value)).strip()
        return ""

    @staticmethod
    def _is_filename_like_title(value: str, filename: str) -> bool:
        def normalized(item: str) -> str:
            stem = os.path.splitext(os.path.basename(item))[0]
            return re.sub(r"[^a-z0-9]+", "", stem.lower())

        return bool(value and filename and normalized(value) == normalized(filename))

    @staticmethod
    def _before_references(text: str) -> str:
        return _REFERENCES_RE.split(text, maxsplit=1)[0]

    @staticmethod
    def _extract_pdf_text_and_metadata(file_bytes: bytes) -> tuple[str, dict[str, str]]:
        text, metadata, _ = CitationGraphService._extract_pdf_document(file_bytes)
        return text, metadata

    @staticmethod
    def _extract_pdf_document(file_bytes: bytes) -> tuple[str, dict[str, str], str]:
        from pypdf import PdfReader

        reader = PdfReader(io.BytesIO(file_bytes))
        raw_metadata = reader.metadata or {}
        metadata = {
            str(key).lstrip("/"): str(value)
            for key, value in raw_metadata.items()
            if value is not None and str(value).strip()
        }
        pages = [page.extract_text() or "" for page in reader.pages]
        return "\n".join(pages), metadata, (pages[0] if pages else "")

    @staticmethod
    def _extract_full_text(file_bytes: bytes, filename: str) -> str:
        ext = filename.rsplit(".", 1)[-1].lower()
        if ext == "pdf":
            return CitationGraphService._extract_pdf_text_and_metadata(file_bytes)[0]
        if ext == "docx":
            import docx

            doc = docx.Document(io.BytesIO(file_bytes))
            return "\n".join(p.text for p in doc.paragraphs)
        return file_bytes.decode("utf-8", errors="replace")

    @staticmethod
    def _clean_line(line: str) -> str:
        return re.sub(r"\s+", " ", line.strip())

    @classmethod
    def _extract_title(cls, text: str) -> str:
        lines = [cls._clean_line(line) for line in cls._before_references(text).splitlines()]
        lines = [line for line in lines if line]
        banned = (
            "arxiv",
            "published as",
            "conference paper",
            "proceedings",
            "abstract",
            "keywords",
            "journal",
            "copyright",
        )
        for line in lines[:80]:
            clean = re.sub(r"[*`#]", "", line).strip(" .:-")
            lower = clean.lower()
            alpha_count = sum(ch.isalpha() for ch in clean)
            if alpha_count < 8 or any(marker in lower for marker in banned):
                continue
            if lower.startswith(("paper:", "section:", "element:", "file:")):
                continue
            if "@" in clean or lower.startswith(("http", "www.")):
                continue
            if len(clean) > 160:
                continue
            return clean
        return ""

    @classmethod
    def _extract_authors(cls, text: str, title: str) -> str:
        # PyMuPDF4LLM emits markdown emphasis (``**bold**``) around headings and
        # author lines. Strip it before matching so a heading like "**Abstract**"
        # still hits the stop condition below instead of being read as an author.
        lines = [
            cls._strip_markdown_marks(cls._clean_line(line))
            for line in cls._before_references(text).splitlines()
        ]
        lines = [line for line in lines if line]
        if not title:
            return ""
        normalized_title = re.sub(r"[^a-z0-9]+", "", title.lower())
        idx = next((
            position for position, line in enumerate(lines)
            if re.sub(r"[^a-z0-9]+", "", line.lower()) == normalized_title
        ), -1)
        if idx < 0:
            return ""
        author_lines: list[str] = []
        for line in lines[idx + 1 : idx + 8]:
            lower = line.lower()
            if lower.startswith(("abstract", "keywords", "1 introduction", "introduction")):
                break
            if lower.startswith(("paper:", "section:", "element:", "file:")):
                continue
            if "@" in line:
                # Many single-column ML papers (OpenAI/Google Brain house style)
                # pack "Name Affiliation email" onto one line with no separator.
                # Recovering at least the leading name keeps the byline from
                # going empty just because it shares a line with an address.
                name_prefix = cls._name_prefix_before_affiliation(line)
                if name_prefix:
                    author_lines.append(name_prefix)
                continue
            if lower.startswith(("http", "www.")):
                continue
            if sum(ch.isalpha() for ch in line) < 4:
                continue
            author_lines.append(line)
            if len(author_lines) >= 3:
                break
        return ", ".join(author_lines)[:300]

    @staticmethod
    def _name_prefix_before_affiliation(line: str) -> str:
        """Recover a leading person name from a combined "Name Affiliation email" line.

        Institution tokens are trimmed from the front by looking for the first
        word that reads as an acronym/brand (ALL-CAPS, or an internal capital
        like "OpenAI") rather than a personal name. What remains is kept only
        if it still looks like a short single name (2-4 words); anything
        longer is more likely several authors' names run together with no
        separator, which this heuristic can't safely split.
        """
        prefix = line.split("@", 1)[0].strip()
        name_tokens: list[str] = []
        for token in prefix.split():
            word = token.strip(".,")
            if not word.isalpha() or word.isupper() or word[1:] != word[1:].lower():
                break
            name_tokens.append(word)
        return " ".join(name_tokens) if 2 <= len(name_tokens) <= 4 else ""

    @staticmethod
    def _strip_markdown_marks(line: str) -> str:
        # PyMuPDF4LLM also emits raw HTML for superscript footnote markers
        # (``<sup>...</sup>``) and a leading ``>`` for the footnote text itself
        # (affiliation lines keyed to those markers). Both need to go before a
        # line is judged as an author candidate, or the leftover markup trips
        # false positives (or false negatives) in the checks around this.
        line = re.sub(r"<[^>]*>", "", line)
        line = re.sub(r"^>\s*", "", line)
        return re.sub(r"[*`#_]", "", line).strip()

    @classmethod
    def _extract_abstract(cls, text: str) -> str:
        # Same markdown-emphasis problem as `_extract_authors`: the "Introduction"
        # boundary this regex looks for is itself a markdown heading (e.g.
        # "**1 Introduction**"), so the section markers must be stripped first or
        # the lookahead never matches and the whole abstract capture fails.
        cleaned = "\n".join(cls._strip_markdown_marks(line) for line in text.splitlines())
        match = _ABSTRACT_RE.search(cleaned)
        if not match:
            return ""
        abstract = re.sub(r"\s+", " ", match.group(1)).strip()
        return abstract[:1200]

    @staticmethod
    def _extract_year(text: str) -> int | None:
        years = [int(y) for y in re.findall(r"\b(19[5-9]\d|20[0-4]\d)\b", text[:6000])]
        return max(years) if years else None

    @staticmethod
    def _metadata_text(document_metadata: dict[str, str] | None) -> str:
        return "\n".join(str(value) for value in (document_metadata or {}).values())

    @staticmethod
    def _extract_arxiv_id(text: str, filename: str = "", document_metadata: dict[str, str] | None = None) -> str:
        filename_match = re.search(r"(\d{4}\.\d{4,5})(?:v\d+)?", filename)
        if filename_match:
            return filename_match.group(1)
        match = _ARXIV_RE.search(CitationGraphService._metadata_text(document_metadata)) or _ARXIV_RE.search(text)
        return match.group(1).split("v", 1)[0] if match else ""

    @staticmethod
    def _extract_doi(text: str, document_metadata: dict[str, str] | None = None) -> str:
        match = _DOI_RE.search(CitationGraphService._metadata_text(document_metadata)) or _DOI_RE.search(text)
        return match.group(0).rstrip(".,);]") if match else ""

    @staticmethod
    def _extract_semantic_scholar_id(text: str, document_metadata: dict[str, str] | None = None) -> str:
        match = _S2_URL_RE.search(CitationGraphService._metadata_text(document_metadata)) or _S2_URL_RE.search(text)
        return match.group(1) if match else ""

    async def _resolve_paper_metadata(self, paper: PaperExtraction) -> dict[str, Any] | None:
        metadata = await self.paper_metadata_service.resolve(paper.model_dump())
        return metadata.candidate_dict()

    @staticmethod
    async def _fetch_arxiv_metadata(arxiv_id: str) -> dict[str, Any] | None:
        try:
            import httpx

            clean_id = arxiv_id.split("v", 1)[0]
            async with httpx.AsyncClient(timeout=8.0) as client:
                resp = await client.get("https://export.arxiv.org/api/query", params={"id_list": clean_id})
            if resp.status_code >= 400:
                return None
            root = ET.fromstring(resp.text)
            ns = {"atom": "http://www.w3.org/2005/Atom", "arxiv": "http://arxiv.org/schemas/atom"}
            entry = root.find("atom:entry", ns)
            if entry is None:
                return None
            title = " ".join((entry.findtext("atom:title", default="", namespaces=ns) or "").split())
            abstract = " ".join((entry.findtext("atom:summary", default="", namespaces=ns) or "").split())
            published = entry.findtext("atom:published", default="", namespaces=ns) or ""
            authors = [
                " ".join((author.findtext("atom:name", default="", namespaces=ns) or "").split())
                for author in entry.findall("atom:author", ns)
            ]
            doi = entry.findtext("arxiv:doi", default="", namespaces=ns) or ""
            return {
                "title": title,
                "authors": ", ".join([a for a in authors if a]),
                "abstract": abstract,
                "year": int(published[:4]) if published[:4].isdigit() else None,
                "doi": doi,
                "url": f"https://arxiv.org/abs/{clean_id}",
            }
        except Exception:
            return None

    @staticmethod
    async def _fetch_openalex_metadata(identifier: str) -> dict[str, Any] | None:
        try:
            import httpx

            api_key = os.getenv("OPENALEX_API_KEY", "")
            params = {"filter": f"doi:{identifier}"}
            if api_key:
                params["api_key"] = api_key
            mailto = os.getenv("OPENALEX_MAILTO", "")
            if mailto:
                params["mailto"] = mailto
            async with httpx.AsyncClient(timeout=8.0) as client:
                resp = await client.get("https://api.openalex.org/works", params=params)
            if resp.status_code >= 400:
                return None
            results = resp.json().get("results", [])
            if not results:
                return None
            work = results[0]
            authors = [
                ((authorship.get("author") or {}).get("display_name") or "")
                for authorship in work.get("authorships", [])
            ]
            return {
                "title": work.get("display_name") or "",
                "authors": ", ".join([a for a in authors if a]),
                "year": work.get("publication_year"),
                "doi": (work.get("doi") or "").replace("https://doi.org/", ""),
                "url": work.get("id", ""),
            }
        except Exception:
            return None

    @staticmethod
    async def _fetch_semantic_scholar_metadata(identifier: str) -> dict[str, Any] | None:
        try:
            import httpx

            headers = {}
            api_key = os.environ.get("S2_API_KEY")
            if api_key:
                headers["x-api-key"] = api_key
            async with httpx.AsyncClient(timeout=8.0) as client:
                resp = await client.get(
                    f"https://api.semanticscholar.org/graph/v1/paper/{identifier}",
                    params={"fields": "title,year,abstract,authors,externalIds"},
                    headers=headers,
                )
            if resp.status_code >= 400:
                return None
            data = resp.json()
            external = data.get("externalIds") or {}
            return {
                "title": data.get("title") or "",
                "authors": ", ".join([a.get("name", "") for a in data.get("authors", []) if a.get("name")]),
                "abstract": data.get("abstract") or "",
                "year": data.get("year"),
                "doi": external.get("DOI", ""),
                "semantic_scholar_id": data.get("paperId", ""),
            }
        except Exception:
            return None

    @staticmethod
    def _extract_references(text: str) -> list[str]:
        match = _REFERENCES_RE.search(text)
        ref_text = match.group(1) if match else text[-50000:]
        ref_text = ref_text.replace("\r", "\n")
        parts = re.split(
            r"\n\s*(?=(?:[-*]\s+|\[\d+\]|\d+\.\s|[A-Z][A-Za-z-]+,\s+[A-Z]))",
            ref_text,
        )
        refs: list[str] = []
        for part in parts:
            part = re.sub(r"\s+", " ", part).strip()
            if not part:
                continue
            if len(part) < 20:
                continue
            refs.append(part[:700])
        return refs[:80]

    @staticmethod
    def _norm(value: str) -> str:
        value = value.lower()
        value = re.sub(r"[^a-z0-9]+", " ", value)
        return re.sub(r"\s+", " ", value).strip()

    @classmethod
    def _title_match_score(cls, title: str, text: str) -> float:
        norm_title = cls._norm(title)
        norm_text = cls._norm(text)
        if not norm_title or not norm_text:
            return 0.0
        if norm_title in norm_text:
            return 1.0
        title_terms = {term for term in norm_title.split() if len(term) > 2}
        if not title_terms:
            return 0.0
        text_terms = set(norm_text.split())
        return len(title_terms & text_terms) / len(title_terms)

    async def _add_verified_citation_edges(
        self,
        nodes: dict[str, CitationGraphNode],
        edges: dict[str, CitationGraphEdge],
        papers: dict[str, PaperExtraction],
        warnings: list[str],
    ) -> None:
        for source in papers.values():
            source_node = nodes.get(f"paper:{source.file_id}")
            if source_node is None:
                continue
            if not self._paper_identifier(source):
                warning = f"No DOI, arXiv ID, or Semantic Scholar ID for {source.title}; citation sync skipped."
                warnings.append(warning)
                source_node.metadata["citation_sync_error"] = warning
                continue
            try:
                result = self.citation_fetcher(source)
                if inspect.isawaitable(result):
                    result = await result
            except Exception as exc:
                result = {"references": [], "error": str(exc)}
            error = (result or {}).get("error")
            if error:
                warning = f"{source.title}: {error}"
                warnings.append(warning)
                source_node.metadata["citation_sync_error"] = error
                continue
            verified_refs = [(ref if isinstance(ref, dict) else {"title": str(ref)}) for ref in (result or {}).get("references", [])]
            verified_unmatched: list[dict[str, Any]] = []
            for ref in verified_refs:
                if "externalIds" in ref or "openAccessPdf" in ref:
                    candidate = self.paper_metadata_service.from_semantic_scholar(ref).candidate_dict()
                else:
                    candidate = self.paper_metadata_service.from_candidate(ref).candidate_dict()
                if candidate.get("provider") == "uploaded_front_matter":
                    candidate["provider"] = "Semantic Scholar"
                candidate = sanitize_acquisition_candidate(candidate)
                target = self._match_reference_to_local(candidate, papers)
                if target and target.file_id != source.file_id:
                    local_evidence = self._local_reference_evidence(source, target)
                    if not local_evidence:
                        coverage = f"Verified provider match to uploaded paper '{target.title}', but no local reference text evidence was extracted."
                        source_node.metadata.setdefault("coverage_warnings", []).append(coverage)
                        warnings.append(f"{source.title}: {coverage}")
                        continue
                    source_id = f"paper:{source.file_id}"
                    target_id = f"paper:{target.file_id}"
                    evidence = f"Verified reference: {local_evidence}"
                    self._add_edge(edges, source_id, target_id, "cites", evidence, source.file_id, 1.0)
                    self._append_unique(source_node.metadata.setdefault("cites", []), target_id)
                    target_node = nodes.get(target_id)
                    if target_node is not None:
                        self._append_unique(target_node.metadata.setdefault("cited_by_project", []), source_id)
                else:
                    title = str(candidate.get("title") or "").strip()
                    if title:
                        verified_unmatched.append(candidate)
            source_node.metadata["verified_unmatched_references"] = verified_unmatched[:50]
            source_node.unmatched_references = [
                str(ref.get("title", "")) for ref in verified_unmatched[:20] if ref.get("title")
            ]

    @staticmethod
    def _append_unique(values: list[str], value: str) -> None:
        if value not in values:
            values.append(value)
            values.sort()

    @staticmethod
    def _paper_identifier(paper: PaperExtraction) -> str:
        if paper.arxiv_id:
            return f"ARXIV:{paper.arxiv_id}"
        if paper.doi:
            return f"DOI:{paper.doi}"
        return paper.semantic_scholar_id

    def _match_reference_to_local(self, ref: dict[str, Any], papers: dict[str, PaperExtraction]) -> PaperExtraction | None:
        ref_doi = str(ref.get("doi") or "").lower().strip()
        ref_arxiv = str(ref.get("arxiv_id") or ref.get("arxivId") or "").lower().replace("arxiv:", "").strip()
        ref_title = self._norm(str(ref.get("title") or ""))
        for paper in papers.values():
            if ref_doi and paper.doi and ref_doi == paper.doi.lower():
                return paper
            if ref_arxiv and paper.arxiv_id and ref_arxiv.split("v", 1)[0] == paper.arxiv_id.lower().split("v", 1)[0]:
                return paper
            if ref_title and self._norm(paper.title) == ref_title:
                return paper
        return None

    @classmethod
    def _local_reference_evidence(cls, source: PaperExtraction, target: PaperExtraction) -> str:
        target_titles = {cls._norm(target.title)}
        if target.filename:
            target_titles.add(cls._norm(target.filename.rsplit(".", 1)[0]))
        reference_ids = source.evidence.get("reference_ids") or []
        for index, ref in enumerate(source.references):
            norm_ref = cls._norm(ref)
            evidence_id = reference_ids[index] if index < len(reference_ids) else ""
            prefix = f"[{evidence_id}] " if evidence_id else ""
            if target.doi and target.doi.lower() in ref.lower():
                return prefix + ref[:300]
            if target.arxiv_id and target.arxiv_id.lower().split("v", 1)[0] in ref.lower():
                return prefix + ref[:300]
            if any(title and title in norm_ref for title in target_titles):
                return prefix + ref[:300]
        return ""

    async def _fetch_semantic_scholar_references(self, paper: PaperExtraction) -> dict[str, Any]:
        identifier = self._paper_identifier(paper)
        if not identifier:
            return {"references": [], "error": "No verifiable identifier."}
        url = f"https://api.semanticscholar.org/graph/v1/paper/{identifier}/references"
        params = {
            "fields": "paperId,title,year,abstract,authors,externalIds,openAccessPdf,citationCount,venue,publicationVenue,url",
            "limit": "1000",
        }
        headers = {}
        api_key = os.environ.get("S2_API_KEY")
        if api_key:
            headers["x-api-key"] = api_key
        try:
            import httpx

            for attempt in range(3):
                async with httpx.AsyncClient(timeout=12.0) as client:
                    resp = await client.get(url, params=params, headers=headers)
                if resp.status_code == 429 and attempt < 2:
                    await asyncio.sleep(1.5 * (attempt + 1))
                    continue
                if resp.status_code == 404:
                    return {"references": [], "error": "Paper not present on Semantic Scholar."}
                if resp.status_code >= 400:
                    return {"references": [], "error": f"Semantic Scholar returned HTTP {resp.status_code}."}
                data = resp.json()
                refs = []
                for row in data.get("data", []):
                    cited = row.get("citedPaper") or {}
                    title = cited.get("title")
                    if not title:
                        continue
                    refs.append(cited)
                return {"references": refs, "error": None}
        except Exception as exc:
            return {"references": [], "error": f"Semantic Scholar lookup failed: {exc}"}
        return {"references": [], "error": "Semantic Scholar lookup failed."}

    @staticmethod
    def _merge_node(nodes: dict[str, CitationGraphNode], node: CitationGraphNode) -> None:
        existing = nodes.get(node.id)
        if existing is None:
            nodes[node.id] = node
            return
        existing.source_file_ids = sorted(set(existing.source_file_ids + node.source_file_ids))
        for file_id, snippets in node.evidence_by_file.items():
            existing.evidence_by_file.setdefault(file_id, [])
            existing.evidence_by_file[file_id] = sorted(set(existing.evidence_by_file[file_id] + snippets))
        existing.confidence = max(existing.confidence, node.confidence)

    def _add_edge(
        self,
        edges: dict[str, CitationGraphEdge],
        source: str,
        target: str,
        relation: CitationRelation,
        evidence: str,
        source_file_id: str,
        confidence: float,
    ) -> None:
        if not source or source == target:
            return
        edge_id = f"{source}->{relation}->{target}"
        edges[edge_id] = CitationGraphEdge(
            id=edge_id,
            source=source,
            target=target,
            relation=relation,
            evidence=evidence[:300],
            source_file_id=source_file_id,
            confidence=confidence,
        )