Spaces:
Sleeping
Sleeping
| """Within-corpus PageRank over the index's own citation graph. | |
| ``referenced_works`` entries are bare work IDs that map directly to document | |
| ``_id`` values, so the graph is whatever the index already holds: an edge | |
| i -> j exists when indexed work i cites indexed work j. References outside | |
| the corpus are dropped, which means the score is *within-corpus* authority — | |
| how the indexed literature cites a work — not a global citation rank. | |
| The pass stores ``pagerank`` as ``log10(pr * n + 1)`` (always positive, as the | |
| rank_feature mapping requires) and also backfills ``authority`` as | |
| ``log1p(cited_by_count)`` so neither signal needs a full re-ingest. | |
| """ | |
| import math | |
| from collections.abc import Iterable | |
| import numpy as np | |
| import scipy.sparse | |
| from elasticsearch.helpers import async_bulk, async_scan | |
| def pagerank( | |
| ids: list[str], | |
| edges: Iterable[tuple[int, int]], | |
| *, | |
| damping: float = 0.85, | |
| iterations: int = 30, | |
| ) -> np.ndarray: | |
| """Power-iterate PageRank; returns a probability vector aligned with ``ids``. | |
| ``edges`` are ``(source, target)`` index pairs into ``ids``. Duplicate and | |
| self edges collapse; dangling nodes (no out-edges) redistribute their mass | |
| uniformly, so scores always sum to ~1. | |
| """ | |
| if not 0 < damping < 1: | |
| raise ValueError("damping must be in (0, 1)") | |
| if iterations < 0: | |
| raise ValueError("iterations must be non-negative") | |
| n = len(ids) | |
| scores = np.full(n, 1.0 / n) if n else np.zeros(0) | |
| if n == 0: | |
| return scores | |
| pairs = {(i, j) for i, j in edges if 0 <= i < n and 0 <= j < n and i != j} | |
| rows = [i for i, _ in pairs] | |
| cols = [j for _, j in pairs] | |
| adjacency = scipy.sparse.csr_matrix( | |
| (np.ones(len(pairs)), (rows, cols)), shape=(n, n), dtype=np.float64 | |
| ) | |
| out_degree = np.asarray(adjacency.sum(axis=1)).ravel() | |
| linked = out_degree > 0 | |
| inv_out = np.zeros(n) | |
| inv_out[linked] = 1.0 / out_degree[linked] | |
| transition = scipy.sparse.diags(inv_out) @ adjacency | |
| teleport = (1.0 - damping) / n | |
| for _ in range(iterations): | |
| dangling_share = damping * scores[~linked].sum() / n | |
| scores = damping * (transition.T @ scores) + teleport + dangling_share | |
| return scores | |
| async def update_authority( | |
| repository, | |
| *, | |
| damping: float = 0.85, | |
| iterations: int = 30, | |
| batch_size: int = 500, | |
| ) -> dict: | |
| """Score the indexed citation graph and bulk-update both rank features. | |
| One scroll reads ``record_kind=study`` IDs, references and citation counts; | |
| one bulk pass writes the signals back. Edges to works outside the index are | |
| dropped, so the graph is exactly the stored corpus. | |
| """ | |
| ids: list[str] = [] | |
| references: list[list[str]] = [] | |
| citations: list = [] | |
| async for hit in async_scan( | |
| repository.client, | |
| index=repository.index, | |
| query={ | |
| "query": {"term": {"record_kind": "study"}}, | |
| "_source": ["referenced_works", "cited_by_count"], | |
| }, | |
| size=batch_size, | |
| ): | |
| ids.append(hit["_id"]) | |
| source = hit.get("_source") or {} | |
| references.append( | |
| [ | |
| str(ref).rsplit("/", 1)[-1] | |
| for ref in source.get("referenced_works") or [] | |
| ] | |
| ) | |
| citations.append(source.get("cited_by_count")) | |
| position = {identifier: i for i, identifier in enumerate(ids)} | |
| edges = { | |
| (i, position[ref]) | |
| for i, refs in enumerate(references) | |
| for ref in refs | |
| if ref in position | |
| } | |
| scores = pagerank(ids, sorted(edges), damping=damping, iterations=iterations) | |
| n = len(ids) | |
| actions = [] | |
| for i, identifier in enumerate(ids): | |
| patch = {"pagerank": math.log10(scores[i] * n + 1.0)} | |
| count = citations[i] | |
| if isinstance(count, (int, float)) and not isinstance(count, bool) and count > 0: | |
| patch["authority"] = math.log1p(count) | |
| actions.append( | |
| {"_op_type": "update", "_index": repository.index, "_id": identifier, "doc": patch} | |
| ) | |
| updates = 0 | |
| if actions: | |
| updates, _ = await async_bulk( | |
| repository.client, actions, refresh="wait_for", chunk_size=batch_size | |
| ) | |
| top = np.argsort(-scores)[:5] if n else [] | |
| return { | |
| "nodes": n, | |
| "edges": len(edges), | |
| "updates": int(updates), | |
| "top_pagerank": [{"id": ids[i], "pagerank": float(scores[i])} for i in top], | |
| } | |