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)