Edit on GitHub

agent_search_gateway.academic.normalization

Pure normalization helpers for academic paper identity and metadata.

View Source

  1"""Pure normalization helpers for academic paper identity and metadata."""
  2
  3from __future__ import annotations
  4
  5import re
  6import unicodedata
  7from datetime import date
  8from urllib.parse import unquote, urlsplit
  9
 10_DOI_RE = re.compile(r"^10\.\d{4,9}/\S+$", re.IGNORECASE)
 11_ARXIV_RE = re.compile(r"^(?P<work>\d{4}\.\d{4,5})(?:v\d+)?$", re.IGNORECASE)
 12_SEMANTIC_SCHOLAR_RE = re.compile(r"^[A-Za-z0-9]{6,64}$")
 13_OPENALEX_RE = re.compile(r"^W\d+$", re.IGNORECASE)
 14_CORE_RE = re.compile(r"^[A-Za-z0-9._:-]+$")
 15_WHITESPACE_RE = re.compile(r"\s+")
 16
 17
 18def _nfkc(value: str) -> str:
 19    return unicodedata.normalize("NFKC", value)
 20
 21
 22def _collapsed(value: str) -> str:
 23    return _WHITESPACE_RE.sub(" ", _nfkc(value).strip())
 24
 25
 26def normalize_doi(value: str) -> str | None:
 27    """Canonicalize a DOI from a bare value, ``doi:`` form, or DOI URL."""
 28
 29    if not isinstance(value, str):
 30        return None
 31    candidate = _nfkc(value).strip()
 32    if not candidate:
 33        return None
 34    if candidate.casefold().startswith("doi:"):
 35        candidate = candidate[4:].strip()
 36    else:
 37        try:
 38            parsed = urlsplit(candidate)
 39        except ValueError:
 40            return None
 41        if parsed.scheme or parsed.netloc:
 42            if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
 43                return None
 44            if parsed.hostname.casefold() not in {"doi.org", "www.doi.org", "dx.doi.org"}:
 45                return None
 46            if parsed.query or parsed.fragment:
 47                return None
 48            candidate = unquote(parsed.path.lstrip("/"))
 49    candidate = candidate.strip()
 50    if not _DOI_RE.fullmatch(candidate):
 51        return None
 52    return candidate.casefold()
 53
 54
 55def normalize_arxiv_id(value: str) -> str | None:
 56    """Canonicalize a modern arXiv identifier and remove its version suffix."""
 57
 58    if not isinstance(value, str):
 59        return None
 60    candidate = _nfkc(value).strip()
 61    if not candidate:
 62        return None
 63    if candidate.casefold().startswith("arxiv:"):
 64        candidate = candidate[6:].strip()
 65    else:
 66        try:
 67            parsed = urlsplit(candidate)
 68        except ValueError:
 69            return None
 70        if parsed.scheme or parsed.netloc:
 71            if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
 72                return None
 73            if parsed.hostname.casefold() not in {"arxiv.org", "www.arxiv.org"}:
 74                return None
 75            if parsed.query or parsed.fragment:
 76                return None
 77            path = parsed.path.strip("/")
 78            if path.startswith("abs/"):
 79                candidate = path[4:]
 80            elif path.startswith("pdf/") and path.casefold().endswith(".pdf"):
 81                candidate = path[4:-4]
 82            else:
 83                return None
 84    match = _ARXIV_RE.fullmatch(candidate.strip())
 85    return match.group("work") if match else None
 86
 87
 88def normalize_semantic_scholar_id(value: str) -> str | None:
 89    if not isinstance(value, str):
 90        return None
 91    candidate = _nfkc(value).strip()
 92    return candidate if _SEMANTIC_SCHOLAR_RE.fullmatch(candidate) else None
 93
 94
 95def normalize_openalex_id(value: str) -> str | None:
 96    if not isinstance(value, str):
 97        return None
 98    candidate = _nfkc(value).strip()
 99    if not candidate:
100        return None
101    try:
102        parsed = urlsplit(candidate)
103    except ValueError:
104        return None
105    if parsed.scheme or parsed.netloc:
106        if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
107            return None
108        if parsed.hostname.casefold() not in {"openalex.org", "www.openalex.org"}:
109            return None
110        if parsed.query or parsed.fragment:
111            return None
112        candidate = parsed.path.strip("/")
113    if not _OPENALEX_RE.fullmatch(candidate):
114        return None
115    return candidate.upper()
116
117
118def normalize_dblp_key(value: str) -> str | None:
119    if not isinstance(value, str):
120        return None
121    candidate = _nfkc(value).strip()
122    if not candidate:
123        return None
124    try:
125        parsed = urlsplit(candidate)
126    except ValueError:
127        return None
128    if parsed.scheme or parsed.netloc:
129        if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
130            return None
131        if parsed.hostname.casefold() not in {"dblp.org", "www.dblp.org"}:
132            return None
133        if parsed.query or parsed.fragment:
134            return None
135        path = parsed.path.strip("/")
136        if not path.startswith("rec/"):
137            return None
138        candidate = path[4:]
139        if candidate.casefold().endswith(".html"):
140            candidate = candidate[:-5]
141    candidate = candidate.strip("/")
142    if not candidate or any(character.isspace() for character in candidate):
143        return None
144    return candidate
145
146
147def normalize_core_id(value: str) -> str | None:
148    if not isinstance(value, str):
149        return None
150    candidate = _nfkc(value).strip()
151    return candidate if candidate and _CORE_RE.fullmatch(candidate) else None
152
153
154def normalize_source_name(value: str) -> str:
155    if not isinstance(value, str):
156        return ""
157    return _collapsed(value).casefold().replace(" ", "_")
158
159
160def normalize_source_id(source: str, value: str) -> tuple[str, str] | None:
161    """Canonicalize a native identifier and namespace it by provider source."""
162
163    source_name = normalize_source_name(source)
164    if not source_name or not isinstance(value, str):
165        return None
166    normalizer = {
167        "arxiv": normalize_arxiv_id,
168        "semantic_scholar": normalize_semantic_scholar_id,
169        "openalex": normalize_openalex_id,
170        "dblp": normalize_dblp_key,
171        "crossref": normalize_doi,
172        "core": normalize_core_id,
173    }.get(source_name)
174    normalized = normalizer(value) if normalizer is not None else _nfkc(value).strip()
175    if not normalized:
176        return None
177    return source_name, normalized
178
179
180def source_identity_key(source: str, value: str) -> str | None:
181    identity = normalize_source_id(source, value)
182    return None if identity is None else f"{identity[0]}:{identity[1]}"
183
184
185def normalize_title(value: str) -> str:
186    return _collapsed(value).casefold() if isinstance(value, str) else ""
187
188
189def normalize_author(value: str) -> str:
190    return _collapsed(value).casefold() if isinstance(value, str) else ""
191
192
193def normalize_topic(value: str) -> str:
194    return _collapsed(value) if isinstance(value, str) else ""
195
196
197def publication_year(value: date | None) -> int | None:
198    return value.year if isinstance(value, date) else None
199
200
201def normalized_authors(authors: tuple[str, ...]) -> frozenset[str]:
202    return frozenset(normalized for author in authors if (normalized := normalize_author(author)))
203
204
205def bibliographic_fingerprint(
206    title: str,
207    authors: tuple[str, ...],
208    published_date: date | None,
209) -> tuple[str, frozenset[str], int] | None:
210    """Build strict weak-identity evidence only when all required parts exist."""
211
212    normalized_title = normalize_title(title)
213    author_evidence = normalized_authors(authors)
214    year = publication_year(published_date)
215    if not normalized_title or not author_evidence or year is None:
216        return None
217    return normalized_title, author_evidence, year
218
219
220def bibliographic_fingerprints_match(
221    left: tuple[str, frozenset[str], int] | None,
222    right: tuple[str, frozenset[str], int] | None,
223) -> bool:
224    if left is None or right is None:
225        return False
226    left_title, left_authors, left_year = left
227    right_title, right_authors, right_year = right
228    return (
229        left_title == right_title and left_year == right_year and bool(left_authors & right_authors)
230    )
231
232
233def weak_bibliographic_match(
234    *,
235    left_title: str,
236    left_authors: tuple[str, ...],
237    left_published_date: date | None,
238    right_title: str,
239    right_authors: tuple[str, ...],
240    right_published_date: date | None,
241) -> bool:
242    return bibliographic_fingerprints_match(
243        bibliographic_fingerprint(left_title, left_authors, left_published_date),
244        bibliographic_fingerprint(right_title, right_authors, right_published_date),
245    )

def normalize_doi(value: str) -> str | None: View Source

27def normalize_doi(value: str) -> str | None:
28    """Canonicalize a DOI from a bare value, ``doi:`` form, or DOI URL."""
29
30    if not isinstance(value, str):
31        return None
32    candidate = _nfkc(value).strip()
33    if not candidate:
34        return None
35    if candidate.casefold().startswith("doi:"):
36        candidate = candidate[4:].strip()
37    else:
38        try:
39            parsed = urlsplit(candidate)
40        except ValueError:
41            return None
42        if parsed.scheme or parsed.netloc:
43            if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
44                return None
45            if parsed.hostname.casefold() not in {"doi.org", "www.doi.org", "dx.doi.org"}:
46                return None
47            if parsed.query or parsed.fragment:
48                return None
49            candidate = unquote(parsed.path.lstrip("/"))
50    candidate = candidate.strip()
51    if not _DOI_RE.fullmatch(candidate):
52        return None
53    return candidate.casefold()

Canonicalize a DOI from a bare value, doi: form, or DOI URL.

def normalize_arxiv_id(value: str) -> str | None: View Source

56def normalize_arxiv_id(value: str) -> str | None:
57    """Canonicalize a modern arXiv identifier and remove its version suffix."""
58
59    if not isinstance(value, str):
60        return None
61    candidate = _nfkc(value).strip()
62    if not candidate:
63        return None
64    if candidate.casefold().startswith("arxiv:"):
65        candidate = candidate[6:].strip()
66    else:
67        try:
68            parsed = urlsplit(candidate)
69        except ValueError:
70            return None
71        if parsed.scheme or parsed.netloc:
72            if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
73                return None
74            if parsed.hostname.casefold() not in {"arxiv.org", "www.arxiv.org"}:
75                return None
76            if parsed.query or parsed.fragment:
77                return None
78            path = parsed.path.strip("/")
79            if path.startswith("abs/"):
80                candidate = path[4:]
81            elif path.startswith("pdf/") and path.casefold().endswith(".pdf"):
82                candidate = path[4:-4]
83            else:
84                return None
85    match = _ARXIV_RE.fullmatch(candidate.strip())
86    return match.group("work") if match else None

Canonicalize a modern arXiv identifier and remove its version suffix.

def normalize_semantic_scholar_id(value: str) -> str | None: View Source

89def normalize_semantic_scholar_id(value: str) -> str | None:
90    if not isinstance(value, str):
91        return None
92    candidate = _nfkc(value).strip()
93    return candidate if _SEMANTIC_SCHOLAR_RE.fullmatch(candidate) else None

def normalize_openalex_id(value: str) -> str | None: View Source

 96def normalize_openalex_id(value: str) -> str | None:
 97    if not isinstance(value, str):
 98        return None
 99    candidate = _nfkc(value).strip()
100    if not candidate:
101        return None
102    try:
103        parsed = urlsplit(candidate)
104    except ValueError:
105        return None
106    if parsed.scheme or parsed.netloc:
107        if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
108            return None
109        if parsed.hostname.casefold() not in {"openalex.org", "www.openalex.org"}:
110            return None
111        if parsed.query or parsed.fragment:
112            return None
113        candidate = parsed.path.strip("/")
114    if not _OPENALEX_RE.fullmatch(candidate):
115        return None
116    return candidate.upper()

def normalize_dblp_key(value: str) -> str | None: View Source

119def normalize_dblp_key(value: str) -> str | None:
120    if not isinstance(value, str):
121        return None
122    candidate = _nfkc(value).strip()
123    if not candidate:
124        return None
125    try:
126        parsed = urlsplit(candidate)
127    except ValueError:
128        return None
129    if parsed.scheme or parsed.netloc:
130        if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
131            return None
132        if parsed.hostname.casefold() not in {"dblp.org", "www.dblp.org"}:
133            return None
134        if parsed.query or parsed.fragment:
135            return None
136        path = parsed.path.strip("/")
137        if not path.startswith("rec/"):
138            return None
139        candidate = path[4:]
140        if candidate.casefold().endswith(".html"):
141            candidate = candidate[:-5]
142    candidate = candidate.strip("/")
143    if not candidate or any(character.isspace() for character in candidate):
144        return None
145    return candidate

def normalize_core_id(value: str) -> str | None: View Source

148def normalize_core_id(value: str) -> str | None:
149    if not isinstance(value, str):
150        return None
151    candidate = _nfkc(value).strip()
152    return candidate if candidate and _CORE_RE.fullmatch(candidate) else None

def normalize_source_name(value: str) -> str: View Source

155def normalize_source_name(value: str) -> str:
156    if not isinstance(value, str):
157        return ""
158    return _collapsed(value).casefold().replace(" ", "_")

def normalize_source_id(source: str, value: str) -> tuple[str, str] | None: View Source

161def normalize_source_id(source: str, value: str) -> tuple[str, str] | None:
162    """Canonicalize a native identifier and namespace it by provider source."""
163
164    source_name = normalize_source_name(source)
165    if not source_name or not isinstance(value, str):
166        return None
167    normalizer = {
168        "arxiv": normalize_arxiv_id,
169        "semantic_scholar": normalize_semantic_scholar_id,
170        "openalex": normalize_openalex_id,
171        "dblp": normalize_dblp_key,
172        "crossref": normalize_doi,
173        "core": normalize_core_id,
174    }.get(source_name)
175    normalized = normalizer(value) if normalizer is not None else _nfkc(value).strip()
176    if not normalized:
177        return None
178    return source_name, normalized

Canonicalize a native identifier and namespace it by provider source.

def source_identity_key(source: str, value: str) -> str | None: View Source

181def source_identity_key(source: str, value: str) -> str | None:
182    identity = normalize_source_id(source, value)
183    return None if identity is None else f"{identity[0]}:{identity[1]}"

def normalize_title(value: str) -> str: View Source

186def normalize_title(value: str) -> str:
187    return _collapsed(value).casefold() if isinstance(value, str) else ""

def normalize_author(value: str) -> str: View Source

190def normalize_author(value: str) -> str:
191    return _collapsed(value).casefold() if isinstance(value, str) else ""

def normalize_topic(value: str) -> str: View Source

194def normalize_topic(value: str) -> str:
195    return _collapsed(value) if isinstance(value, str) else ""

def publication_year(value: datetime.date | None) -> int | None: View Source

198def publication_year(value: date | None) -> int | None:
199    return value.year if isinstance(value, date) else None

def normalized_authors(authors: tuple[str, ...]) -> frozenset[str]: View Source

202def normalized_authors(authors: tuple[str, ...]) -> frozenset[str]:
203    return frozenset(normalized for author in authors if (normalized := normalize_author(author)))

def bibliographic_fingerprint( title: str, authors: tuple[str, ...], published_date: datetime.date | None) -> tuple[str, frozenset[str], int] | None: View Source

206def bibliographic_fingerprint(
207    title: str,
208    authors: tuple[str, ...],
209    published_date: date | None,
210) -> tuple[str, frozenset[str], int] | None:
211    """Build strict weak-identity evidence only when all required parts exist."""
212
213    normalized_title = normalize_title(title)
214    author_evidence = normalized_authors(authors)
215    year = publication_year(published_date)
216    if not normalized_title or not author_evidence or year is None:
217        return None
218    return normalized_title, author_evidence, year

Build strict weak-identity evidence only when all required parts exist.

def bibliographic_fingerprints_match( left: tuple[str, frozenset[str], int] | None, right: tuple[str, frozenset[str], int] | None) -> bool: View Source

221def bibliographic_fingerprints_match(
222    left: tuple[str, frozenset[str], int] | None,
223    right: tuple[str, frozenset[str], int] | None,
224) -> bool:
225    if left is None or right is None:
226        return False
227    left_title, left_authors, left_year = left
228    right_title, right_authors, right_year = right
229    return (
230        left_title == right_title and left_year == right_year and bool(left_authors & right_authors)
231    )

def weak_bibliographic_match( *, left_title: str, left_authors: tuple[str, ...], left_published_date: datetime.date | None, right_title: str, right_authors: tuple[str, ...], right_published_date: datetime.date | None) -> bool: View Source

234def weak_bibliographic_match(
235    *,
236    left_title: str,
237    left_authors: tuple[str, ...],
238    left_published_date: date | None,
239    right_title: str,
240    right_authors: tuple[str, ...],
241    right_published_date: date | None,
242) -> bool:
243    return bibliographic_fingerprints_match(
244        bibliographic_fingerprint(left_title, left_authors, left_published_date),
245        bibliographic_fingerprint(right_title, right_authors, right_published_date),
246    )