Repository Public Research preview
1"""Crossref adapter: pre-2018 economics / finance journal articles under reusable Creative Commons licences.23Discovery is metadata-first through the Crossref REST API (``/works`` with ``filter=issn:…,until-pub-date:…,4has-license:true``, ``rows=1000``, deep cursor paging, ``mailto`` polite pool — the contact address is sent to5Crossref only). For every item the adapter671. picks the publication date ``published-print`` → ``published-online`` → ``issued`` (date-parts) and keeps the8 item only if that date is before 2018-01-01 (evidence string ``"crossref <field> date-parts"``);92. picks the licence from ``license[]``: the entry whose ``content-version`` is ``vor`` or ``unspecified`` with the10 earliest ``start`` (already in force), normalised with ``normalize_license(uri=URL)``; no such entry → skipped;113. picks a full-text route: Crossref ``link[]`` entries with ``intended-application`` text-mining or12 similarity-checking (``application/xml`` JATS first, then PDF), else the journal's configured canonical open13 PDF template (MDPI ``<landing>/pdf``, SpringerOpen ``track/pdf``);144. downloads the file (robots.txt honoured with RFC 9309 longest-match semantics, ≥ 2 s between requests to the15 same publisher host, no contact address sent to publishers), checks the magic bytes, and writes16 ``data/raw/crossref/<doi-sanitised>.{pdf,xml}`` plus ``<stem>.meta.json`` (stub keys documented in17 ``htr.ingest.corpus``).1819Sampling is deterministic per journal (``random.Random(f"{seed}:{issn}")``) and resumable: DOIs whose stub is20already on disk are skipped, and every attempt is appended to ``_ledger.jsonl``.21"""22from __future__ import annotations2324import datetime as _dt25import json26import random27import re28from dataclasses import dataclass, field29from pathlib import Path30from urllib.parse import quote, urlparse3132from htr.licenses.policy import LicenseRecord, normalize_license33from htr.utils.hashing import sha256_bytes, short_id34from htr.utils.http import DEFAULT_POLICIES, USER_AGENT, HostPolicy, PoliteClient, RateLimiter35from htr.utils.io import append_jsonl, atomic_write_bytes, ensure_dir, read_json, write_json36from htr.utils.provenance import utc_now3738API = "https://api.crossref.org"39CROSSREF_MAILTO = "spbou4@icloud.com" # Crossref polite pool only — never sent to publishers40PUBLISHER_MIN_INTERVAL_S = 2.041MAX_FILE_BYTES = 60 * 1024 * 102442SELECT_FIELDS = "DOI,type,title,author,container-title,ISSN,issn-type,publisher,volume,issue,page,published-print,published-online,issued,license,link,resource,subject"434445# ---------------------------------------------------------------------------------------------------------46# HTTP plumbing474849class PublisherRateLimiter(RateLimiter):50 """RateLimiter whose default interval for unknown (publisher) hosts is ``PUBLISHER_MIN_INTERVAL_S``."""5152 def wait(self, host: str) -> None: # noqa: D40153 if host not in self.policies:54 self.policies[host] = HostPolicy(PUBLISHER_MIN_INTERVAL_S, True)55 super().wait(host)565758def crossref_client() -> PoliteClient:59 return PoliteClient(timeout=90, mailto=CROSSREF_MAILTO)606162def publisher_client() -> PoliteClient:63 return PoliteClient(timeout=120, limiter=PublisherRateLimiter(policies=dict(DEFAULT_POLICIES)))646566class Robots:67 """robots.txt evaluation with RFC 9309 semantics (longest matching rule wins, ``*`` and ``$`` wildcards,68 Allow wins ties). ``urllib.robotparser`` applies first-match order — which wrongly refuses69 ``link.springer.com/content/pdf/…`` (``Disallow: /`` listed before ``Allow: /content/``) — and fetches with70 the ``Python-urllib`` agent that some CDNs block (MDPI answers 403 → "disallow all"). Rules are fetched71 through the polite client with the project user agent. 4xx → allowed (RFC 9309 §2.3.1.3); 5xx or network72 failure → disallowed."""7374 def __init__(self, client: PoliteClient, agent: str = USER_AGENT.split("/")[0]):75 self.client = client76 self.agent = agent.lower()77 self.cache: dict[str, list[tuple[bool, str]] | None | bool] = {}7879 @staticmethod80 def _parse(text: str) -> dict[str, list[tuple[bool, str]]]:81 groups: dict[str, list[tuple[bool, str]]] = {}82 agents: list[str] = []83 last_was_agent = False84 for raw in text.splitlines():85 line = raw.split("#", 1)[0].strip()86 if ":" not in line:87 continue88 k, v = (x.strip() for x in line.split(":", 1))89 k = k.lower()90 if k == "user-agent":91 if not last_was_agent:92 agents = []93 agents.append(v.lower())94 groups.setdefault(v.lower(), [])95 last_was_agent = True96 elif k in ("allow", "disallow"):97 last_was_agent = False98 if k == "disallow" and v == "":99 continue100 for a in agents:101 groups.setdefault(a, []).append((k == "allow", v))102 else:103 last_was_agent = False104 return groups105106 @staticmethod107 def _match(pattern: str, path: str) -> bool:108 rx = "^" + re.escape(pattern).replace(r"\*", ".*")109 if rx.endswith(r"\$"):110 rx = rx[:-2] + "$"111 return re.match(rx, path) is not None112113 def allowed(self, url: str) -> bool:114 u = urlparse(url)115 host = f"{u.scheme}://{u.netloc}"116 if host not in self.cache:117 try:118 r = self.client.get(f"{host}/robots.txt", check_robots=False)119 if 400 <= r.status_code < 500:120 self.cache[host] = True121 elif r.status_code >= 500:122 self.cache[host] = False123 else:124 groups = self._parse(r.text)125 rules = None126 for name, rl in groups.items():127 if name != "*" and name == self.agent:128 rules = rl129 break130 self.cache[host] = rules if rules is not None else groups.get("*", [])131 except Exception: # noqa: BLE001132 self.cache[host] = False133 rules = self.cache[host]134 if isinstance(rules, bool):135 return rules136 path = (u.path or "/") + (("?" + u.query) if u.query else "")137 best: tuple[int, bool] | None = None138 for allow, pat in rules:139 if self._match(pat, path):140 cand = (len(pat), allow)141 if best is None or cand[0] > best[0] or (cand[0] == best[0] and allow):142 best = cand143 return True if best is None else best[1]144145146# ---------------------------------------------------------------------------------------------------------147# Item-level helpers (pure functions, unit-testable)148149150def sanitize_doi(doi: str) -> str:151 return re.sub(r"[^a-z0-9._-]+", "_", doi.lower()).strip("_")152153154def _date_iso(parts: list) -> str | None:155 if not parts or not parts[0] or parts[0][0] is None:156 return None157 p = [int(x) for x in parts[0] if x is not None]158 iso = f"{p[0]:04d}"159 if len(p) > 1:160 iso += f"-{p[1]:02d}"161 if len(p) > 2:162 iso += f"-{p[2]:02d}"163 return iso164165166def pick_date(item: dict) -> tuple[str | None, str | None]:167 for fld in ("published-print", "published-online", "issued"):168 iso = _date_iso((item.get(fld) or {}).get("date-parts") or [])169 if iso:170 return iso, f"crossref {fld} date-parts"171 return None, None172173174def is_before(iso: str | None, until: str) -> bool:175 if not iso:176 return False177 full = iso + ("-01" if len(iso) == 7 else "-01-01" if len(iso) == 4 else "")178 limit = _dt.date.fromisoformat(until) + _dt.timedelta(days=1)179 return _dt.date.fromisoformat(full) < limit180181182def pick_license(item: dict, today: _dt.date | None = None) -> tuple[str, dict | None]:183 """Canonical code + the chosen ``license[]`` entry (vor/unspecified, earliest start already in force)."""184 today = today or _dt.date.today()185 cands = []186 for lic in item.get("license") or []:187 if lic.get("content-version") not in ("vor", "unspecified"):188 continue189 start = _date_iso((lic.get("start") or {}).get("date-parts") or []) or "0001"190 full = start + ("-01" if len(start) == 7 else "-01-01" if len(start) == 4 else "")191 try:192 sd = _dt.date.fromisoformat(full)193 except ValueError:194 sd = _dt.date(1, 1, 1)195 if sd > today:196 continue197 cands.append((sd, lic))198 if not cands:199 return "UNKNOWN", None200 cands.sort(key=lambda t: t[0])201 lic = cands[0][1]202 return normalize_license(None, lic.get("URL")), lic203204205def _is_pdf_link(link: dict) -> bool:206 ct = (link.get("content-type") or "").lower()207 url = link.get("URL") or ""208 return ct == "application/pdf" or (ct in ("unspecified", "") and re.search(r"(/pdf/?$|\.pdf$|/pdf/)", url, re.I) is not None)209210211def fulltext_candidates(item: dict, journal: dict) -> list[tuple[str, str, str]]:212 """Ordered (url, kind, route) candidates: Crossref XML links, Crossref PDF links, configured fallbacks."""213 xml, pdf = [], []214 for link in item.get("link") or []:215 if link.get("intended-application") not in ("text-mining", "similarity-checking"):216 continue217 if link.get("content-version") not in ("vor", "am", None):218 continue219 ct = (link.get("content-type") or "").lower()220 url = link.get("URL")221 if not url:222 continue223 route = f"crossref link {link.get('intended-application')} {ct or 'unspecified'}"224 if ct in ("application/xml", "text/xml", "application/jats+xml"):225 xml.append((url, "xml", route))226 elif _is_pdf_link(link):227 pdf.append((url, "pdf", route))228 out: list[tuple[str, str, str]] = []229 seen: set[str] = set()230 for url, kind, route in xml + pdf:231 key = re.sub(r"^https?://", "", url).rstrip("/").lower()232 if key not in seen:233 seen.add(key)234 out.append((url.replace("http://", "https://", 1), kind, route))235 doi = item.get("DOI", "")236 landing = ((item.get("resource") or {}).get("primary") or {}).get("URL", "")237 for tpl in journal.get("fallback_pdf", []) or []:238 url = tpl.format(doi=doi, doi_q=quote(doi, safe=""), landing=landing.rstrip("/"), landing_https=landing.replace("http://", "https://", 1).rstrip("/"), landing_host=urlparse(landing).netloc)239 key = re.sub(r"^https?://", "", url).rstrip("/").lower()240 if url and key not in seen:241 seen.add(key)242 out.append((url, "pdf", "publisher canonical open pdf"))243 return out244245246def author_names(item: dict) -> list[str]:247 out = []248 for a in item.get("author") or []:249 nm = " ".join(x for x in (a.get("given"), a.get("family")) if x) or a.get("name") or ""250 nm = re.sub(r"\s+", " ", nm).strip()251 if nm:252 out.append(nm)253 return out254255256def author_group(authors: list[str]) -> str | None:257 """Same normalisation as ``htr.parse.jats``: last token of the first author's name + first initial."""258 if not authors:259 return None260 toks = authors[0].split()261 if not toks:262 return None263 return re.sub(r"[^a-z]", "", toks[-1].lower()) + ("_" + toks[0][0].lower() if len(toks) > 1 else "")264265266def trim_item(item: dict) -> dict:267 keep = ("DOI", "type", "title", "container-title", "ISSN", "issn-type", "publisher", "volume", "issue", "page", "published-print", "published-online", "issued", "license", "link", "subject", "language")268 out = {k: item[k] for k in keep if k in item}269 out["author"] = [{k: a[k] for k in ("given", "family", "name", "sequence") if k in a} for a in item.get("author") or []]270 prim = ((item.get("resource") or {}).get("primary") or {}).get("URL")271 if prim:272 out["resource_primary"] = prim273 return out274275276# ---------------------------------------------------------------------------------------------------------277# Crossref API278279280def load_journals(cfg: dict | str | Path) -> list[dict]:281 if not isinstance(cfg, dict):282 cfg = read_json(cfg)283 return [j for j in cfg.get("journals", []) if j.get("enabled", True)]284285286def iter_works(client: PoliteClient, issns: list[str], until: str, rows: int = 1000, extra_filter: str = ""):287 filt = ",".join([f"issn:{i}" for i in issns] + [f"until-pub-date:{until}", "has-license:true"]) + (f",{extra_filter}" if extra_filter else "")288 cursor = "*"289 while True:290 r = client.get(f"{API}/works", params={"filter": filt, "rows": str(rows), "cursor": cursor, "select": SELECT_FIELDS, "mailto": CROSSREF_MAILTO})291 r.raise_for_status()292 msg = r.json()["message"]293 items = msg.get("items", [])294 yield from items295 nxt = msg.get("next-cursor")296 if not items or not nxt or len(items) < rows:297 return298 cursor = nxt299300301def works_total(client: PoliteClient, issns: list[str], until: str, has_license: bool = True) -> int:302 filt = ",".join([f"issn:{i}" for i in issns] + [f"until-pub-date:{until}"] + (["has-license:true"] if has_license else []))303 r = client.get(f"{API}/works", params={"filter": filt, "rows": "0", "mailto": CROSSREF_MAILTO})304 r.raise_for_status()305 return int(r.json()["message"]["total-results"])306307308@dataclass309class Candidate:310 item: dict311 journal: dict312 pub_date: str313 evidence: str314 code: str315 lic_entry: dict316 routes: list[tuple[str, str, str]] = field(default_factory=list)317318319def screen_journal(client: PoliteClient, journal: dict, allowed_codes: set[str], until: str, stats: dict, today: _dt.date | None = None) -> tuple[list[Candidate], dict]:320 """Enumerate all Crossref records of one journal and apply the date / licence / route screen."""321 issns = journal["issn"] if isinstance(journal["issn"], list) else [journal["issn"]]322 js = {"records_seen": 0, "not_journal_article": 0, "not_pre2018": 0, "no_license": 0, "license_excluded": 0, "no_fulltext": 0, "eligible": 0, "by_license": {}, "years": {}}323 out: list[Candidate] = []324 seen_dois: set[str] = set()325 for it in iter_works(client, issns, until):326 doi = (it.get("DOI") or "").lower()327 if not doi or doi in seen_dois:328 continue329 seen_dois.add(doi)330 js["records_seen"] += 1331 stats["records_seen"] += 1332 if it.get("type") not in ("journal-article", None):333 js["not_journal_article"] += 1334 stats["not_journal_article"] = stats.get("not_journal_article", 0) + 1335 continue336 iso, ev = pick_date(it)337 if not is_before(iso, until):338 js["not_pre2018"] += 1339 stats["not_pre2018"] += 1340 continue341 code, lic = pick_license(it, today)342 js["by_license"][code] = js["by_license"].get(code, 0) + 1343 if lic is None:344 js["no_license"] += 1345 stats["no_license"] += 1346 continue347 if code not in allowed_codes:348 js["license_excluded"] += 1349 stats["license_excluded"] += 1350 continue351 routes = fulltext_candidates(it, journal)352 if not routes:353 js["no_fulltext"] += 1354 stats["no_fulltext"] += 1355 continue356 js["eligible"] += 1357 js["years"][iso[:4]] = js["years"].get(iso[:4], 0) + 1358 out.append(Candidate(it, journal, iso, ev, code, lic, routes))359 return out, js360361362# ---------------------------------------------------------------------------------------------------------363# Acquisition364365366def _fetch_one(pub: PoliteClient, robots: Robots, cand: Candidate, blocked_hosts: dict[str, int]) -> tuple[bytes, str, str, str] | tuple[None, str, None, None]:367 """Try the routes in order; return (data, kind, url, route) or (None, reason, None, None)."""368 reasons = []369 for url, kind, route in cand.routes:370 host = urlparse(url).netloc371 if blocked_hosts.get(host, 0) >= 3:372 reasons.append(f"{host}: host blocked earlier")373 continue374 if not robots.allowed(url):375 reasons.append(f"{url}: robots.txt disallows")376 continue377 try:378 r = pub.get(url, check_robots=False, accept="application/pdf" if kind == "pdf" else "application/xml")379 except Exception as e: # noqa: BLE001380 reasons.append(f"{url}: {e!r}"[:200])381 continue382 if r.status_code != 200:383 reasons.append(f"{url}: HTTP {r.status_code}")384 if r.status_code in (401, 403, 202):385 blocked_hosts[host] = blocked_hosts.get(host, 0) + 1386 continue387 data = r.content388 if len(data) > MAX_FILE_BYTES:389 reasons.append(f"{url}: too large ({len(data)} B)")390 continue391 if kind == "pdf":392 if not data.lstrip()[:5].startswith(b"%PDF"):393 reasons.append(f"{url}: not a PDF ({r.headers.get('content-type')})")394 blocked_hosts[host] = blocked_hosts.get(host, 0) + 1395 continue396 blocked_hosts[host] = 0397 return data, "pdf", str(r.url), route398 head = data[:4000].decode("utf-8", "replace")399 if re.search(r"<article[\s>]", head) and "<html" not in head.lower()[:500]:400 blocked_hosts[host] = 0401 return data, "jats", str(r.url), route402 reasons.append(f"{url}: XML is not JATS")403 return None, "; ".join(reasons) or "no route", None, None404405406def _subject_area(journal: dict) -> str:407 return journal.get("subject_area") or "economics"408409410def write_stub(out_dir: Path, cand: Candidate, data: bytes, fmt: str, url: str, route: str) -> dict:411 it = cand.item412 doi = it["DOI"].lower()413 stem = sanitize_doi(doi)414 ext = "pdf" if fmt == "pdf" else "xml"415 raw_path = out_dir / f"{stem}.{ext}"416 atomic_write_bytes(raw_path, data)417 authors = author_names(it)418 lic = LicenseRecord(code=cand.code, uri=cand.lic_entry.get("URL"), evidence="crossref license[].URL", raw=json.dumps(cand.lic_entry, sort_keys=True))419 notes = [f"full-text route: {route}", f"license content-version={cand.lic_entry.get('content-version')}"]420 if "NC" in cand.code:421 notes.append(f"non-commercial licence ({cand.code}): research use only, derivatives released under the same terms")422 issn = cand.journal["issn"] if isinstance(cand.journal["issn"], list) else [cand.journal["issn"]]423 stub = {424 "doc_id": short_id("doc", "crossref", doi),425 "source": "crossref",426 "source_ids": {"doi": doi, "issn": ",".join(issn)},427 "title": " ".join(it.get("title") or []) or None,428 "authors": authors,429 "author_group": author_group(authors),430 "venue": (it.get("container-title") or [cand.journal.get("name")])[0],431 "publisher": it.get("publisher") or cand.journal.get("publisher"),432 "pub_date": cand.pub_date,433 "pub_year": int(cand.pub_date[:4]),434 "pub_date_evidence": cand.evidence,435 "subject_area": _subject_area(cand.journal),436 "subject_raw": list(it.get("subject") or []),437 "doc_type": "article",438 "language": it.get("language"),439 "raw_format": fmt,440 "license": lic.to_dict(),441 "raw_path": str(raw_path),442 "raw_sha256": sha256_bytes(data),443 "raw_bytes": len(data),444 "acquired_at": utc_now(),445 "source_url": url,446 "provider_meta": trim_item(it),447 "notes": notes,448 }449 write_json(out_dir / f"{stem}.meta.json", stub)450 return stub451452453def _new_stats() -> dict:454 return {"records_seen": 0, "not_journal_article": 0, "no_license": 0, "license_excluded": 0, "not_pre2018": 0, "no_fulltext": 0, "already_present": 0, "fetched": 0, "fetch_errors": 0, "by_license": {}, "by_journal": {}, "by_route": {}, "screen": {}}455456457def count_crossref(journals_cfg: dict | str | Path, allowed_codes: set[str], until: str = "2017-12-31", client: PoliteClient | None = None) -> dict:458 """Exact per-journal counts of pre-cutoff, CC-licensed (allowed) records with a full-text route.459 Metadata only — nothing is downloaded."""460 client = client or crossref_client()461 stats = _new_stats()462 for j in load_journals(journals_cfg):463 issns = j["issn"] if isinstance(j["issn"], list) else [j["issn"]]464 try:465 total_all = works_total(client, issns, until, has_license=False)466 total_lic = works_total(client, issns, until, has_license=True)467 cands, js = screen_journal(client, j, allowed_codes, until, stats)468 except Exception as e: # noqa: BLE001469 stats["screen"][j["name"]] = {"error": repr(e)[:300]}470 continue471 js.update({"crossref_total_until": total_all, "crossref_total_until_has_license": total_lic, "issn": issns})472 if cands:473 yrs = sorted(js["years"])474 js["first_year"], js["last_year"] = int(yrs[0]), int(yrs[-1])475 stats["screen"][j["name"]] = js476 for code, n in js["by_license"].items():477 stats["by_license"][code] = stats["by_license"].get(code, 0) + n478 stats["eligible_total"] = sum(v.get("eligible", 0) for v in stats["screen"].values())479 stats["http"] = client.stats480 return stats481482483def discover_crossref(484 out_dir: Path,485 journals_cfg: dict | str | Path,486 allowed_codes: set[str],487 until: str = "2017-12-31",488 max_docs: int | None = None,489 seed: int = 1,490 per_journal_cap: int | None = None,491 client: PoliteClient | None = None,492 log_path: Path | None = None,493 fetch_client: PoliteClient | None = None,494) -> dict:495 """Screen every configured journal, sample deterministically, and download full texts + manifest stubs.496497 ``client`` talks to api.crossref.org (mailto polite pool); ``fetch_client`` (no contact address in the user498 agent, ≥ 2 s per publisher host) downloads publisher files. Journals are visited round-robin so that a run499 with ``max_docs`` spreads across journals; ``per_journal_cap`` bounds documents per journal (existing files500 count towards the caps, so re-running the same command is a no-op)."""501 client = client or crossref_client()502 pub = fetch_client or publisher_client()503 robots = Robots(pub)504 out_dir = ensure_dir(out_dir)505 ledger = out_dir / "_ledger.jsonl"506 stats = _new_stats()507 queues: list[tuple[str, list[Candidate]]] = []508 for j in load_journals(journals_cfg):509 issns = j["issn"] if isinstance(j["issn"], list) else [j["issn"]]510 try:511 cands, js = screen_journal(client, j, allowed_codes, until, stats)512 except Exception as e: # noqa: BLE001513 stats["screen"][j["name"]] = {"error": repr(e)[:300]}514 if log_path:515 append_jsonl(log_path, {"journal": j["name"], "error": repr(e)[:300]})516 continue517 stats["screen"][j["name"]] = js518 cands.sort(key=lambda c: c.item["DOI"].lower())519 random.Random(f"{seed}:{issns[0]}").shuffle(cands)520 queues.append((j["name"], cands))521 done_per_journal: dict[str, int] = {}522 blocked_hosts: dict[str, int] = {}523 total = 0524 active = True525 while active and (not max_docs or total < max_docs):526 active = False527 for name, q in queues:528 if max_docs and total >= max_docs:529 break530 if per_journal_cap and done_per_journal.get(name, 0) >= per_journal_cap:531 continue532 while q:533 cand = q.pop(0)534 doi = cand.item["DOI"].lower()535 stem = sanitize_doi(doi)536 if (out_dir / f"{stem}.meta.json").exists():537 stats["already_present"] += 1538 done_per_journal[name] = done_per_journal.get(name, 0) + 1539 total += 1540 active = True541 break542 data, kind, url, route = _fetch_one(pub, robots, cand, blocked_hosts)543 if data is None:544 stats["fetch_errors"] += 1545 append_jsonl(ledger, {"doi": doi, "journal": name, "status": "error", "reason": kind, "at": utc_now()})546 if log_path:547 append_jsonl(log_path, {"doi": doi, "journal": name, "error": kind})548 continue549 write_stub(out_dir, cand, data, kind, url, route)550 append_jsonl(ledger, {"doi": doi, "journal": name, "status": "fetched", "format": kind, "route": route, "url": url, "bytes": len(data), "at": utc_now()})551 stats["fetched"] += 1552 stats["by_license"][cand.code] = stats["by_license"].get(cand.code, 0) + 1553 stats["by_journal"][name] = stats["by_journal"].get(name, 0) + 1554 stats["by_route"][route] = stats["by_route"].get(route, 0) + 1555 done_per_journal[name] = done_per_journal.get(name, 0) + 1556 total += 1557 active = True558 break559 stats["blocked_hosts"] = {h: n for h, n in blocked_hosts.items() if n}560 stats["robots_cache"] = {h: (v if isinstance(v, bool) else f"{len(v)} rules") for h, v in robots.cache.items()}561 stats["http_crossref"] = client.stats562 stats["http_publishers"] = pub.stats563 return stats564