Step 2: search InspireHEP for replacements for flagged entries.
Loads the results.json produced by Step 1 and, for each missing or
mismatched entry, queries InspireHEP using eprint, DOI, or
author+year as fallback strategies.
suggest_replacements
suggest_replacements(
results_path, client=None, verbose=False
)
Load Step 1 output and search for replacement candidates.
Only entries with status "missing" or "mismatch" are processed.
| Parameters: |
-
results_path
(str or Path)
–
Path to the results.json file written by Step 1.
-
client
(InspireClient, default:
None
)
–
API client to use. A default client is created when None.
-
verbose
(bool, default:
False
)
–
Print progress to stdout for each entry. Default is False.
|
| Returns: |
-
suggestions( list[Suggestion]
) –
Flat list of candidates, up to 5 per actionable entry.
|
Source code in src/bib_checker/searcher.py
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 | def suggest_replacements(
results_path: str | Path,
client: InspireClient | None = None,
verbose: bool = False,
) -> list[Suggestion]:
"""Load Step 1 output and search for replacement candidates.
Only entries with status ``"missing"`` or ``"mismatch"`` are processed.
Parameters
----------
results_path : str or Path
Path to the results.json file written by Step 1.
client : InspireClient, optional
API client to use. A default client is created when ``None``.
verbose : bool, optional
Print progress to stdout for each entry. Default is ``False``.
Returns
-------
suggestions : list[Suggestion]
Flat list of candidates, up to 5 per actionable entry.
"""
if client is None:
client = InspireClient()
results_path = Path(results_path)
with results_path.open() as fh:
results: list[dict[str, Any]] = json.load(fh)
actionable = [r for r in results if r.get("status") in ("missing", "mismatch")]
suggestions: list[Suggestion] = []
for i, result in enumerate(actionable, start=1):
key = result["key"]
local = result.get("local_entry") or {}
if verbose:
print(f"[{i}/{len(actionable)}] Searching for {key} …")
queries = _build_queries(local)
found: list[dict[str, Any]] = []
for query in queries:
hits = client.search(query, size=5)
if hits:
found = hits
break # first successful strategy wins
local_title = local.get("title", "").strip()
for record in found:
suggestions.append(_record_to_suggestion(key, record, local_title=local_title))
return suggestions
|