study-buddy / app /services /citation_graph.py
GitHub Actions
deploy d092bea3608b7a29952f16357fda39b7a29e399b
2e818da
Raw
History Blame Contribute Delete
41.9 kB
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,
)