Coverage for src/local_deep_research/utilities/citation_normalizer.py: 91%
186 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Normalize search engine result dicts into structured citation metadata.
4Converts engine-specific field names and formats into CSL-JSON vocabulary.
5Each engine has its own dict shape; this module handles the differences.
6"""
8import re
9from datetime import date
10from typing import Any, Optional
11from urllib.parse import urlparse
13from .arxiv import extract_arxiv_id as extract_arxiv_id_from_url
14from .arxiv import is_arxiv_family_url
17__all__ = [
18 "normalize_citation",
19 "normalize_issn",
20 "detect_engine",
21]
24_ISSN_CHARS = re.compile(r"[^0-9Xx]")
27def normalize_issn(s: Optional[str]) -> Optional[str]:
28 """Canonicalize an ISSN to the 8-character no-dash form.
30 Strips dashes, whitespace, and any non-digit/non-X characters; uppercases
31 a trailing "x" check digit. Returns the canonical 8-char form, or None if
32 the input is missing or cannot be coerced into 8 characters.
34 Examples:
35 normalize_issn("1522-9645") == "15229645"
36 normalize_issn("15229645") == "15229645"
37 normalize_issn("1234-567x") == "1234567X"
38 normalize_issn("bad") is None
39 normalize_issn(None) is None
41 This is a structural canonicalization — it does not verify the ISSN
42 checksum. The goal is format-independent equality for lookup.
43 """
44 if not s:
45 return None
46 cleaned = _ISSN_CHARS.sub("", s).upper()
47 if len(cleaned) != 8:
48 return None
49 return cleaned
52# Academic source engines that produce citation-worthy metadata
53ACADEMIC_ENGINES = {
54 "arxiv",
55 "openalex",
56 "semantic_scholar",
57 "pubmed",
58 "nasa_ads",
59}
61# URL patterns to detect source engine from URLs
62_ENGINE_PATTERNS = [
63 (re.compile(r"arxiv\.org"), "arxiv"),
64 (re.compile(r"openalex\.org"), "openalex"),
65 (re.compile(r"semanticscholar\.org"), "semantic_scholar"),
66 (re.compile(r"ncbi\.nlm\.nih\.gov|pubmed"), "pubmed"),
67 (re.compile(r"adsabs\.harvard\.edu|ui\.adsabs"), "nasa_ads"),
68 (re.compile(r"doi\.org"), "doi"),
69]
72def detect_engine(source: dict) -> Optional[str]:
73 """Detect which search engine produced this result.
75 Checks explicit source_engine field first, then URL patterns.
76 Returns None for non-academic sources (web, news, etc.).
77 """
78 # Explicit engine field
79 engine = source.get("source_engine") or source.get("source")
80 if engine:
81 engine_lower = engine.lower().strip()
82 if engine_lower in ACADEMIC_ENGINES:
83 return engine_lower
85 # Detect from URL
86 url = source.get("link", "") or source.get("url", "")
87 for pattern, engine_name in _ENGINE_PATTERNS:
88 if pattern.search(url):
89 return engine_name
91 return None
94def _parse_authors_list(authors: Any) -> Optional[list[dict]]:
95 """Convert various author formats to CSL-JSON name objects.
97 Handles:
98 - List of strings: ["John Smith", "Jane Doe"]
99 - Comma-separated string: "John Smith, Jane Doe"
100 - List of dicts with "name": [{"name": "John Smith"}]
101 - Already CSL format: [{"family": "Smith", "given": "John"}]
102 """
103 if not authors:
104 return None
106 if isinstance(authors, str):
107 authors = [a.strip() for a in authors.split(",") if a.strip()]
109 if not isinstance(authors, list): 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 return None
112 result = []
113 for author in authors:
114 if isinstance(author, dict):
115 if "family" in author:
116 # Whitelist only CSL name fields to ensure the dict is
117 # JSON-serializable. Engines like OpenAlex and
118 # Semantic Scholar attach nested affiliation objects,
119 # ORCIDs, etc., which may contain non-primitive types
120 # that would crash json.dumps() when stored in the
121 # paper_metadata JSON column.
122 safe = {"family": author["family"]}
123 if "given" in author:
124 safe["given"] = author["given"]
125 if "suffix" in author: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 safe["suffix"] = author["suffix"]
127 result.append(safe)
128 elif "name" in author:
129 result.append(_parse_name(author["name"]))
130 elif "display_name" in author:
131 result.append(_parse_name(author["display_name"]))
132 elif isinstance(author, str): 132 ↛ 113line 132 didn't jump to line 113 because the condition on line 132 was always true
133 result.append(_parse_name(author))
135 return result if result else None
138def _parse_name(name: str) -> dict:
139 """Parse a name string into CSL {"family", "given"} format."""
140 name = name.strip()
141 if not name: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 return {"literal": ""}
144 # Handle "Last, First" format
145 if "," in name:
146 parts = name.split(",", 1)
147 return {"family": parts[0].strip(), "given": parts[1].strip()}
149 # Handle "First Last" format
150 parts = name.rsplit(" ", 1)
151 if len(parts) == 2:
152 return {"given": parts[0].strip(), "family": parts[1].strip()}
154 return {"literal": name}
157def _parse_date(source: dict) -> tuple[Optional[date], Optional[int]]:
158 """Extract publication date and year from various formats.
160 Returns (date_obj, year_int). Either may be None.
161 """
162 year = None
164 # Try explicit year field
165 raw_year = source.get("year") or source.get("publication_year")
166 if raw_year:
167 try:
168 year = int(raw_year)
169 except (ValueError, TypeError):
170 pass
172 # Try date string
173 date_str = (
174 source.get("publication_date")
175 or source.get("published")
176 or source.get("date")
177 or source.get("pubdate")
178 )
179 if date_str and isinstance(date_str, str):
180 # Try ISO format: YYYY-MM-DD
181 match = re.match(r"(\d{4})-(\d{1,2})-(\d{1,2})", date_str)
182 if match: 182 ↛ 191line 182 didn't jump to line 191 because the condition on line 182 was always true
183 try:
184 d = date(int(match[1]), int(match[2]), int(match[3]))
185 if year is None:
186 year = d.year
187 return d, year
188 except ValueError:
189 pass
190 # Try just year
191 if year is None:
192 match = re.match(r"(\d{4})", date_str)
193 if match:
194 year = int(match[1])
196 return None, year
199_HOST_LABEL = r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?"
200_PLAIN_HOST_RE = re.compile(rf"{_HOST_LABEL}(?:\.{_HOST_LABEL})*")
203def _with_scheme_for_arxiv_host(url: str) -> str:
204 """Prefix ``https://`` onto a scheme-less arXiv-family link.
206 The leading segment (up to the first ``/``, lower-cased, trailing dots
207 dropped) must be a plain DNS name -- dot-separated labels of ASCII
208 letters, digits and inner hyphens -- on an arXiv-family host as the
209 strict parser defines it (:func:`is_arxiv_family_url`): ``arxiv.org``,
210 ``ar5iv.org`` or a subdomain of either (``www.``, ``export.``,
211 ``ar5iv.labs.``). Only then does the link get a scheme; any other
212 scheme-less string is returned unchanged. So is any string containing
213 ``://`` anywhere (even only in its query) or one ``urlparse`` reads as
214 having a scheme (``arxiv.org:443/...``), so such links yield no id.
215 The strict parser applied afterwards still decides whether the result
216 is an arXiv paper. A string ``urlparse`` rejects outright (an
217 unbalanced ``[`` or ``]`` in the netloc, or a netloc NFKC folds into a
218 delimiter) is returned unchanged too, rather than raising out of
219 :func:`normalize_citation`; the strict parser rejects it in turn.
220 """
221 if "://" in url:
222 return url
223 try:
224 has_scheme = bool(urlparse(url).scheme)
225 except ValueError:
226 return url
227 if has_scheme: 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true
228 return url
229 host = url.split("/", 1)[0].lower().rstrip(".")
230 # A plain DNS-name host only: a backslash, "@", ":", an empty or
231 # hyphen-edged label, or any other delimiter in the leading segment
232 # could make the parsed host differ from what a browser would open, so
233 # such input is left as it is.
234 if not _PLAIN_HOST_RE.fullmatch(host):
235 return url
236 if is_arxiv_family_url(f"https://{host}/"):
237 return f"https://{url}"
238 return url
241def _extract_arxiv_id(source: dict) -> Optional[str]:
242 """Extract arXiv ID from URL or explicit field."""
243 arxiv_id = source.get("arxiv_id")
244 if arxiv_id:
245 return arxiv_id
247 url = source.get("link", "") or source.get("url", "")
248 return (
249 extract_arxiv_id_from_url(_with_scheme_for_arxiv_host(url))
250 if isinstance(url, str)
251 else None
252 )
255def _extract_doi(source: dict) -> Optional[str]:
256 """Extract a DOI string from a result/source dict.
258 Tries (in order):
259 1. ``source["doi"]`` — set by ArXiv, NASA ADS, OpenAlex
260 2. ``source["external_ids"]["DOI"]`` / ``externalIds["DOI"]`` — Semantic
261 Scholar style
262 3. A DOI embedded in ``source["link"]`` (https://doi.org/...)
264 URL prefixes (``https://doi.org/``, ``http://doi.org/``, ``doi:``) are
265 stripped so the returned value is a bare DOI like ``10.1038/...``. This
266 is the single source of truth for DOI extraction across the codebase —
267 `openalex_enrichment` and other modules import it from here.
268 """
269 doi = source.get("doi")
270 if isinstance(doi, list):
271 doi = doi[0] if doi else None
273 # Semantic Scholar exposes DOIs through external_ids / externalIds
274 if not doi:
275 ext_ids = source.get("external_ids") or source.get("externalIds") or {}
276 if isinstance(ext_ids, dict): 276 ↛ 282line 276 didn't jump to line 282 because the condition on line 276 was always true
277 doi = ext_ids.get("DOI") or ext_ids.get("doi")
279 # DOI embedded in a link URL — anchor on scheme to avoid CodeQL
280 # py/incomplete-url-substring-sanitization (an attacker-controlled
281 # URL path could contain "doi.org/" otherwise).
282 if not doi:
283 link = source.get("link") or ""
284 if isinstance(link, str): 284 ↛ 295line 284 didn't jump to line 295 because the condition on line 284 was always true
285 for prefix in (
286 "https://doi.org/",
287 "http://doi.org/",
288 "https://dx.doi.org/",
289 "http://dx.doi.org/",
290 ):
291 if link.startswith(prefix):
292 doi = link[len(prefix) :]
293 break
295 if not doi:
296 return None
298 doi = str(doi)
299 for prefix in (
300 "https://doi.org/",
301 "http://doi.org/",
302 "https://dx.doi.org/",
303 "http://dx.doi.org/",
304 "doi:",
305 ):
306 if doi.startswith(prefix):
307 doi = doi[len(prefix) :]
308 break
309 return doi if doi else None
312def normalize_citation(source: dict) -> Optional[dict]:
313 """Normalize a search result dict into citation metadata fields.
315 Returns a dict of field values ready for Paper creation,
316 or None if the source is not an academic result.
318 The returned dict has keys matching Paper column names.
319 """
320 engine = detect_engine(source)
321 if engine is None:
322 return None
324 pub_date, year = _parse_date(source)
325 doi = _extract_doi(source)
327 # Guard against source["metadata"] being a non-dict truthy value
328 # (e.g., a string) — .get({}).get() crashes with AttributeError
329 # in that case because the default only applies when the key is
330 # absent/None.
331 nested_meta = source.get("metadata")
332 if not isinstance(nested_meta, dict):
333 nested_meta = {}
335 # Prefer a structured CSL list (e.g. NASA ADS publishes "Last, First"
336 # names which lose their pairing if re-split from a comma-joined
337 # display string) over the display-string fallback.
338 authors_input = (
339 source.get("authors_csl")
340 or nested_meta.get("authors_csl")
341 or source.get("authors")
342 or nested_meta.get("authors")
343 )
344 authors = _parse_authors_list(authors_input)
346 # Container title (journal/conference name).
347 # Checks CSL-style ``container_title`` and ``container-title`` too
348 # so callers that already use CSL vocabulary don't have their
349 # journal silently dropped.
350 container = (
351 source.get("journal_ref")
352 or source.get("journal")
353 or source.get("venue")
354 or source.get("container_title")
355 or source.get("container-title")
356 or nested_meta.get("journal")
357 )
358 # Placeholder sentinels from upstream engines (e.g. OpenAlex /
359 # NASA ADS set these when no venue is indexed). Filter them out
360 # so they don't become container_title literals — there's an
361 # actual OpenAlex source named "unknown" (Q1, h_index=5) that
362 # would otherwise get hit by the name-based lookup.
363 if isinstance(container, str) and container.strip().lower() in (
364 "unknown",
365 "",
366 ):
367 container = None
369 # Item type (CSL vocabulary)
370 item_type = _infer_item_type(engine, source)
372 fields = {
373 "source_engine": engine,
374 "doi": doi,
375 "arxiv_id": _extract_arxiv_id(source) if engine == "arxiv" else None,
376 "pmid": source.get("pmid"),
377 "pmcid": source.get("pmcid"),
378 "authors": authors,
379 "publication_date": pub_date,
380 "year": year,
381 "volume": source.get("volume"),
382 "issue": source.get("issue"),
383 "pages": source.get("pages"),
384 "container_title": container,
385 "publisher": source.get("publisher"),
386 "item_type": item_type,
387 }
389 # Build CSL-JSON item (uses the date object while still native)
390 fields["csl_json"] = _build_csl_json(source, fields)
392 # Convert publication_date to ISO string so it survives JSON
393 # serialization when stored in paper_metadata. _build_csl_json has
394 # already consumed the native date object above.
395 if fields.get("publication_date") is not None:
396 fields["publication_date"] = fields["publication_date"].isoformat()
398 # Strip None values to avoid unnecessary DB writes
399 return {k: v for k, v in fields.items() if v is not None}
402def _infer_item_type(engine: str, source: dict) -> str:
403 """Infer CSL item type from engine and source metadata."""
404 source_type = source.get("source_type", "")
405 if source_type == "conference": 405 ↛ 406line 405 didn't jump to line 406 because the condition on line 405 was never true
406 return "paper-conference"
408 if engine == "arxiv":
409 # ArXiv papers with journal_ref are published; without are preprints
410 return "article-journal" if source.get("journal_ref") else "article"
412 return "article-journal"
415def _build_csl_json(source: dict, fields: dict) -> dict:
416 """Build a CSL-JSON item from normalized fields."""
417 csl: dict[str, Any] = {
418 "type": fields.get("item_type", "article-journal"),
419 "title": source.get("title", ""),
420 }
422 if fields.get("authors"):
423 csl["author"] = fields["authors"]
425 if fields.get("container_title"):
426 csl["container-title"] = fields["container_title"]
428 if fields.get("doi"):
429 csl["DOI"] = fields["doi"]
431 if fields.get("volume"):
432 csl["volume"] = fields["volume"]
433 if fields.get("issue"):
434 csl["issue"] = fields["issue"]
435 if fields.get("pages"):
436 csl["page"] = fields["pages"]
438 if fields.get("publisher"): 438 ↛ 439line 438 didn't jump to line 439 because the condition on line 438 was never true
439 csl["publisher"] = fields["publisher"]
441 if fields.get("year"):
442 date_parts = [[fields["year"]]]
443 if fields.get("publication_date"):
444 d = fields["publication_date"]
445 date_parts = [[d.year, d.month, d.day]]
446 csl["issued"] = {"date-parts": date_parts}
448 url = source.get("link") or source.get("url")
449 if url:
450 csl["URL"] = url
452 return csl