Skills › Scientific Agent Skills › paper-lookup › openalex_abstract.py
openalex_abstract.py
169 lines · skills/paper-lookup/scripts/openalex_abstract.py at 5f2bc03
1#!/usr/bin/env python3
2"""Reconstruct OpenAlex abstracts from `abstract_inverted_index`.
4OpenAlex never returns an abstract as a string. It returns
5`{"word": [positions], ...}`, and the caller has to invert it. The naive
6inversion loses words: building `{position: word}` and joining silently drops
7every duplicate position, and real payloads do contain them.
9Reads a single work object, a list of works, or a `/works` list response
10(`{"meta": ..., "results": [...]}`). Emits each work's id, doi, title, and
11reconstructed abstract, plus a per-work note when the abstract could not be
12rebuilt.
14 curl -s "https://api.openalex.org/works/doi:10.7717/peerj.4375" | python3 openalex_abstract.py -
15 python3 openalex_abstract.py results.json --text-only
16"""
18from __future__ import annotations
20import argparse
21import sys
22from pathlib import Path
23from typing import Any
25sys.path.insert(0, str(Path(__file__).resolve().parent))
27from _common import InputError, collapse_ws, emit, fail, load_json # noqa: E402
30def reconstruct(inverted_index: dict[str, list[int]]) -> tuple[str, list[str]]:
31 """Rebuild abstract text from an inverted index.
33 Returns the text and any anomalies worth reporting. Words are bucketed by
34 position rather than assigned, so a position claimed by two words keeps both
35 (joined in index order) instead of one overwriting the other.
36 """
37 anomalies: list[str] = []
38 buckets: dict[int, list[str]] = {}
40 for word, positions in inverted_index.items():
41 if not isinstance(positions, list):
42 anomalies.append(f"positions for {word!r} were {type(positions).__name__}, not a list")
43 continue
44 for position in positions:
45 if not isinstance(position, int) or isinstance(position, bool):
46 anomalies.append(f"non-integer position {position!r} for {word!r}")
47 continue
48 if position < 0:
49 anomalies.append(f"negative position {position} for {word!r}")
50 continue
51 buckets.setdefault(position, []).append(word)
53 if not buckets:
54 return "", anomalies
56 collisions = sum(1 for words in buckets.values() if len(words) > 1)
57 if collisions:
58 anomalies.append(
59 f"{collisions} position(s) claimed by more than one token; kept all, joined in index order"
60 )
62 ordered = sorted(buckets)
63 missing = ordered[-1] - ordered[0] + 1 - len(ordered)
64 if missing:
65 anomalies.append(f"{missing} position(s) absent from the index; the abstract has gaps")
66 if ordered[0] != 0:
67 anomalies.append(f"index starts at position {ordered[0]}, not 0; leading words may be missing")
69 text = " ".join(" ".join(buckets[position]) for position in ordered)
70 return collapse_ws(text), anomalies
73def works_from(payload: Any) -> list[dict[str, Any]]:
74 """Accept a single work, a bare list, or a `/works` list response."""
75 if isinstance(payload, dict):
76 if "error" in payload or ("message" in payload and "id" not in payload and "results" not in payload):
77 raise InputError("OpenAlex returned an error object, not a work")
78 if isinstance(payload.get("results"), list):
79 return [w for w in payload["results"] if isinstance(w, dict)]
80 if not payload.get("id"):
81 raise InputError("OpenAlex object lacks a work id/results array")
82 return [payload]
83 if isinstance(payload, list):
84 return [w for w in payload if isinstance(w, dict)]
85 raise InputError(f"expected a work object or list, got {type(payload).__name__}")
88def summarize(work: dict[str, Any]) -> dict[str, Any]:
89 record: dict[str, Any] = {
90 "id": work.get("id"),
91 "doi": work.get("doi"),
92 "title": collapse_ws(work.get("title") or work.get("display_name")),
93 "publication_year": work.get("publication_year"),
94 }
96 index = work.get("abstract_inverted_index")
97 if isinstance(index, dict) and index:
98 text, anomalies = reconstruct(index)
99 record["abstract"] = text
100 record["abstract_word_count"] = len(text.split()) if text else 0
101 if anomalies:
102 record["abstract_warnings"] = anomalies
103 else:
104 record["abstract"] = None
105 # Distinguish the two reasons an abstract is missing: the field was not
106 # requested, or OpenAlex has none. Reporting them the same way would let
107 # a `select=` mistake read as a coverage gap.
108 record["abstract_warnings"] = [
109 "no abstract_inverted_index on this work: either OpenAlex has no abstract for it, "
110 "or the field was excluded by `select=` -- re-request without `select` to tell which"
111 ]
112 return record
115def build_parser() -> argparse.ArgumentParser:
116 parser = argparse.ArgumentParser(
117 description=(
118 "Reconstruct readable abstracts from OpenAlex abstract_inverted_index payloads."
119 ),
120 epilog='curl -s "https://api.openalex.org/works/doi:10.7717/peerj.4375" | %(prog)s -',
121 )
122 parser.add_argument("source", help="path to an OpenAlex JSON payload, or - for stdin")
123 parser.add_argument("-o", "--output", help="write JSON here instead of stdout")
124 parser.add_argument(
125 "--text-only",
126 action="store_true",
127 help="print just the abstract text, one blank-line-separated block per work",
128 )
129 return parser
132def main(argv: list[str] | None = None) -> int:
133 args = build_parser().parse_args(argv)
135 try:
136 payload = load_json(args.source)
137 works = works_from(payload)
138 except InputError as error:
139 fail(str(error))
141 if not works:
142 fail("payload contained no work objects")
144 records = [summarize(work) for work in works]
146 if args.text_only:
147 blocks = [record["abstract"] for record in records if record["abstract"]]
148 if not blocks:
149 fail("no abstracts could be reconstructed from this payload")
150 text = "\n\n".join(blocks) + "\n"
151 if args.output:
152 Path(args.output).write_text(text, encoding="utf-8")
153 else:
154 sys.stdout.write(text)
155 return 0
157 emit(
158 {
159 "count": len(records),
160 "with_abstract": sum(1 for r in records if r["abstract"]),
161 "works": records,
162 },
163 args.output,
164 )
165 return 0
168if __name__ == "__main__":
169 raise SystemExit(main())