Spaces:
Sleeping
Sleeping
File size: 14,382 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 | from __future__ import annotations
import hashlib
import time
from collections.abc import Mapping
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
from app.services.citation_exploration_store import CachedNeighborhood, CitationExplorationStore
from app.services.citation_graph import CitationGraph, CitationGraphEdge, CitationGraphNode, CitationGraphService
from app.services.citation_topology import CitationTopologyService, VerifiedTopologyEdge
from app.services.paper_acquisition import PaperAcquisitionResult, PaperAcquisitionService, sanitize_acquisition_candidate
from app.services.paper_metadata import PaperMetadataService
class CitationExplorationResponse(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
graph: CitationGraph
seed_id: str
seed_candidate_id: str
neighborhood_key: str
exploration_kind: Literal["paper", "author"] = "paper"
primary_state: Literal["complete", "degraded"] = "complete"
induced_state: Literal["pending", "complete", "degraded"] = "pending"
class CitationExplorationDelta(BaseModel):
model_config = ConfigDict(extra="forbid", strict=True)
neighborhood_key: str
nodes: list[CitationGraphNode] = Field(default_factory=list)
edges: list[CitationGraphEdge] = Field(default_factory=list)
induced_state: Literal["complete", "degraded"]
warnings: list[str] = Field(default_factory=list)
class CitationExplorationService:
"""Cached transient paper and author graph orchestration."""
def __init__(
self,
root: str | None = None,
graph_service: CitationGraphService | None = None,
topology_service: CitationTopologyService | None = None,
metadata_service: PaperMetadataService | None = None,
acquisition_service: PaperAcquisitionService | None = None,
) -> None:
self.root = root
self.graph_service = graph_service or CitationGraphService()
self.metadata_service = metadata_service or PaperMetadataService()
self.topology = topology_service or CitationTopologyService(self.metadata_service)
self.acquisition_service = acquisition_service or PaperAcquisitionService()
def _store(self, project_id: str) -> CitationExplorationStore:
return CitationExplorationStore(project_id, self.root)
def _project_candidate(self, project_id: str, candidate_id: str) -> dict[str, Any] | None:
graph = self.graph_service.load(project_id)
for node in graph.nodes:
if node.status == "owned" and node.metadata.get("external_candidate_id") == candidate_id:
return self.metadata_service.from_candidate({
**node.metadata,
"candidate_id": candidate_id,
"title": node.title,
"authors": node.authors,
"year": node.year,
"abstract": node.abstract,
"url": node.url,
"doi": node.doi,
"arxiv_id": node.arxiv_id,
"semantic_scholar_id": node.semantic_scholar_id,
"cited_by": node.cited_by,
}).candidate_dict()
references = node.metadata.get("verified_unmatched_references") or []
for reference in references:
if isinstance(reference, dict) and reference.get("candidate_id") == candidate_id:
return sanitize_acquisition_candidate(reference)
return None
def get_candidate(self, project_id: str, candidate_id: str) -> dict[str, Any] | None:
store = self._store(project_id)
cached = store.candidate(candidate_id)
if cached:
return cached
project_candidate = self._project_candidate(project_id, candidate_id)
if project_candidate:
store.upsert_papers([project_candidate])
return project_candidate
@staticmethod
def _node(candidate: Mapping[str, Any], *, seed: bool = False) -> CitationGraphNode:
candidate_id = str(candidate.get("candidate_id") or "")
title = str(candidate.get("title") or "Untitled paper")
author_details = candidate.get("author_details") if isinstance(candidate.get("author_details"), list) else []
authors = str(candidate.get("authors") or ", ".join(
str(row.get("name") or "") for row in author_details if isinstance(row, Mapping)
))
year_value = candidate.get("year")
try:
year = int(year_value) if year_value is not None else None
except (TypeError, ValueError):
year = None
first_author = authors.split(",", 1)[0].split(" and ", 1)[0].strip()
label = f"{first_author or 'Paper'}, {year}" if year else first_author or "Paper"
metadata = dict(candidate)
metadata["external_candidate_id"] = candidate_id
metadata["exploration_seed"] = seed
return CitationGraphNode(
id=f"external:{candidate_id}",
type="paper",
label=label,
title=title,
abstract=str(candidate.get("abstract") or ""),
url=str(candidate.get("url") or candidate.get("canonical_url") or ""),
year=year,
authors=authors,
cited_by=max(0, int(candidate.get("cited_by") or 0)),
status="connector",
source_file_ids=[],
confidence=float(candidate.get("resolution_confidence") or 0.8),
arxiv_id=str(candidate.get("arxiv_id") or ""),
doi=str(candidate.get("doi") or ""),
semantic_scholar_id=str(candidate.get("semantic_scholar_id") or ""),
metadata=metadata,
)
@staticmethod
def _edge(edge: VerifiedTopologyEdge, *, induced: bool) -> CitationGraphEdge:
evidence = f"Verified by {edge.provider} work identifiers"
digest = hashlib.sha256(
f"{edge.provider}|{edge.source_candidate_id}|{edge.target_candidate_id}".encode("utf-8")
).hexdigest()[:20]
return CitationGraphEdge(
id=f"explore:{digest}",
source=f"external:{edge.source_candidate_id}",
target=f"external:{edge.target_candidate_id}",
relation="cites",
evidence=f"{evidence}{' (induced)' if induced else ''}",
confidence=1.0,
)
def _graph_from_neighborhood(
self, project_id: str, store: CitationExplorationStore, neighborhood: CachedNeighborhood
) -> CitationGraph:
cache = store.load()
nodes = [
self._node(cache.papers[candidate_id], seed=candidate_id == neighborhood.seed_candidate_id)
for candidate_id in neighborhood.paper_ids
if candidate_id in cache.papers
]
edges = [CitationGraphEdge(**row) for row in [*neighborhood.primary_edges, *neighborhood.induced_edges]]
return CitationGraph(
project_id=project_id,
nodes=nodes,
edges=edges,
updated_at=neighborhood.updated_at,
warnings=neighborhood.warnings,
)
def _response(
self, project_id: str, store: CitationExplorationStore, neighborhood: CachedNeighborhood
) -> CitationExplorationResponse:
graph = self._graph_from_neighborhood(project_id, store, neighborhood)
if neighborhood.seed_candidate_id:
seed_candidate_id = neighborhood.seed_candidate_id
elif graph.nodes:
seed_candidate_id = str(graph.nodes[0].metadata.get("external_candidate_id") or "")
else:
seed_candidate_id = ""
return CitationExplorationResponse(
graph=graph,
seed_id=f"external:{seed_candidate_id}" if seed_candidate_id else "",
seed_candidate_id=seed_candidate_id,
neighborhood_key=neighborhood.key,
exploration_kind=neighborhood.exploration_kind,
primary_state=neighborhood.primary_state,
induced_state=neighborhood.induced_state,
)
async def explore(self, project_id: str, candidate_id: str) -> CitationExplorationResponse:
seed = self.get_candidate(project_id, candidate_id)
if seed is None:
raise LookupError("Verified citation candidate not found.")
store = self._store(project_id)
key = f"paper:{candidate_id}"
cached = store.neighborhood(key)
if cached:
return self._response(project_id, store, cached)
topology = await self.topology.paper_neighborhood(seed)
store.upsert_papers(topology.papers)
store.save_reference_sets(topology.reference_sets)
primary_edges = [self._edge(edge, induced=False).model_dump() for edge in topology.primary_edges]
resolved_seed_id = str(topology.seed.get("candidate_id") or candidate_id)
neighborhood = store.new_neighborhood(
key=key,
exploration_kind="paper",
seed_candidate_id=resolved_seed_id,
paper_ids=[str(paper.get("candidate_id") or "") for paper in topology.papers if paper.get("candidate_id")],
primary_edges=primary_edges,
warnings=topology.warnings,
degraded=bool(topology.warnings),
)
store.save_neighborhood(neighborhood)
return self._response(project_id, store, neighborhood)
async def enrich_edges(self, project_id: str, neighborhood_key: str) -> CitationExplorationDelta:
store = self._store(project_id)
neighborhood = store.neighborhood(neighborhood_key, fresh_only=False)
if neighborhood is None:
raise LookupError("Exploration neighborhood not found.")
if neighborhood.induced_state in {"complete", "degraded"} and neighborhood.fresh:
return CitationExplorationDelta(
neighborhood_key=neighborhood_key,
edges=[CitationGraphEdge(**row) for row in neighborhood.induced_edges],
induced_state=neighborhood.induced_state,
warnings=neighborhood.warnings,
)
cache = store.load()
papers = [cache.papers[candidate_id] for candidate_id in neighborhood.paper_ids if candidate_id in cache.papers]
reference_sets = {
candidate_id: cache.reference_sets[candidate_id]
for candidate_id in neighborhood.paper_ids
if candidate_id in cache.reference_sets and cache.reference_sets[candidate_id].fresh
}
warnings = [] if len(reference_sets) == len(papers) else ["Some paper reference sets were unavailable."]
induced = self.topology.induced_edges(papers, reference_sets)
primary_pairs = {
(str(row.get("source") or ""), str(row.get("target") or ""))
for row in neighborhood.primary_edges
}
edge_rows = [
self._edge(edge, induced=True).model_dump()
for edge in induced
if (f"external:{edge.source_candidate_id}", f"external:{edge.target_candidate_id}") not in primary_pairs
]
merged = store.merge_induced_edges(neighborhood_key, edge_rows, warnings=warnings)
return CitationExplorationDelta(
neighborhood_key=neighborhood_key,
edges=[CitationGraphEdge(**row) for row in merged.induced_edges],
induced_state=merged.induced_state,
warnings=merged.warnings,
)
async def details(self, project_id: str, candidate_id: str) -> CitationGraphNode:
candidate = self.get_candidate(project_id, candidate_id)
if candidate is None:
raise LookupError("Verified citation candidate not found.")
resolved = await self.metadata_service.resolve(candidate)
payload = resolved.candidate_dict()
self._store(project_id).upsert_papers([payload])
return self._node(payload, seed=bool(candidate.get("exploration_seed")))
async def explore_author(
self, project_id: str, candidate_id: str, author_index: int
) -> CitationExplorationResponse:
candidate = self.get_candidate(project_id, candidate_id)
if candidate is None:
raise LookupError("Verified citation candidate not found.")
# Detail resolution supplies provider author identifiers before accepting the index.
resolved = await self.metadata_service.resolve(candidate)
resolved_candidate = resolved.candidate_dict()
store = self._store(project_id)
store.upsert_papers([resolved_candidate])
author_details = resolved_candidate.get("author_details") or []
if author_index < 0 or author_index >= len(author_details):
raise LookupError("Author is not available on this paper.")
author = author_details[author_index]
openalex_id = str(author.get("openalex_id") or "") if isinstance(author, Mapping) else ""
cached_key = f"author:openalex:{self.topology._work_id(openalex_id)}:top10" if openalex_id else ""
if cached_key:
cached = store.neighborhood(cached_key)
if cached:
return self._response(project_id, store, cached)
key, papers, reference_sets = await self.topology.author_neighborhood(resolved_candidate, author_index)
store.upsert_papers(papers)
store.save_reference_sets(reference_sets)
seed_candidate_id = str(papers[0].get("candidate_id") or "") if papers else ""
neighborhood = store.new_neighborhood(
key=key,
exploration_kind="author",
author_key=key,
seed_candidate_id=seed_candidate_id,
paper_ids=[str(paper.get("candidate_id") or "") for paper in papers if paper.get("candidate_id")],
primary_edges=[],
warnings=[] if papers else ["No papers were returned for this author."],
degraded=not papers,
)
store.save_neighborhood(neighborhood)
return self._response(project_id, store, neighborhood)
async def download(self, project_id: str, candidate_id: str) -> PaperAcquisitionResult:
candidate = self.get_candidate(project_id, candidate_id)
if candidate is None:
raise LookupError("Verified citation candidate not found.")
return await self.acquisition_service.acquire(project_id, candidate)
|