Skills › Scientific Agent Skills › citation-management › search_openalex.py
search_openalex.py
325 lines · skills/citation-management/scripts/search_openalex.py at 5f2bc03
1#!/usr/bin/env python3
2"""
3OpenAlex Search Tool
4Search OpenAlex and export results as JSON or BibTeX.
6OpenAlex provides a documented REST API. Casual keyless requests are supported;
7set OPENALEX_API_KEY for an account budget. The key is sent only in the OpenAlex
8Authorization header. OPENALEX_EMAIL / --email remains an optional contact
9identifier; it does not establish a higher API quota.
11"""
13from __future__ import annotations
15import argparse
16import json
17import os
18import sys
19import time
20from typing import Dict, List, Optional
22import requests
24sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))
26from _common import ( # noqa: E402
27 citation_key,
28 format_pages,
29 protect_title,
30 render_entry,
31)
33API_URL = 'https://api.openalex.org/works'
35# OpenAlex work types -> BibTeX entry types.
36TYPE_MAP = {
37 'article': 'article',
38 'journal-article': 'article',
39 'review': 'article',
40 'book': 'book',
41 'book-chapter': 'incollection',
42 'monograph': 'book',
43 'proceedings-article': 'inproceedings',
44 'proceedings': 'inproceedings',
45 'dissertation': 'phdthesis',
46 'report': 'techreport',
47 'preprint': 'misc',
48 'posted-content': 'misc',
49 'dataset': 'misc',
50}
53class OpenAlexSearcher:
54 """Search OpenAlex via its public REST API."""
56 def __init__(self, email: Optional[str] = None):
57 self.email = email or os.getenv('OPENALEX_EMAIL', '')
58 api_key = os.getenv('OPENALEX_API_KEY', '')
59 self.session = requests.Session()
60 self.session.headers.update({
61 'User-Agent': 'OpenAlexSearcher/1.0 (Citation Management Tool)'
62 })
63 if api_key:
64 self.session.headers['Authorization'] = f'Bearer {api_key}'
66 def search(self, query: str, max_results: int = 50,
67 year_start: Optional[int] = None, year_end: Optional[int] = None,
68 work_type: Optional[str] = None,
69 sort_by: str = 'relevance') -> List[Dict]:
70 """
71 Search OpenAlex and return normalised metadata records.
73 Args:
74 query: Free-text search query
75 max_results: Maximum number of works to return
76 year_start: Earliest publication year, inclusive
77 year_end: Latest publication year, inclusive
78 work_type: OpenAlex work type filter (e.g. 'article', 'review')
79 sort_by: 'relevance' or 'citations'
81 Returns:
82 List of metadata dictionaries
83 """
84 if max_results < 1:
85 raise ValueError("max_results must be positive")
86 filters = []
87 if year_start:
88 filters.append(f'from_publication_date:{year_start}-01-01')
89 if year_end:
90 filters.append(f'to_publication_date:{year_end}-12-31')
91 if work_type:
92 filters.append(f'type:{work_type}')
94 params = {
95 'search': query,
96 'per-page': str(min(100, max_results)),
97 'cursor': '*',
98 }
99 if filters:
100 params['filter'] = ','.join(filters)
101 if sort_by == 'citations':
102 params['sort'] = 'cited_by_count:desc'
103 if self.email:
104 params['mailto'] = self.email
106 print(f'Searching OpenAlex: {query}', file=sys.stderr)
108 results: List[Dict] = []
109 seen_cursors = {'*'}
110 seen_work_ids = set()
111 reported_total = False
113 while len(results) < max_results:
114 try:
115 response = self.session.get(API_URL, params=params, timeout=30)
116 response.raise_for_status()
117 payload = response.json()
118 except (requests.exceptions.RequestException, ValueError) as error:
119 # Partial bibliographies must not look like a completed search.
120 # Avoid printing request URLs or headers in diagnostics.
121 raise RuntimeError(
122 f'OpenAlex search failed after {len(results)} records '
123 f'({type(error).__name__}); retry or narrow the query'
124 ) from None
126 if not reported_total:
127 total = payload.get('meta', {}).get('count', 0)
128 print(f'Found {total} results, retrieving up to {max_results}', file=sys.stderr)
129 reported_total = True
131 works = payload.get('results', [])
132 next_cursor = payload.get('meta', {}).get('next_cursor')
133 if next_cursor is not None and (
134 not isinstance(next_cursor, str) or next_cursor in seen_cursors
135 ):
136 raise RuntimeError(
137 f'OpenAlex search failed after {len(results)} records '
138 '(invalid or repeated cursor); retry or narrow the query'
139 )
140 if not works:
141 if next_cursor is not None:
142 raise RuntimeError(
143 f'OpenAlex search failed after {len(results)} records '
144 '(empty nonterminal page); retry or narrow the query'
145 )
146 break
147 for work in works:
148 work_id = work.get('id')
149 if not isinstance(work_id, str) or not work_id or work_id in seen_work_ids:
150 raise RuntimeError(
151 f'OpenAlex search failed after {len(results)} records '
152 '(missing or duplicate work ID); retry or narrow the query'
153 )
154 seen_work_ids.add(work_id)
155 results.append(self._normalise(work))
156 if len(results) >= max_results:
157 break
159 if next_cursor is None or len(results) >= max_results:
160 break
161 seen_cursors.add(next_cursor)
162 params['cursor'] = next_cursor
164 # Pace requests; quota exhaustion remains an explicit error.
165 time.sleep(0.1)
167 print(f'Retrieved {len(results)} results', file=sys.stderr)
168 return results
170 def _normalise(self, work: Dict) -> Dict:
171 """Flatten one OpenAlex work into this skill's metadata shape."""
172 authors = [
173 authorship.get('author', {}).get('display_name', '')
174 for authorship in work.get('authorships', [])
175 ]
176 authors = [name for name in authors if name]
178 location = work.get('primary_location') or {}
179 source = location.get('source') or {}
180 biblio = work.get('biblio') or {}
182 first_page = biblio.get('first_page') or ''
183 last_page = biblio.get('last_page') or ''
184 if first_page and last_page and first_page != last_page:
185 pages = f'{first_page}-{last_page}'
186 else:
187 pages = first_page or ''
189 doi = work.get('doi') or ''
190 if doi.startswith('https://doi.org/'):
191 doi = doi[len('https://doi.org/'):]
193 return {
194 'openalex_id': work.get('id', ''),
195 'doi': doi,
196 'title': work.get('title') or work.get('display_name') or '',
197 'authors': ' and '.join(authors),
198 'journal': source.get('display_name', '') or '',
199 'publisher': source.get('host_organization_name', '') or '',
200 'year': str(work.get('publication_year') or ''),
201 'volume': biblio.get('volume') or '',
202 'issue': biblio.get('issue') or '',
203 'pages': pages,
204 'type': work.get('type') or '',
205 'citations': work.get('cited_by_count', 0),
206 'is_open_access': (work.get('open_access') or {}).get('is_oa', False),
207 'abstract': self._abstract(work.get('abstract_inverted_index')),
208 }
210 @staticmethod
211 def _abstract(inverted_index: Optional[Dict]) -> str:
212 """Rebuild the abstract from OpenAlex's inverted index."""
213 if not inverted_index:
214 return ''
215 positions = []
216 for word, indices in inverted_index.items():
217 for index in indices:
218 positions.append((index, word))
219 positions.sort()
220 return ' '.join(word for _, word in positions)
222 def metadata_to_bibtex(self, metadata: Dict) -> str:
223 """Convert a normalised record to a BibTeX entry."""
224 entry_type = TYPE_MAP.get(metadata.get('type', ''), 'misc')
226 key = citation_key(
227 metadata.get('authors', ''),
228 metadata.get('year', ''),
229 metadata.get('title', ''),
230 )
232 fields = {
233 'author': metadata.get('authors', ''),
234 'title': protect_title(metadata.get('title', '')),
235 'year': metadata.get('year', ''),
236 'volume': metadata.get('volume', ''),
237 'number': metadata.get('issue', ''),
238 'pages': format_pages(metadata.get('pages')),
239 'doi': metadata.get('doi', ''),
240 }
242 venue = metadata.get('journal', '')
243 if entry_type in ('inproceedings', 'incollection'):
244 fields['booktitle'] = venue
245 elif entry_type != 'misc':
246 fields['journal'] = venue
248 if entry_type in ('book', 'incollection', 'inproceedings'):
249 fields['publisher'] = metadata.get('publisher', '')
250 elif entry_type == 'techreport':
251 fields['institution'] = metadata.get('publisher', '')
253 if entry_type == 'misc':
254 if venue:
255 fields['howpublished'] = venue
256 if metadata.get('type') in ('preprint', 'posted-content'):
257 fields['note'] = 'Preprint'
259 if not fields['doi'] and metadata.get('openalex_id'):
260 fields['url'] = metadata['openalex_id']
262 return render_entry(entry_type, key, fields)
265def main():
266 """Command-line interface."""
267 parser = argparse.ArgumentParser(
268 description='Search OpenAlex (optional OPENALEX_API_KEY for account quota)',
269 epilog='Example: python search_openalex.py "CRISPR gene editing" --limit 50 --format bibtex'
270 )
272 parser.add_argument('query', help='Search query')
273 parser.add_argument('--limit', type=int, default=50,
274 help='Maximum number of results (default: 50)')
275 parser.add_argument('--year-start', type=int, help='Earliest publication year')
276 parser.add_argument('--year-end', type=int, help='Latest publication year')
277 parser.add_argument('--type', dest='work_type',
278 help='OpenAlex work type filter (e.g. article, review, preprint)')
279 parser.add_argument('--sort-by', choices=['relevance', 'citations'], default='relevance',
280 help='Sort order (default: relevance)')
281 parser.add_argument('-o', '--output', help='Output file (default: stdout)')
282 parser.add_argument('--format', choices=['json', 'bibtex'], default='json',
283 help='Output format (default: json)')
284 parser.add_argument('--email',
285 help='Optional contact email (or set OPENALEX_EMAIL)')
287 args = parser.parse_args()
289 searcher = OpenAlexSearcher(email=args.email)
290 try:
291 results = searcher.search(
292 args.query,
293 max_results=args.limit,
294 year_start=args.year_start,
295 year_end=args.year_end,
296 work_type=args.work_type,
297 sort_by=args.sort_by,
298 )
299 except (RuntimeError, ValueError) as error:
300 print(f'Error: {error}', file=sys.stderr)
301 sys.exit(2)
303 if not results:
304 print('No results found', file=sys.stderr)
305 sys.exit(1)
307 if args.format == 'json':
308 output = json.dumps({
309 'query': args.query,
310 'count': len(results),
311 'results': results,
312 }, indent=2)
313 else:
314 output = '\n\n'.join(searcher.metadata_to_bibtex(r) for r in results) + '\n'
316 if args.output:
317 with open(args.output, 'w', encoding='utf-8') as f:
318 f.write(output)
319 print(f'Wrote {len(results)} results to {args.output}', file=sys.stderr)
320 else:
321 print(output)
324if __name__ == '__main__':
325 main()