pdf_crawler.py

"""
pdf_crawler.py — source-discovery crawler for the AIM Composites Materials
Database (component 1 of the autonomous-ingestion architecture, Section 6
of the project report).

Finds and downloads PDFs about thermoplastic composites from:

  1. OpenAlex          — open-access papers; full-text search per query plus
                         a 1-credit topic walk (set OPENALEX_API_KEY: since
                         2026 OpenAlex bills per call, anonymous = $0.10/day
                         per IP, free key = $1/day)
  2. Semantic Scholar  — bulk search, up to 1,000 OA-PDF papers per call
                         (no key needed; S2_API_KEY gives a dedicated quota)
  3. arXiv             — preprints (no API key), ANDed-term queries
  4. Unpaywall         — OA-PDF fallback for DOIs OpenAlex has no PDF for
  5. Datasheets        — best-effort depth-1 crawl of manufacturer seed pages,
                         harvesting links to .pdf files (respects robots.txt)
  6. --from-csv FILE   — candidate URLs produced elsewhere, e.g. by
                         matweb_discovery.py (MatWeb-as-index workflow)

Crawl-strategy ideas (HEAD pre-checks, junk-URL filters, per-source yield
tracking) adapted from InDesDatabaseTool `agentic` branch (Frontier.py,
Downloader.py, Discovery.py).

Output:
  <out>/pdfs/*.pdf       downloaded, verified (%PDF magic), sha256-deduped
  <out>/sources.csv      provenance: filename, title, doi, url, year, source, sha256, query
  <out>/state.json       seen URL/hash sets + transient-failure retry counts —
                         reruns resume without re-downloading, and re-attempt
                         URLs that failed transiently (5xx/timeout/bot-block)

The pdfs/ folder feeds directly into batch_ingest.py --input.

Usage:
    python pdf_crawler.py --out ./crawl_out --max-per-query 10
    python pdf_crawler.py --out ./crawl_out --queries "PEEK carbon fiber tensile" "PPS GF30 datasheet"
    python pdf_crawler.py --out ./crawl_out --skip-datasheets
    python pdf_crawler.py --out ./crawl_out --max-total 50
    python pdf_crawler.py --out ./crawl_out --seed-urls https://www.ensingerplastics.com/en/shapes/products

Requires: requests  (pip install requests)

Author: Mathias Heider, AIM Composites project, June 2026.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import logging
import os
import re
import sys
import time
import urllib.parse
import urllib.robotparser
import xml.etree.ElementTree as ET
from dataclasses import dataclass, field, asdict
from html.parser import HTMLParser
from pathlib import Path
from typing import Any, Iterable, Optional

import requests

try:
    import fitz  # PyMuPDF — optional structural PDF validation in download_pdf()
except Exception:  # pragma: no cover - optional dependency
    fitz = None

log = logging.getLogger("pdf_crawler")

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

CONTACT_EMAIL = "mathias.heider@gmail.com"  # used for polite API identification
USER_AGENT = f"AIMCompositesCrawler/0.1 (research; mailto:{CONTACT_EMAIL})"

# Default search intents. Edit freely or override with --queries.
DEFAULT_QUERIES = [
    "thermoplastic composite mechanical properties",
    "carbon fiber PEEK composite tensile properties",
    "glass fiber polypropylene composite properties",
    "carbon fiber PPS laminate mechanical",
    "PEKK composite thermoforming properties",
    "PEI carbon fiber composite processing",
    "PA66 glass fiber composite mechanical properties",
    "thermoplastic prepreg consolidation properties",
]

# Keywords used to score relevance of titles/abstracts. A candidate must
# reach MIN_SCORE to be downloaded.
MATERIAL_KEYWORDS = {
    "thermoplastic": 3, "composite": 2, "laminate": 2, "prepreg": 2,
    "peek": 3, "pekk": 3, "pps": 2, "pei": 2, "paek": 3, "ultem": 2,
    "polypropylene": 2, "polyamide": 2, "pa6": 2, "pa66": 2, "pa12": 2,
    "nylon": 1, "petg": 1, "pet": 1, "abs": 1, "pc": 1,
    "carbon fiber": 2, "carbon fibre": 2, "glass fiber": 2, "glass fibre": 2,
    "cf/": 2, "gf/": 2, "fiber reinforced": 2, "fibre reinforced": 2,
    "tensile": 1, "modulus": 1, "flexural": 1, "strength": 1,
    "datasheet": 2, "mechanical propert": 2, "thermoforming": 1,
    "consolidation": 1, "crystallinity": 1, "matrix": 1,
}
MIN_SCORE = 4

# Best-effort manufacturer/datasheet seed pages (depth-1: fetch page, harvest
# links to PDFs and to same-domain pages that look like datasheet listings).
# Live-checked 2026-08-15 with this crawler's User-Agent (status / ~PDF links
# on the page): victrex 200/228, hexcel 200/138, toraytac 200/39,
# avient advanced-composites 200/7, ensinger 200/1 (JS-rendered finder).
# Re-checked 2026-09-17 through crawl_datasheet_seed (PDF links yielded):
# victrex 40 (cap), hexcel 40 (cap), toraytac 28, avient 17, ensinger 2,
# rtpcompany 34 (new), plasticomp 17 (new, long-fibre thermoplastics).
# Checked and dropped (0 links, JS-rendered finders): lati, lanxess/Tepex,
# teijincarbon, celanese, evonik healthcare, domochemicals, mcam.
# Dropped: solvay.com composites (404 — the business moved to syensqo.com,
# which answers 406 to non-browser clients, so it can't be seeded here).
DEFAULT_SEED_URLS = [
    "https://www.victrex.com/en/datasheets",
    "https://www.hexcel.com/Resources/DataSheets",
    "https://www.toraytac.com/products/thermoplastic",
    "https://www.avient.com/advanced-composites",
    "https://www.ensingerplastics.com/en/plastic-material-selection",
    "https://www.rtpcompany.com/products/product-guide/",
    "https://www.plasticomp.com/",
]

MAX_PDF_BYTES = 40 * 1024 * 1024     # skip anything larger than 40 MB
MIN_PDF_BYTES = 10 * 1024            # smaller is probably junk, not a paper/datasheet
REQUEST_TIMEOUT = 30                  # seconds
PER_DOMAIN_DELAY = 1.5                # politeness delay between hits to a domain
DATASHEET_LINK_LIMIT = 40             # max PDF links harvested per seed page

# Transient-failure retry policy. OpenAlex and (especially) Semantic Scholar's
# free tier return 429s under load, so a request that fails once is often fine
# moments later. Retry connection errors and 429/5xx, honoring Retry-After.
MAX_RETRIES = 3
RETRY_STATUS = frozenset({429, 500, 502, 503, 504})
BACKOFF_BASE = 2.0                    # seconds; exponential per attempt
# Cap on honoring Retry-After. A server answering "Retry-After: 86400" must
# never park a cycle for a day (this cap existed in the Space's Aug 13
# anti-hang hardening and was lost in the Aug 25 crawler sync; the agent's
# stuck cycles all sat in exactly this sleep).
MAX_RETRY_AFTER_S = 300.0

# Per-host circuit breaker. A 429 whose Retry-After exceeds the cap (OpenAlex
# answers "Insufficient budget … Resets at midnight UTC" with Retry-After ≈
# seconds-to-midnight once the daily credit budget is spent) is not a
# transient blip: retrying it MAX_RETRIES times at the capped sleep burned
# 15 min per search call and 60 min per round in Sep 2026, which is exactly
# the watchdog budget. Instead the host is closed until the server's reset
# time and every call to it returns None immediately; the orchestrator logs
# one event per round per closed source. Closure is per process (a container
# restart re-probes), bounded by MAX_SOURCE_CLOSE_S.

# OpenAlex bills per call since 2026 (anonymous $0.10/day per IP, free
# account key $1/day). Set OPENALEX_API_KEY (Space secret) — without it the
# Space shares the anonymous budget of every other tenant on its egress IP.
OPENALEX_API_KEY = os.environ.get("OPENALEX_API_KEY", "").strip()
# Prefer gold/hybrid (publisher OA with a licence) and green (repository
# copies) over "bronze" (free-to-read publisher pages that are JS-gated and
# 403 scripted downloads). Empty string = any is_oa:true work.
OPENALEX_OA_STATUS = os.environ.get("OPENALEX_OA_STATUS", "gold|hybrid|green")
OPENALEX_MAX_PER_PAGE = 200          # API maximum; price is per call, not per row

# Relevance gate relief when a record has no abstract: OpenAlex lacks the
# abstract for a large share of works, and MIN_SCORE=4 on a title alone
# rejects e.g. "Fatigue of carbon/PEEK laminates" (score 5) but also
# "Consolidation quality of AFP-made thermoplastic parts" (score 3).
NO_ABSTRACT_RELIEF = 2

# Hosts that never 403 a scripted client and hold the author/publisher copy
# — ranked ahead of publisher URLs when a work has several OA locations.
REPOSITORY_HOST_HINTS = (
    "arxiv.org", "ncbi.nlm.nih.gov", "europepmc.org", "zenodo.org",
    "hal.science", "hal.archives-ouvertes.fr", "osti.gov", "ntrs.nasa.gov",
    "core.ac.uk", "semanticscholar.org", "researchgate.net", "ssrn.com",
    "dspace", "eprints", "repository", "handle.net", "scholar", "digitalcommons",
    "udspace.udel.edu", "escholarship.org", "pure.", "research-", "openaccess",
)

# URL pre-filter for *academic* candidates (idea from InDesDatabaseTool
# agentic branch, Downloader.is_likely_paper_pdf): skip obvious non-papers
# before spending a download on them.
JUNK_URL_PATTERNS = (
    "syllabus", "application", "template", "newsletter",
    "handbook", "slide", "presentation", "poster", "cv.pdf", "resume",
    "/toc/", "_toc.pdf", "/toc.pdf",      # library tables-of-contents (gbv.de)
)


# ---------------------------------------------------------------------------
# Small helpers
# ---------------------------------------------------------------------------

@dataclass
class Candidate:
    """A discovered PDF candidate, pre-download."""
    title: str
    pdf_url: str
    source: str                       # openalex | arxiv | unpaywall | datasheet
    query: str = ""
    doi: str = ""
    year: str = ""
    abstract: str = ""
    # Alternate OA copies of the same work (other OpenAlex locations).
    # download_pdf() falls back to these when the primary URL fails: the
    # publisher-hosted primary (MDPI/Elsevier/SAGE/...) 403s any scripted
    # client from any IP, while the repository copy downloads fine.
    alt_urls: list = field(default_factory=list)
    # Landing page of the best OA location. Last-resort fallback in
    # download_pdf(): the page's <meta name="citation_pdf_url"> is how
    # Unpaywall and Google Scholar locate PDFs that no API lists directly.
    landing_url: str = ""


# A URL whose download failed for a *transient* reason (5xx, timeout, momentary
# bot-block) is retried on later runs up to this many attempts before it is
# given up on. Deterministic rejections (not a PDF, too large, duplicate
# content, saved OK) go straight into seen_urls and are never retried.
MAX_URL_ATTEMPTS = 3


class CrawlerState:
    """Persisted seen-set so reruns are incremental.

    ``seen_urls``   URLs with a *final* outcome (saved, or deterministically
                    rejected) — never fetched again.
    ``seen_hashes`` sha256 of saved PDF bytes — content dedup.
    ``failed``      {url: attempts} for transient failures; retried on the next
                    run until MAX_URL_ATTEMPTS, then moved to seen_urls. (URLs
                    used to be marked seen *before* the download was attempted,
                    so a ten-minute vendor 503 permanently blacklisted every
                    PDF hit in that window.)
    """

    def __init__(self, path: Path):
        self.path = path
        self.seen_urls: set[str] = set()
        self.seen_hashes: set[str] = set()
        self.failed: dict[str, int] = {}
        # Per-run only (not persisted): URLs already attempted in THIS process.
        # The same pdf_url surfaces several times per run (overlapping queries,
        # several lanes); without this a single outage would burn all
        # MAX_URL_ATTEMPTS in one run and blacklist the URL after all.
        self.attempted: set[str] = set()
        if path.exists():
            try:
                data = json.loads(path.read_text())
                self.seen_urls = set(data.get("seen_urls", []))
                self.seen_hashes = set(data.get("seen_hashes", []))
                self.failed = {str(k): int(v) for k, v in (data.get("failed") or {}).items()}
            except Exception:
                log.warning("Could not parse state file; starting fresh.")

    def mark_final(self, url: str) -> None:
        """Record a final outcome for `url` (saved or deterministically rejected)."""
        self.seen_urls.add(url)
        self.failed.pop(url, None)

    def mark_transient_failure(self, url: str) -> None:
        """Count a transient failure; give up (mark seen) after MAX_URL_ATTEMPTS."""
        n = self.failed.get(url, 0) + 1
        if n >= MAX_URL_ATTEMPTS:
            log.info("Giving up on %s after %d attempts", url, n)
            self.mark_final(url)
        else:
            self.failed[url] = n

    def save(self) -> None:
        self.path.write_text(json.dumps({
            "seen_urls": sorted(self.seen_urls),
            "seen_hashes": sorted(self.seen_hashes),
            "failed": dict(sorted(self.failed.items())),
        }, indent=1))


class _DomainThrottle:
    def __init__(self, delay: float):
        self.delay = delay
        self._last: dict[str, float] = {}

    def wait(self, url: str) -> None:
        dom = urllib.parse.urlparse(url).netloc
        now = time.monotonic()
        last = self._last.get(dom, 0.0)
        if now - last < self.delay:
            time.sleep(self.delay - (now - last))
        self._last[dom] = time.monotonic()


THROTTLE = _DomainThrottle(PER_DOMAIN_DELAY)
SESSION = requests.Session()
SESSION.headers.update({"User-Agent": USER_AGENT})


def _retry_after(r: requests.Response) -> Optional[float]:
    """Seconds from a Retry-After header (delta-seconds form), else None.

    The HTTP-date form is not parsed; callers fall back to exponential
    backoff when this returns None.
    """
    raw = _raw_retry_after(r)
    return None if raw is None else min(raw, MAX_RETRY_AFTER_S)


def _raw_retry_after(r: requests.Response) -> Optional[float]:
    """Uncapped Retry-After seconds (None if absent / not delta-seconds)."""
    val = r.headers.get("Retry-After")
    if not val:
        return None
    try:
        return max(0.0, float(val))
    except ValueError:
        return None


# --- per-host circuit breaker ------------------------------------------------
# A budget refusal is a decision, not a delay. OpenAlex (per-request billing
# since Sep 2026) answers an exhausted daily allowance with 429, a Retry-After
# of many hours and X-RateLimit-Remaining: 0; retrying that three times at
# the 300 s cap cost the agent 15 minutes per query and the whole 60-minute
# cycle budget (runs 211-260, Sep 2026). Such a host is closed until its
# window resets (bounded by MAX_SOURCE_CLOSE_S) and every call to it returns
# None at once, with no request and no sleep, until then. Process-local: a
# container restart reopens everything, which is fine (one cheap probe).
MAX_SOURCE_CLOSE_S = 24 * 3600.0
_CLOSED_HOSTS: dict[str, tuple[float, str]] = {}   # host -> (reopen_epoch, why)


def _host_of(url: str) -> str:
    try:
        return urllib.parse.urlsplit(url).netloc.lower()
    except Exception:
        return ""


def close_source(url_or_host: str, seconds: float, why: str = "") -> None:
    host = _host_of(url_or_host) if "://" in url_or_host else url_or_host.lower()
    if not host:
        return
    seconds = max(1.0, min(float(seconds), MAX_SOURCE_CLOSE_S))
    _CLOSED_HOSTS[host] = (time.time() + seconds, why)
    log.warning("source closed for %.0f min: %s (%s)", seconds / 60.0, host, why)


def source_closed(url_or_host: str) -> Optional[str]:
    """Reason string if the host is closed right now, else None."""
    host = _host_of(url_or_host) if "://" in url_or_host else url_or_host.lower()
    entry = _CLOSED_HOSTS.get(host)
    if not entry:
        return None
    reopen, why = entry
    if time.time() >= reopen:
        _CLOSED_HOSTS.pop(host, None)
        return None
    return f"{why} (reopens in {max(0.0, reopen - time.time()) / 60.0:.0f} min)"


def source_status() -> dict[str, str]:
    """{host: reason} for every host currently closed (expired ones dropped)."""
    return {h: r for h in list(_CLOSED_HOSTS) if (r := source_closed(h))}


def reopen_sources() -> None:
    _CLOSED_HOSTS.clear()


def _budget_refusal(r: requests.Response) -> Optional[tuple[float, str]]:
    """(close_seconds, why) when a 429 means 'budget gone for this window'
    rather than 'slow down': a Retry-After beyond the retry cap, or a
    zero remaining budget (OpenAlex sends both)."""
    if r.status_code != 429:
        return None
    remaining = (r.headers.get("X-RateLimit-Remaining") or "").strip()
    raw = _raw_retry_after(r)
    if remaining == "0" or (raw is not None and raw > MAX_RETRY_AFTER_S):
        secs = raw if raw is not None else 3600.0
        why = (f"429 budget refusal: retry-after {secs / 3600.0:.1f} h"
               + (", remaining 0" if remaining == "0" else ""))
        # The server's own words ("Insufficient budget ... Resets at midnight
        # UTC") are the most useful thing the event log can carry.
        try:
            body = r.json() if "json" in (r.headers.get("Content-Type") or "") else {}
        except ValueError:
            body = {}
        msg = (body or {}).get("message") or (body or {}).get("error") or ""
        if msg:
            why += f": {str(msg)[:160]}"
        return max(60.0, secs), why
    return None


def http_get(url: str, **kw) -> Optional[requests.Response]:
    """Polite GET with throttling and bounded retry/backoff.

    Retries connection errors and 429/5xx responses up to MAX_RETRIES times,
    honoring a Retry-After header when present. Returns None on a
    non-retryable status or once retries are exhausted.

    A 429 whose Retry-After exceeds MAX_RETRY_AFTER_S (a spent daily budget,
    not a momentary limit) closes the host until that time instead of
    retrying — see close_source(). A closed host returns None at once.
    """
    closed = source_closed(url)
    if closed:
        log.debug("GET %s skipped: source closed, %s", url, closed)
        return None
    for attempt in range(MAX_RETRIES + 1):
        try:
            THROTTLE.wait(url)
            r = SESSION.get(url, timeout=REQUEST_TIMEOUT, **kw)
            if r.status_code == 200:
                return r
            refusal = _budget_refusal(r)
            if refusal:
                close_source(url, *refusal)
                return None
            if r.status_code in RETRY_STATUS and attempt < MAX_RETRIES:
                delay = _retry_after(r) or BACKOFF_BASE * (2 ** attempt)
                log.debug("GET %s -> %s; retry %d/%d in %.1fs",
                          url, r.status_code, attempt + 1, MAX_RETRIES, delay)
                time.sleep(delay)
                continue
            log.debug("GET %s -> %s", url, r.status_code)
            return None
        except requests.RequestException as e:
            if attempt < MAX_RETRIES:
                delay = BACKOFF_BASE * (2 ** attempt)
                log.debug("GET %s failed: %s; retry %d/%d in %.1fs",
                          url, e, attempt + 1, MAX_RETRIES, delay)
                time.sleep(delay)
                continue
            log.debug("GET %s failed: %s", url, e)
            return None
    return None


def _json(r: Optional[requests.Response]) -> dict:
    """Parse a JSON body defensively; returns {} on None or non-JSON.

    Some endpoints answer 200 with an HTML error/challenge page, which would
    otherwise raise inside .json() and abort the whole query.
    """
    if r is None:
        return {}
    try:
        return r.json() or {}
    except ValueError:
        log.debug("Non-JSON 200 body from %s", r.url)
        return {}


def relevance_score(text: str) -> int:
    t = text.lower()
    return sum(w for k, w in MATERIAL_KEYWORDS.items() if k in t)


def effective_min_score(min_score: int, abstract: str) -> int:
    """Gate threshold for a record: `min_score`, relaxed by NO_ABSTRACT_RELIEF
    (floor 1) when there is no abstract to score against."""
    if abstract and abstract.strip():
        return min_score
    return max(1, min_score - NO_ABSTRACT_RELIEF)


def _passes_gate(title: str, abstract: str, pdf_url: str, min_score: int) -> bool:
    """The junk-URL + relevance gate, applied INSIDE each search lane.

    The lanes over-fetch (min(limit*5, 100)) precisely because this gate
    discards a good share of hits; counting toward `limit` before applying it
    made the over-fetch a no-op (the first `limit` entries were yielded and
    then filtered in main(), exactly as if nothing had been over-fetched).
    """
    if looks_like_junk(pdf_url):
        return False
    return relevance_score(title + " " + abstract) >= effective_min_score(min_score, abstract)


def slugify(title: str, maxlen: int = 70) -> str:
    s = re.sub(r"[^a-zA-Z0-9]+", "_", title).strip("_").lower()
    return s[:maxlen] or "untitled"


# ---------------------------------------------------------------------------
# Source 1: OpenAlex
# ---------------------------------------------------------------------------

OPENALEX_HOST = "api.openalex.org"


def _openalex_common_params() -> str:
    """mailto (polite pool) + api_key (daily budget) for every OpenAlex call."""
    q = f"&mailto={CONTACT_EMAIL}"
    if OPENALEX_API_KEY:
        q += f"&api_key={urllib.parse.quote(OPENALEX_API_KEY)}"
    return q


def _openalex_oa_filter() -> str:
    f = "open_access.is_oa:true"
    if OPENALEX_OA_STATUS:
        f += f",open_access.oa_status:{OPENALEX_OA_STATUS}"
    return f


def _openalex_per_page(limit: int) -> int:
    # Over-fetch for the relevance gate; the call costs the same at 200 rows.
    return max(25, min(limit * 5, OPENALEX_MAX_PER_PAGE))


def _openalex_abstract(work: dict) -> str:
    """Reconstruct the abstract from OpenAlex's inverted index."""
    inv = work.get("abstract_inverted_index")
    if not inv:
        return ""
    pos: dict[int, str] = {}
    for word, idxs in inv.items():
        for i in idxs:
            pos[i] = word
    return " ".join(pos[i] for i in sorted(pos))[:2000]


def _is_repository_location(loc: dict) -> bool:
    src = (loc or {}).get("source") or {}
    if (src.get("type") or "").lower() == "repository":
        return True
    u = ((loc or {}).get("pdf_url") or (loc or {}).get("landing_page_url") or "").lower()
    return any(h in u for h in REPOSITORY_HOST_HINTS)


def _rank_oa_urls(work: dict) -> tuple[list[str], str]:
    """Order a work's OA PDF URLs repository-first; return (urls, landing_url).

    Publisher-hosted PDFs (MDPI/Elsevier/SAGE/Wiley/…) 403 any scripted client
    from any IP; the accepted/published copy in a repository (PMC, arXiv,
    HAL, university IR) downloads fine. So: repository copies first (published
    version before accepted), then the publisher's best_oa_location, then
    everything else. The landing page returned is the first location's, for
    download_pdf()'s citation_pdf_url fallback.
    """
    best = work.get("best_oa_location") or {}
    locs = [best] + [l for l in (work.get("locations") or []) if l]
    ranked: list[tuple[int, str, str]] = []
    seen: set[str] = set()
    for loc in locs:
        u = (loc or {}).get("pdf_url")
        if not u or u in seen or looks_like_junk(u):
            continue
        seen.add(u)
        version = (loc.get("version") or "").lower()
        if _is_repository_location(loc):
            rank = 0 if version == "publishedversion" else 1
        elif loc is best:
            rank = 2
        else:
            rank = 3
        ranked.append((rank, u, loc.get("landing_page_url") or ""))
    ranked.sort(key=lambda t: t[0])
    urls = [u for _, u, _ in ranked]
    landing = next((l for _, _, l in ranked if l), "") or (best.get("landing_page_url") or "")
    return urls, landing


def _openalex_work_to_candidate(work: dict, query: str, min_score: int,
                                source: str = "openalex") -> Optional[Candidate]:
    title = work.get("display_name") or ""
    doi = (work.get("doi") or "").replace("https://doi.org/", "")
    abstract = _openalex_abstract(work)
    # Gate on relevance before counting toward `limit`.
    if relevance_score(title + " " + abstract) < effective_min_score(min_score, abstract):
        return None
    urls, landing = _rank_oa_urls(work)
    if not urls and not doi:
        return None                       # nothing to download, nothing to resolve
    # NOTE: no Unpaywall lookup here. Discovery used to resolve
    # missing-PDF works via the (throttled) Unpaywall API one DOI at a
    # time, which scaled with page size and stalled whole cycles.
    # download_pdf() already consults Unpaywall by DOI as its last
    # resort, and only for candidates that are actually attempted.
    return Candidate(
        title=title, pdf_url=urls[0] if urls else "", source=source, query=query,
        doi=doi, year=str(work.get("publication_year") or ""),
        abstract=abstract, alt_urls=urls[1:6], landing_url=landing,
    )


def search_openalex(query: str, limit: int, min_score: int = MIN_SCORE) -> Iterable[Candidate]:
    """Full-text search lane (10 OpenAlex credits per call)."""
    url = (
        "https://api.openalex.org/works"
        f"?search={urllib.parse.quote(query)}"
        f"&filter={_openalex_oa_filter()}"
        f"&per-page={_openalex_per_page(limit)}"
        + _openalex_common_params()
    )
    r = http_get(url)
    if r is None:
        log.warning("OpenAlex unreachable for query %r%s", query,
                    f" — {source_closed(OPENALEX_HOST)}" if source_closed(OPENALEX_HOST) else "")
        return
    n = 0
    for work in _json(r).get("results", []):
        if n >= limit:
            break
        cand = _openalex_work_to_candidate(work, query, min_score)
        if cand is None:
            continue
        yield cand
        n += 1


def resolve_openalex_topics(names: Iterable[str]) -> list[dict]:
    """Map topic display names / keywords to OpenAlex topic ids (1 credit each).

    Returns [{"id": "T10521", "display_name": ..., "query": name}] for the
    best match of each name (highest works_count among results whose
    display_name or keywords mention every word of the name).
    """
    out: list[dict] = []
    for name in names:
        name = (name or "").strip()
        if not name:
            continue
        r = http_get("https://api.openalex.org/topics"
                     f"?search={urllib.parse.quote(name)}&per-page=25"
                     + _openalex_common_params())
        if r is None:
            continue
        words = [w for w in re.split(r"\W+", name.lower()) if len(w) > 2]
        best, best_n = None, -1
        for t in _json(r).get("results", []):
            hay = " ".join([t.get("display_name") or "",
                            " ".join(t.get("keywords") or []),
                            t.get("description") or ""]).lower()
            if not all(w in hay for w in words):
                continue
            n = int(t.get("works_count") or 0)
            if n > best_n:
                best, best_n = t, n
        if best:
            tid = (best.get("id") or "").rsplit("/", 1)[-1]
            out.append({"id": tid, "display_name": best.get("display_name"),
                        "query": name, "works_count": best_n})
    return out


def list_openalex_topic(topic_id: str, per_page: int, cursor: str = "*",
                        min_score: int = MIN_SCORE, query: str = "",
                        sort: str = "cited_by_count:desc") -> tuple[list[Candidate], Optional[str]]:
    """One page of OA works in an OpenAlex topic — a pure filter call, which
    costs 1 credit (a `search=` call costs 10). Returns (candidates,
    next_cursor); next_cursor is None when the topic is exhausted. Persist the
    cursor between cycles to walk the topic deeper each time.
    """
    url = ("https://api.openalex.org/works"
           f"?filter={_openalex_oa_filter()},topics.id:{urllib.parse.quote(topic_id)}"
           f"&sort={urllib.parse.quote(sort)}"
           f"&per-page={max(25, min(per_page, OPENALEX_MAX_PER_PAGE))}"
           f"&cursor={urllib.parse.quote(cursor or '*')}"
           + _openalex_common_params())
    r = http_get(url)
    if r is None:
        return [], cursor or "*"          # keep the cursor; retry next cycle
    data = _json(r)
    cands: list[Candidate] = []
    for work in data.get("results", []):
        cand = _openalex_work_to_candidate(work, query or f"topic:{topic_id}", min_score,
                                           source="openalex")
        if cand is not None:
            cands.append(cand)
    nxt = (data.get("meta") or {}).get("next_cursor")
    return cands, (nxt or None)


def unpaywall_pdf_url(doi: str) -> Optional[str]:
    r = http_get(f"https://api.unpaywall.org/v2/{urllib.parse.quote(doi)}"
                 f"?email={CONTACT_EMAIL}")
    loc = _json(r).get("best_oa_location") or {}
    return loc.get("url_for_pdf")


# ---------------------------------------------------------------------------
# Source 2: Semantic Scholar (free tier, no key; ~100 req / 5 min)
# ---------------------------------------------------------------------------

S2_HOST = "api.semanticscholar.org"
S2_API = "https://api.semanticscholar.org/graph/v1/paper/search"
# Bulk search: up to 1,000 papers per call, filterable to papers that have an
# open-access PDF URL (the `openAccessPdf` flag) — one call per query instead
# of paging 100 at a time, and every hit is downloadable by construction.
# Verified 2026-09-17: "thermoplastic composite tensile" → 1,276 OA-PDF
# papers, 87 % with abstracts, sorted by citation count.
S2_BULK_API = "https://api.semanticscholar.org/graph/v1/paper/search/bulk"
S2_FIELDS = "title,abstract,year,externalIds,openAccessPdf"
S2_BULK_SORT = "citationCount:desc"
# The unauthenticated S2 endpoint shares one tiny global rate bucket and 429s
# almost constantly. A free key (https://www.semanticscholar.org/product/api)
# gives a dedicated quota; set S2_API_KEY in the env or pass --s2-api-key.
S2_API_KEY = os.environ.get("S2_API_KEY") or os.environ.get("SEMANTIC_SCHOLAR_API_KEY") or ""


def _s2_headers() -> Optional[dict]:
    return {"x-api-key": S2_API_KEY} if S2_API_KEY else None


def _s2_candidates(papers: Iterable[dict], query: str, min_score: int) -> list[tuple[int, Candidate]]:
    """Gate S2 paper records; returns (relevance score, candidate) pairs."""
    out: list[tuple[int, Candidate]] = []
    for p in papers or []:
        ext = p.get("externalIds") or {}
        if "ArXiv" in ext and not ext.get("DOI"):
            continue                       # arXiv lane already covers these
        oa = p.get("openAccessPdf") or {}
        pdf_url = oa.get("url")
        if not pdf_url:
            continue
        title = p.get("title") or ""
        abstract = (p.get("abstract") or "")[:2000]
        if not _passes_gate(title, abstract, pdf_url, min_score):
            continue
        out.append((relevance_score(title + " " + abstract), Candidate(
            title=title, pdf_url=pdf_url,
            source="semanticscholar", query=query,
            doi=str(ext.get("DOI") or ""), year=str(p.get("year") or ""),
            abstract=abstract,
        )))
    return out


def search_semantic_scholar(query: str, limit: int, min_score: int = MIN_SCORE) -> Iterable[Candidate]:
    """Semantic Scholar bulk lane: one call → up to 1,000 papers with an OA
    PDF; the `limit` most relevant (by our keyword score) are yielded."""
    r = http_get(S2_BULK_API, params={"query": query, "openAccessPdf": "",
                                      "fields": S2_FIELDS, "sort": S2_BULK_SORT},
                 headers=_s2_headers())
    if r is None:
        log.warning("Semantic Scholar unreachable for query %r%s", query,
                    "" if S2_API_KEY else " (no API key — set S2_API_KEY to avoid 429s)")
        return
    scored = _s2_candidates(_json(r).get("data") or [], query, min_score)
    scored.sort(key=lambda t: -t[0])       # stable: ties keep citation order
    for _score, cand in scored[:limit]:
        yield cand


# ---------------------------------------------------------------------------
# Source 2b: NASA Technical Reports Server (NTRS)
#
# Free, unauthenticated JSON API, no rate limiting or paywall, direct PDF
# links; Langley's PEEK/PEKK/AFP/welding/thermoforming reports are squarely
# in scope (480 hits for "thermoplastic composite" on 17 Sep 2026). Records
# usually have no DOI, so the registry key is "ntrs:<id>" (stable across
# filename changes); the Unpaywall fallback is skipped for such keys.
# ---------------------------------------------------------------------------

NTRS_API = "https://ntrs.nasa.gov/api/citations/search"
NTRS_BASE = "https://ntrs.nasa.gov"
# Slide decks are bullet fragments: rows fail grounding and add nothing a
# paper or report does not; skip them unless a caller opts in.
NTRS_SKIP_STI_TYPES = {"PRESENTATION"}


def _ntrs_pdf_url(rec: dict) -> str:
    """First download entry with a PDF link, absolute; '' if none."""
    for d in rec.get("downloads") or []:
        links = d.get("links") or {}
        path = links.get("pdf") or links.get("original") or ""
        if path and path.lower().endswith(".pdf"):
            return path if path.startswith("http") else NTRS_BASE + path
    return ""


def _ntrs_year(rec: dict) -> str:
    for pub in rec.get("publications") or []:
        d = pub.get("publicationDate") or ""
        if d[:4].isdigit():
            return d[:4]
    for key in ("distributionDate", "created"):
        d = rec.get(key) or ""
        if d[:4].isdigit():
            return d[:4]
    return ""


def search_ntrs(query: str, limit: int, min_score: int = MIN_SCORE,
                skip_sti_types: Optional[set] = None) -> Iterable[Candidate]:
    """NTRS citations search -> Candidates with direct PDF links.

    Over-fetches like the other lanes and applies the junk-URL + relevance
    gate inside the loop, so `limit` counts relevant candidates."""
    skip = NTRS_SKIP_STI_TYPES if skip_sti_types is None else skip_sti_types
    r = http_get(NTRS_API, params={"q": query, "page[size]": min(limit * 5, 100),
                                   "downloadsAvailable": "true"})
    if r is None:
        log.warning("NTRS unreachable for query %r", query)
        return
    results = _json(r).get("results")
    if not isinstance(results, list):
        return                                  # HTML challenge page or schema change
    n = 0
    for rec in results:
        if n >= limit:
            break
        if not isinstance(rec, dict):
            continue
        if (rec.get("stiType") or "").upper() in skip:
            continue
        pdf_url = _ntrs_pdf_url(rec)
        if not pdf_url:
            continue
        title = rec.get("title") or ""
        abstract = (rec.get("abstract") or "")[:2000]
        if not _passes_gate(title, abstract, pdf_url, min_score):
            continue
        doi = ""
        for pub in rec.get("publications") or []:
            if pub.get("doi"):
                doi = str(pub["doi"]).replace("https://doi.org/", "").strip()
                break
        rec_id = str(rec.get("id") or "").strip()
        yield Candidate(
            title=title, pdf_url=pdf_url, source="ntrs", query=query,
            doi=doi or (f"ntrs:{rec_id}" if rec_id else ""),
            year=_ntrs_year(rec), abstract=abstract,
        )
        n += 1


# ---------------------------------------------------------------------------
# Source: UDSpace (University of Delaware's DSpace 7 repository)
# ---------------------------------------------------------------------------
# CCM theses and dissertations on thermoplastic composites: dense property
# tables, fully open, no bot-blocking, no key. Discovery via the REST
# discover endpoint; the PDF is the ORIGINAL bundle's first .pdf bitstream
# (two more small calls per item, made only for items that pass the gate).

UDSPACE_HOST = "udspace.udel.edu"
UDSPACE_API = f"https://{UDSPACE_HOST}/server/api"


def _dspace_meta(item: dict, key: str) -> str:
    vals = ((item.get("metadata") or {}).get(key)) or []
    for v in vals:
        if isinstance(v, dict) and v.get("value"):
            return str(v["value"])
    return ""


def _udspace_pdf_url(item_uuid: str) -> str:
    """First PDF bitstream of the item's ORIGINAL bundle, '' if none."""
    r = http_get(f"{UDSPACE_API}/core/items/{item_uuid}/bundles", params={"size": 20})
    bundles = ((_json(r).get("_embedded") or {}).get("bundles")) or []
    original = [b for b in bundles if (b.get("name") or "").upper() == "ORIGINAL"] or bundles
    for b in original[:2]:
        buuid = b.get("uuid")
        if not buuid:
            continue
        rb = http_get(f"{UDSPACE_API}/core/bundles/{buuid}/bitstreams", params={"size": 20})
        bits = ((_json(rb).get("_embedded") or {}).get("bitstreams")) or []
        for bit in bits:
            name = (bit.get("name") or "").lower()
            fmt = (bit.get("format") or {}) if isinstance(bit.get("format"), dict) else {}
            href = ((bit.get("_links") or {}).get("content") or {}).get("href") or ""
            if href and (name.endswith(".pdf") or "pdf" in (fmt.get("mimetype") or "").lower()):
                return href
    return ""


def search_udspace(query: str, limit: int, min_score: int = MIN_SCORE) -> Iterable[Candidate]:
    """UDSpace discover search -> Candidates with direct bitstream links."""
    r = http_get(f"{UDSPACE_API}/discover/search/objects",
                 params={"query": query, "dsoType": "item", "size": min(limit * 3, 50),
                         "sort": "score,DESC"})
    if r is None:
        log.warning("UDSpace unreachable for query %r", query)
        return
    result = (_json(r).get("_embedded") or {}).get("searchResult") or {}
    objects = (result.get("_embedded") or {}).get("objects") or []
    n = 0
    for obj in objects:
        if n >= limit:
            break
        item = ((obj.get("_embedded") or {}).get("indexableObject")) or {}
        uuid_ = item.get("uuid") or ""
        if not uuid_ or (item.get("type") or "item") != "item":
            continue
        title = item.get("name") or _dspace_meta(item, "dc.title")
        abstract = _dspace_meta(item, "dc.description.abstract")[:2000]
        if relevance_score(title + " " + abstract) < effective_min_score(min_score, abstract):
            continue
        pdf_url = _udspace_pdf_url(uuid_)
        if not pdf_url or looks_like_junk(pdf_url):
            continue
        handle = _dspace_meta(item, "dc.identifier.uri")
        year = _dspace_meta(item, "dc.date.issued")[:4]
        yield Candidate(
            title=title, pdf_url=pdf_url, source="udspace", query=query,
            doi=(handle.replace("https://", "").replace("http://", "") if handle else f"udspace:{uuid_}"),
            year=year if year.isdigit() else "", abstract=abstract, landing_url=handle,
        )
        n += 1


# ---------------------------------------------------------------------------
# Source: CORE (core.ac.uk) — hosts the full texts itself, so its download
# URLs never 403 a scripted client. Needs a (free) API key: CORE_API_KEY.
# ---------------------------------------------------------------------------

CORE_HOST = "api.core.ac.uk"
CORE_API = f"https://{CORE_HOST}/v3"
CORE_API_KEY = os.environ.get("CORE_API_KEY", "").strip()


def search_core(query: str, limit: int, min_score: int = MIN_SCORE) -> Iterable[Candidate]:
    """CORE works search -> Candidates whose pdf_url is CORE's own copy."""
    if not CORE_API_KEY:
        return                                  # lane inactive without a key
    r = http_get(f"{CORE_API}/search/works",
                 params={"q": f"({query}) AND _exists_:downloadUrl", "limit": min(limit * 4, 100)},
                 headers={"Authorization": f"Bearer {CORE_API_KEY}"})
    if r is None:
        log.warning("CORE unreachable for query %r", query)
        return
    results = _json(r).get("results")
    if not isinstance(results, list):
        return
    n = 0
    for w in results:
        if n >= limit:
            break
        if not isinstance(w, dict):
            continue
        pdf_url = (w.get("downloadUrl") or "").strip()
        if not pdf_url:
            continue
        title = w.get("title") or ""
        abstract = (w.get("abstract") or "")[:2000]
        if not _passes_gate(title, abstract, pdf_url, min_score):
            continue
        doi = (w.get("doi") or "").replace("https://doi.org/", "").strip()
        cid = str(w.get("id") or "").strip()
        yield Candidate(
            title=title, pdf_url=pdf_url, source="core", query=query,
            doi=doi or (f"core:{cid}" if cid else ""),
            year=str(w.get("yearPublished") or ""), abstract=abstract,
        )
        n += 1


# ---------------------------------------------------------------------------
# Source 3: arXiv
# ---------------------------------------------------------------------------

ARXIV_NS = {"a": "http://www.w3.org/2005/Atom"}

ARXIV_HOST = "export.arxiv.org"
ARXIV_STOPWORDS = frozenset({
    "and", "of", "the", "for", "in", "on", "with", "a", "an", "to", "by",
    "properties", "property", "behavior", "behaviour", "study", "analysis",
    "characterization", "characterisation", "effect", "effects", "based",
})
ARXIV_MAX_TERMS = 3
# Which terms survive when an intent has more than ARXIV_MAX_TERMS words:
# the matrix polymer beats the reinforcement beats "composite" beats the
# property/process word. Unlisted words weigh 1.
_ARXIV_TERM_WEIGHT = {
    **{m: 4 for m in (
        "peek", "pekk", "paek", "pps", "pei", "pa6", "pa66", "pa12", "pa11", "pp",
        "pc", "pet", "tpu", "lcp", "ppa", "psu", "pla", "ultem", "nylon",
        "polypropylene", "polyamide", "polycarbonate", "polyetherimide",
        "polyetheretherketone", "polyphenylene", "polysulfone", "polyester",
        "polyurethane", "polylactic", "polyaryletherketone", "thermoplastic")},
    **{r: 3 for r in ("carbon", "glass", "aramid", "basalt", "flax", "fiber", "fibre",
                      "fabric", "tape", "prepreg", "laminate", "organosheet",
                      "unidirectional", "woven")},
    "composite": 2, "composites": 2, "datasheet": 2,
}
_ARXIV_SPELLINGS = {"fiber": ("fiber", "fibre"), "fibre": ("fiber", "fibre")}


def arxiv_search_query(query: str, max_terms: int = ARXIV_MAX_TERMS) -> str:
    """Build arXiv's `search_query` from a free-text intent.

    The pre-Sep-2026 lane wrapped the whole intent in quotes, i.e. an exact
    phrase search: `all:"thermoplastic composite mechanical properties"`
    matches 0 papers, while `all:thermoplastic AND all:composite AND
    all:mechanical` matches 28 and `all:thermoplastic AND all:composite`
    56 (measured 2026-09-17). So: AND the most specific terms — generic
    words are dropped, and with more than ARXIV_MAX_TERMS terms left the
    highest-weighted (matrix > reinforcement > composite > rest) are kept,
    in their original order. fiber/fibre is searched as either spelling.
    """
    tokens = [t for t in re.split(r"[^a-z0-9/]+", query.lower()) if len(t) > 1]
    terms = [t for t in tokens if t not in ARXIV_STOPWORDS]
    if not terms:
        terms = tokens or [query.strip().lower()]
    terms = list(dict.fromkeys(terms))            # first-seen order, no dupes
    max_terms = max(1, int(max_terms))
    if len(terms) > max_terms:
        ranked = sorted(terms, key=lambda t: (-_ARXIV_TERM_WEIGHT.get(t, 1), -len(t)))
        keep = set(ranked[:max_terms])
        terms = [t for t in terms if t in keep]
    parts = []
    for t in terms:
        alts = _ARXIV_SPELLINGS.get(t)
        parts.append("(" + " OR ".join(f"all:{a}" for a in alts) + ")" if alts else f"all:{t}")
    return " AND ".join(parts)


def _arxiv_entries(search_query: str, max_results: int) -> Optional[list]:
    """One arXiv API call → list of <entry> elements (None if unreachable)."""
    url = (f"https://export.arxiv.org/api/query?search_query={urllib.parse.quote(search_query)}"
           f"&max_results={max_results}&sortBy=relevance")
    r = http_get(url)
    if r is None:
        return None
    try:
        root = ET.fromstring(r.text)
    except ET.ParseError:
        return []
    return root.findall("a:entry", ARXIV_NS)


def search_arxiv(query: str, limit: int, min_score: int = MIN_SCORE) -> Iterable[Candidate]:
    # Over-fetch like OpenAlex/S2 do and apply the relevance gate INSIDE the
    # loop (see _passes_gate) so `limit` counts relevant candidates, not raw
    # entries — asking for exactly `limit` left arXiv under-yielding ~3x.
    max_results = min(limit * 5, 100)
    sq = arxiv_search_query(query)
    entries = _arxiv_entries(sq, max_results)
    if entries is None:
        log.warning("arXiv unreachable for query %r", query)
        return
    if not entries and " AND " in sq:
        # Applied-composites intents are often too specific for arXiv
        # ("pa66 AND glass AND fiber" → 0); one broader retry with two terms.
        sq2 = arxiv_search_query(query, max_terms=ARXIV_MAX_TERMS - 1)
        if sq2 != sq:
            entries = _arxiv_entries(sq2, max_results) or []
    n = 0
    for entry in entries:
        if n >= limit:
            break
        title = re.sub(r"\s+", " ", entry.findtext("a:title", "", ARXIV_NS)).strip()
        abstract = re.sub(r"\s+", " ", entry.findtext("a:summary", "", ARXIV_NS)).strip()
        year = (entry.findtext("a:published", "", ARXIV_NS) or "")[:4]
        pdf_url = ""
        for link in entry.findall("a:link", ARXIV_NS):
            if link.get("title") == "pdf" or link.get("type") == "application/pdf":
                pdf_url = link.get("href", "")
        if pdf_url and _passes_gate(title, abstract, pdf_url, min_score):
            yield Candidate(title=title, pdf_url=pdf_url, source="arxiv",
                            query=query, year=year, abstract=abstract)
            n += 1


# ---------------------------------------------------------------------------
# Source 3: manufacturer datasheet pages (best effort)
# ---------------------------------------------------------------------------

class _LinkExtractor(HTMLParser):
    def __init__(self):
        super().__init__()
        self.links: list[tuple[str, str]] = []   # (href, anchor text)
        self._href: Optional[str] = None
        self._text: list[str] = []

    def handle_starttag(self, tag, attrs):
        if tag == "a":
            self._href = dict(attrs).get("href")
            self._text = []

    def handle_data(self, data):
        if self._href is not None:
            self._text.append(data)

    def handle_endtag(self, tag):
        if tag == "a" and self._href:
            self.links.append((self._href, " ".join(self._text).strip()))
            self._href = None


_ROBOTS_CACHE: dict[str, urllib.robotparser.RobotFileParser] = {}

def robots_allows(url: str) -> bool:
    dom = urllib.parse.urlparse(url)
    base = f"{dom.scheme}://{dom.netloc}"
    rp = _ROBOTS_CACHE.get(base)
    if rp is None:
        rp = urllib.robotparser.RobotFileParser()
        try:
            r = http_get(base + "/robots.txt")
            rp.parse(r.text.splitlines() if r is not None else [])
        except Exception:
            rp.parse([])
        _ROBOTS_CACHE[base] = rp
    try:
        return rp.can_fetch(USER_AGENT, url)
    except Exception:
        return True


def crawl_datasheet_seed(seed_url: str) -> Iterable[Candidate]:
    """Depth-1 crawl: harvest PDF links from the seed page and from
    same-domain subpages whose anchor text looks datasheet-related."""
    if not robots_allows(seed_url):
        log.info("robots.txt disallows %s — skipping", seed_url)
        return
    r = http_get(seed_url)
    if r is None:
        log.warning("Seed unreachable: %s", seed_url)
        return

    def extract(page_url: str, html: str) -> tuple[list[tuple[str, str]], list[str]]:
        p = _LinkExtractor()
        try:
            p.feed(html)
        except Exception:
            pass
        pdfs, subpages = [], []
        dom = urllib.parse.urlparse(page_url).netloc
        for href, text in p.links:
            full = urllib.parse.urljoin(page_url, href)
            u = urllib.parse.urlparse(full)
            if u.scheme not in ("http", "https"):
                continue
            if u.path.lower().endswith(".pdf"):
                pdfs.append((full, text))
            elif (u.netloc == dom and
                  re.search(r"datasheet|data-sheet|tds|product|material|grade",
                            (href + " " + text).lower())):
                subpages.append(full)
        return pdfs, subpages

    pdfs, subpages = extract(seed_url, r.text)
    for sub in subpages[:8]:                      # depth 1, bounded
        if not robots_allows(sub):
            continue
        sr = http_get(sub)
        if sr is not None:
            more, _ = extract(sub, sr.text)
            pdfs.extend(more)

    seen = set()
    for full, text in pdfs[:DATASHEET_LINK_LIMIT]:
        if full in seen:
            continue
        seen.add(full)
        title = text or Path(urllib.parse.urlparse(full).path).stem
        yield Candidate(title=title, pdf_url=full, source="datasheet",
                        query=seed_url)


# ---------------------------------------------------------------------------
# Download + provenance
# ---------------------------------------------------------------------------

def looks_like_junk(url: str) -> bool:
    u = url.lower()
    return any(p in u for p in JUNK_URL_PATTERNS)


# head_precheck outcomes
PRECHECK_OK = "ok"                # go ahead and GET
PRECHECK_REJECT = "reject"        # deterministic: not a PDF / wrong size / gone
PRECHECK_TRANSIENT = "transient"  # blocked or errored right now; worth a retry later


def head_precheck(url: str) -> str:
    """HEAD request before download (agentic-branch Downloader idea): reject
    blocked / wrong-type / wrong-sized URLs without spending bandwidth on a
    GET. Permissive on errors — many servers mishandle HEAD.

    Returns PRECHECK_OK / PRECHECK_REJECT / PRECHECK_TRANSIENT so the caller
    can tell "never a PDF" (mark seen) from "blocked right now" (retry later).
    """
    try:
        THROTTLE.wait(url)
        h = SESSION.head(url, timeout=10, allow_redirects=True)
        if h.status_code in (404, 410):
            log.debug("HEAD %s missing: %s", h.status_code, url)
            return PRECHECK_REJECT
        if h.status_code in (401, 403, 429) or h.status_code >= 500:
            # Paywalled or bot-blocked (MDPI/SAGE/Wiley/Hindawi do this); the
            # GET would return the same block page right now, so don't bother
            # — but blocks and 5xx are often momentary, so don't blacklist.
            log.debug("HEAD %s blocked/unavailable: %s", h.status_code, url)
            return PRECHECK_TRANSIENT
        ctype = (h.headers.get("Content-Type") or "").lower()
        if ctype.startswith("text/html"):
            log.debug("HEAD content-type %s, not a PDF: %s", ctype, url)
            return PRECHECK_REJECT
        size = int(h.headers.get("Content-Length") or 0)
        if size and not (MIN_PDF_BYTES <= size <= MAX_PDF_BYTES):
            log.debug("HEAD size %d out of range: %s", size, url)
            return PRECHECK_REJECT
    except (requests.RequestException, ValueError):
        pass
    return PRECHECK_OK


def download_pdf(cand: Candidate, pdf_dir: Path, state: CrawlerState) -> Optional[dict]:
    """Fetch, validate and save one candidate; return its provenance row or None.

    Tries the candidate's URL chain in order: the primary pdf_url, any
    alternate OA locations discovered with it, and — when the work has a
    DOI — a fresh Unpaywall lookup as the last resort. A blocked
    publisher-hosted primary therefore no longer sinks the whole candidate
    when a repository hosts a downloadable copy of the same paper.
    """
    tried: list[str] = []
    for url in dict.fromkeys([cand.pdf_url, *cand.alt_urls]):
        if not url:
            continue
        tried.append(url)
        row = _download_url(cand, url, pdf_dir, state)
        if row:
            return row
    if cand.doi and cand.doi.startswith("10."):
        # Only a real DOI can be looked up; NTRS records without one carry a
        # namespaced registry key ("ntrs:<id>") that Unpaywall cannot know.
        alt = unpaywall_pdf_url(cand.doi)
        if alt and alt not in tried:
            tried.append(alt)
            row = _download_url(cand, alt, pdf_dir, state)
            if row:
                return row
    # Last resort: the landing page's citation_pdf_url. Only when every listed
    # URL failed (one extra page fetch per candidate, never per cycle).
    if cand.landing_url and cand.landing_url not in state.seen_urls:
        alt = pdf_url_from_landing_page(cand.landing_url)
        state.mark_final(cand.landing_url)     # one look per landing page, ever
        if alt and alt not in tried:
            return _download_url(cand, alt, pdf_dir, state)
    return None


_META_PDF_RE = re.compile(
    r'<meta[^>]+(?:name|property)\s*=\s*["\'](?:citation_pdf_url|og:pdf)["\'][^>]*'
    r'content\s*=\s*["\']([^"\']+)["\']', re.I)
_META_PDF_RE2 = re.compile(
    r'<meta[^>]+content\s*=\s*["\']([^"\']+)["\'][^>]*'
    r'(?:name|property)\s*=\s*["\']citation_pdf_url["\']', re.I)
_LINK_PDF_RE = re.compile(
    r'<link[^>]+type\s*=\s*["\']application/pdf["\'][^>]*href\s*=\s*["\']([^"\']+)["\']', re.I)


def pdf_url_from_landing_page(landing_url: str) -> Optional[str]:
    """Read `<meta name="citation_pdf_url">` (Highwire tag every repository
    and most publishers emit — it is how Unpaywall/Google Scholar find PDFs)
    or `<link type="application/pdf">` from an article's landing page."""
    if not landing_url or not robots_allows(landing_url):
        return None
    r = http_get(landing_url, allow_redirects=True)
    if r is None:
        return None
    ctype = (r.headers.get("Content-Type") or "").lower()
    if "pdf" in ctype:
        return r.url                       # the landing page IS the PDF
    html = r.text[:400_000]
    for rx in (_META_PDF_RE, _META_PDF_RE2, _LINK_PDF_RE):
        m = rx.search(html)
        if m:
            u = urllib.parse.urljoin(r.url, m.group(1).strip())
            if not looks_like_junk(u):
                return u
    return None


def _download_url(cand: Candidate, url: str, pdf_dir: Path,
                  state: CrawlerState) -> Optional[dict]:
    """Fetch, validate and save one URL of a candidate.

    Outcome bookkeeping (see CrawlerState): a save or a deterministic rejection
    is *final* (URL goes to seen_urls); a transient failure is counted in
    state.failed and the URL is retried on a later run.
    """
    if url in state.seen_urls or url in state.attempted:
        return None
    state.attempted.add(url)

    pre = head_precheck(url)
    if pre == PRECHECK_REJECT:
        state.mark_final(url)
        return None
    if pre == PRECHECK_TRANSIENT:
        state.mark_transient_failure(url)
        return None

    r = http_get(url, stream=True, allow_redirects=True)
    if r is None:
        # http_get already retried 429/5xx/connection errors; a None here is
        # either exhausted retries (transient) or a non-retryable status. Treat
        # 4xx-that-isn't-a-block as final via the HEAD path above; anything
        # that reached here is worth another run.
        state.mark_transient_failure(url)
        return None
    chunks, size = [], 0
    try:
        for chunk in r.iter_content(chunk_size=65536):
            chunks.append(chunk)
            size += len(chunk)
            if size > MAX_PDF_BYTES:
                log.info("Too large, skipping: %s", url)
                state.mark_final(url)
                return None
    except requests.RequestException:
        state.mark_transient_failure(url)
        return None
    data = b"".join(chunks)
    if len(data) < MIN_PDF_BYTES:
        log.debug("Too small (%d B), skipping: %s", len(data), url)
        state.mark_final(url)
        return None
    # Reject anything that isn't a real PDF. Require a "%PDF-" header within the
    # first bytes (tolerating a leading BOM/whitespace), and — when PyMuPDF is
    # available — that the bytes actually open as a non-locked, >=1-page document.
    # This catches paywall/login/CAPTCHA HTML mis-served as application/pdf, which
    # a Content-Type check alone would save and then extract into garbage rows.
    if b"%PDF-" not in data[:1024]:
        log.debug("Not a PDF (no %%PDF- magic): %s", url)
        state.mark_final(url)
        return None
    if fitz is not None:
        try:
            doc = fitz.open(stream=data, filetype="pdf")
            try:
                if doc.needs_pass or doc.page_count < 1:
                    log.debug("Rejected (encrypted/zero-page PDF): %s", url)
                    state.mark_final(url)
                    return None
            finally:
                doc.close()
        except Exception as exc:
            log.debug("Rejected (PyMuPDF cannot open: %s): %s", exc, url)
            state.mark_final(url)
            return None

    sha = hashlib.sha256(data).hexdigest()
    state.mark_final(url)
    if sha in state.seen_hashes:
        log.info("Duplicate content, skipping: %s", cand.title[:60])
        return None
    state.seen_hashes.add(sha)

    fname = f"{cand.source}_{slugify(cand.title)}_{sha[:8]}.pdf"
    (pdf_dir / fname).write_bytes(data)
    log.info("Saved [%s] %s  (%.1f kB)", cand.source, fname, size / 1024)
    return {
        "filename": fname, "title": cand.title, "doi": cand.doi,
        "url": url, "year": cand.year, "source": cand.source,
        "sha256": sha, "query": cand.query, "bytes": size,
    }


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

class _LimitReached(Exception):
    """Raised internally once --max-total downloads have been saved."""


def main(argv: Optional[list[str]] = None) -> int:
    global CONTACT_EMAIL, USER_AGENT, S2_API_KEY, OPENALEX_API_KEY
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--out", default="./crawl_out", help="output directory")
    ap.add_argument("--queries", nargs="*", default=None,
                    help="search queries (default: built-in thermoplastic-composite set)")
    ap.add_argument("--max-per-query", type=int, default=8,
                    help="max PDFs to download per query per source")
    ap.add_argument("--max-total", type=int, default=None, metavar="N",
                    help="stop the whole run after N new PDFs (default: no cap)")
    ap.add_argument("--email", default=None, metavar="ADDR",
                    help="contact email for polite API identification "
                         f"(default: {CONTACT_EMAIL})")
    ap.add_argument("--openalex-api-key", default=None, metavar="KEY",
                    help="OpenAlex API key (else uses $OPENALEX_API_KEY); without one "
                         "OpenAlex shares the anonymous $0.10/day budget of your IP")
    ap.add_argument("--s2-api-key", default=None, metavar="KEY",
                    help="Semantic Scholar API key (else uses $S2_API_KEY); "
                         "without one S2 is heavily rate-limited (429)")
    ap.add_argument("--skip-datasheets", action="store_true",
                    help="skip the manufacturer datasheet crawl")
    ap.add_argument("--skip-academic", action="store_true",
                    help="skip OpenAlex/arXiv")
    ap.add_argument("--seed-urls", nargs="*", default=None,
                    help="datasheet seed pages (default: built-in list)")
    ap.add_argument("--from-csv", default=None, metavar="FILE",
                    help="also download candidates from a CSV with columns "
                         "title,pdf_url[,source,query,doi,year] "
                         "(e.g. produced by matweb_discovery.py)")
    ap.add_argument("--min-score", type=int, default=MIN_SCORE,
                    help="minimum relevance score for academic results")
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args(argv)

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(levelname)s %(message)s")

    if args.email:
        CONTACT_EMAIL = args.email
        USER_AGENT = f"AIMCompositesCrawler/0.1 (research; mailto:{CONTACT_EMAIL})"
        SESSION.headers.update({"User-Agent": USER_AGENT})
    if args.s2_api_key:
        S2_API_KEY = args.s2_api_key
    if args.openalex_api_key:
        OPENALEX_API_KEY = args.openalex_api_key

    out = Path(args.out)
    pdf_dir = out / "pdfs"
    pdf_dir.mkdir(parents=True, exist_ok=True)
    log.info("Output dir: %s  (PDFs -> %s)", out.resolve(), pdf_dir.resolve())
    state = CrawlerState(out / "state.json")
    csv_path = out / "sources.csv"

    fields = ["filename", "title", "doi", "url", "year", "source",
              "sha256", "query", "bytes"]
    new_csv = not csv_path.exists()
    csv_file = csv_path.open("a", newline="", encoding="utf-8")
    writer = csv.DictWriter(csv_file, fieldnames=fields)
    if new_csv:
        writer.writeheader()

    queries = args.queries or DEFAULT_QUERIES
    seeds = args.seed_urls or DEFAULT_SEED_URLS
    downloaded = 0
    yield_by_source: dict[str, int] = {}    # agentic-branch idea: track which
                                            # sources actually produce PDFs

    def save_row(row: Optional[dict]) -> None:
        nonlocal downloaded
        if row:
            writer.writerow(row)
            csv_file.flush()
            downloaded += 1
            yield_by_source[row["source"]] = yield_by_source.get(row["source"], 0) + 1
            if args.max_total and downloaded >= args.max_total:
                raise _LimitReached

    try:
        if args.from_csv:
            log.info("=== candidates from %s ===", args.from_csv)
            with open(args.from_csv, newline="", encoding="utf-8") as f:
                for rec in csv.DictReader(f):
                    if not rec.get("pdf_url"):
                        continue
                    c = Candidate(
                        title=rec.get("title", ""), pdf_url=rec["pdf_url"],
                        source=rec.get("source", "csv"), query=rec.get("query", ""),
                        doi=rec.get("doi", ""), year=rec.get("year", ""))
                    save_row(download_pdf(c, pdf_dir, state))
            state.save()

        if not args.skip_academic:
            for q in queries:
                log.info("=== query: %s ===", q)
                cands = list(search_openalex(q, args.max_per_query, args.min_score))
                cands += list(search_semantic_scholar(q, args.max_per_query, args.min_score))
                cands += list(search_arxiv(q, args.max_per_query, args.min_score))
                for c in cands:
                    if looks_like_junk(c.pdf_url):
                        log.debug("Junk URL filter: %s", c.pdf_url)
                        continue
                    score = relevance_score(c.title + " " + c.abstract)
                    if score < args.min_score:
                        log.debug("Low relevance (%d): %s", score, c.title[:70])
                        continue
                    save_row(download_pdf(c, pdf_dir, state))
                state.save()

        if not args.skip_datasheets:
            for seed in seeds:
                log.info("=== datasheet seed: %s ===", seed)
                for c in crawl_datasheet_seed(seed):
                    # Datasheet anchors are short; score title only, lower bar.
                    if relevance_score(c.title) < 1:
                        continue
                    save_row(download_pdf(c, pdf_dir, state))
                state.save()
    except _LimitReached:
        log.info("Reached --max-total=%d; stopping.", args.max_total)
    finally:
        state.save()
        csv_file.close()

    if yield_by_source:
        log.info("Yield by source: %s",
                 ", ".join(f"{k}={v}" for k, v in sorted(yield_by_source.items())))
    log.info("Done. %d new PDFs in %s — feed this to batch_ingest.py --input %s",
             downloaded, pdf_dir.resolve(), pdf_dir.resolve())
    return 0


if __name__ == "__main__":
    sys.exit(main())