Fulltext Retrieval

#!/usr/bin/env python3
"""
Open-access full-text PDF batch retrieval.

Pipeline: Unpaywall → PMC (Europe PMC REST / OA FTP / web) →
          OpenAlex → Crossref → landing-page scrape.

Usage:
    python fetch_oa.py dois.txt --output pdfs/ --email user@example.com
    python fetch_oa.py dois.txt -o pdfs/ -e user@example.com --verbose
"""

import argparse
import json
import logging
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path

MIN_PDF_BYTES = 10 * 1024
USER_AGENT = "medsci-skills/1.0"

log = logging.getLogger("fetch_oa")


# ============================================================
# Helpers
# ============================================================

def _ua(email: str) -> str:
    """Build a polite User-Agent string with contact email."""
    return f"{USER_AGENT} (mailto:{email})"


def is_valid_pdf(data: bytes) -> bool:
    return data.startswith(b"%PDF-") and len(data) >= MIN_PDF_BYTES


def fetch_bytes(url: str, email: str, accept: str = "*/*",
                timeout: int = 30) -> tuple[bytes, str, str]:
    req = urllib.request.Request(url, headers={
        "User-Agent": _ua(email),
        "Accept": accept,
    })
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return resp.read(), resp.geturl(), resp.headers.get("Content-Type", "")


def save_pdf(data: bytes, path: Path) -> bool:
    if not is_valid_pdf(data):
        return False
    path.write_bytes(data)
    return True


def existing_pdf_ok(path: Path) -> bool:
    if not path.exists():
        return False
    try:
        return is_valid_pdf(path.read_bytes())
    except OSError:
        return False


# ============================================================
# 1. Unpaywall
# ============================================================

def unpaywall_lookup(doi: str, email: str) -> str | None:
    url = f"https://api.unpaywall.org/v2/{urllib.parse.quote(doi, safe='/')}" \
          f"?email={urllib.parse.quote(email)}"
    try:
        req = urllib.request.Request(url, headers={"User-Agent": _ua(email)})
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read())
        best = data.get("best_oa_location")
        if best and best.get("url_for_pdf"):
            return best["url_for_pdf"]
        for loc in data.get("oa_locations", []):
            if loc.get("url_for_pdf"):
                return loc["url_for_pdf"]
        if best and best.get("url"):
            return best["url"]
    except urllib.error.HTTPError as e:
        if e.code == 422:
            log.warning("Unpaywall rejected email '%s' (HTTP 422). "
                        "Use a real email address, not example.com.", email)
        else:
            log.debug("Unpaywall error for %s: %s", doi, e)
    except (urllib.error.URLError, json.JSONDecodeError) as e:
        log.debug("Unpaywall error for %s: %s", doi, e)
    return None


# ============================================================
# 2. PMC (3-method fallback, JS-challenge resistant)
# ============================================================

def id_to_pmcid(identifier: str, email: str) -> str | None:
    """Convert PMID or DOI to PMCID via NCBI ID converter."""
    if not identifier:
        return None
    url = (f"https://www.ncbi.nlm.nih.gov/pmc/utils/idconv/v1.0/"
           f"?ids={urllib.parse.quote(identifier, safe='/')}&format=json")
    try:
        req = urllib.request.Request(url, headers={"User-Agent": _ua(email)})
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read())
        records = data.get("records", [])
        if records and records[0].get("pmcid"):
            return records[0]["pmcid"]
    except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError) as e:
        log.debug("NCBI ID converter error for %s: %s", identifier, e)
    return None


def download_pmc_pdf(pmcid: str, outpath: Path, email: str) -> bool:
    """Download PDF from PMC via Europe PMC → OA FTP → web fallback."""

    # Method A: Europe PMC REST API (most reliable, no JS)
    try:
        url = (f"https://europepmc.org/backend/ptpmcrender.fcgi"
               f"?accid={pmcid}&blobtype=pdf")
        data, _, _ = fetch_bytes(url, email, accept="application/pdf,*/*", timeout=30)
        if save_pdf(data, outpath):
            log.debug("PMC Method A (Europe PMC) succeeded for %s", pmcid)
            return True
    except (urllib.error.URLError, urllib.error.HTTPError, OSError) as e:
        log.debug("PMC Method A failed for %s: %s", pmcid, e)

    # Method B: PMC OA FTP service (XML with direct PDF link)
    try:
        url = f"https://www.ncbi.nlm.nih.gov/pmc/utils/oa/oa.fcgi?id={pmcid}"
        xml_data, _, _ = fetch_bytes(url, email, timeout=15)
        root = ET.fromstring(xml_data)
        # Check for error response (non-OA articles)
        if root.find(".//error") is not None:
            log.debug("PMC Method B: %s is not in OA subset", pmcid)
        else:
            for link in root.iter("link"):
                href = link.get("href", "")
                if href.endswith(".pdf"):
                    if href.startswith("ftp://"):
                        href = href.replace(
                            "ftp://ftp.ncbi.nlm.nih.gov/pub/pmc/",
                            "https://ftp.ncbi.nlm.nih.gov/pub/pmc/", 1)
                    data, _, _ = fetch_bytes(
                        href, email, accept="application/pdf,*/*", timeout=30)
                    if save_pdf(data, outpath):
                        log.debug("PMC Method B (OA FTP) succeeded for %s", pmcid)
                        return True
    except (urllib.error.URLError, urllib.error.HTTPError,
            ET.ParseError, OSError) as e:
        log.debug("PMC Method B failed for %s: %s", pmcid, e)

    # Method C: Direct PMC web URL (may hit JS PoW challenge)
    try:
        url = f"https://www.ncbi.nlm.nih.gov/pmc/articles/{pmcid}/pdf/"
        data, final_url, ct = fetch_bytes(
            url, email, accept="application/pdf,*/*")
        if "pdf" in ct.lower() or final_url.endswith(".pdf"):
            if save_pdf(data, outpath):
                log.debug("PMC Method C (web) succeeded for %s", pmcid)
                return True
    except (urllib.error.URLError, urllib.error.HTTPError, OSError) as e:
        log.debug("PMC Method C failed for %s: %s", pmcid, e)

    return False


# ============================================================
# 3. OpenAlex + Crossref
# ============================================================

def openalex_lookup(doi: str, email: str) -> list[str]:
    url = (f"https://api.openalex.org/works/"
           f"https://doi.org/{urllib.parse.quote(doi, safe='/')}")
    candidates = []
    try:
        req = urllib.request.Request(url, headers={"User-Agent": _ua(email)})
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read())
        oa = data.get("open_access", {}) or {}
        primary = data.get("primary_location", {}) or {}
        for v in [primary.get("pdf_url"), oa.get("oa_url"),
                  primary.get("landing_page_url")]:
            if v and v not in candidates:
                candidates.append(v)
    except (urllib.error.URLError, urllib.error.HTTPError,
            json.JSONDecodeError) as e:
        log.debug("OpenAlex error for %s: %s", doi, e)
    return candidates


def crossref_lookup(doi: str, email: str) -> list[str]:
    url = f"https://api.crossref.org/works/{urllib.parse.quote(doi, safe='/')}"
    candidates = []
    try:
        req = urllib.request.Request(url, headers={"User-Agent": _ua(email)})
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read())
        msg = data.get("message", {}) or {}
        for link in msg.get("link", []) or []:
            v = link.get("URL")
            if v and v not in candidates:
                candidates.append(v)
        primary = ((msg.get("resource") or {}).get("primary") or {}).get("URL")
        if primary and primary not in candidates:
            candidates.append(primary)
    except (urllib.error.URLError, urllib.error.HTTPError,
            json.JSONDecodeError) as e:
        log.debug("Crossref error for %s: %s", doi, e)
    return candidates


# ============================================================
# 4. Landing page scraper
# ============================================================

def scrape_pdf_candidates(html: str) -> list[str]:
    patterns = [
        r'citation_pdf_url"\s+content="([^"]+)"',
        r"name=\"citation_pdf_url\"\s+content=\"([^\"]+)\"",
        r'href="([^"]+\.pdf[^"]*)"',
    ]
    found = []
    for pat in patterns:
        for m in re.findall(pat, html, flags=re.IGNORECASE):
            if m not in found:
                found.append(m)
    return found


def download_from_landing(url: str, outpath: Path, email: str) -> bool:
    try:
        raw, final_url, ct = fetch_bytes(url, email, accept="text/html,*/*")
        if "pdf" in ct.lower():
            return save_pdf(raw, outpath)
        html = raw.decode("utf-8", errors="ignore")
        for candidate in scrape_pdf_candidates(html):
            absolute = urllib.parse.urljoin(final_url, candidate)
            try:
                data, _, _ = fetch_bytes(
                    absolute, email, accept="application/pdf,*/*")
                if save_pdf(data, outpath):
                    return True
            except (urllib.error.URLError, urllib.error.HTTPError, OSError):
                continue
    except (urllib.error.URLError, urllib.error.HTTPError, OSError) as e:
        log.debug("Landing page error for %s: %s", url, e)
    return False


def download_pdf(url: str, outpath: Path, email: str) -> bool:
    try:
        data, _, _ = fetch_bytes(url, email, accept="application/pdf,*/*")
        return save_pdf(data, outpath)
    except (urllib.error.URLError, urllib.error.HTTPError, OSError) as e:
        log.debug("Direct download error for %s: %s", url, e)
    return False


# ============================================================
# 5. Main pipeline
# ============================================================

def gather_candidates(doi: str, email: str) -> list[str]:
    """Collect OA PDF candidate URLs from multiple sources."""
    urls: list[str] = []

    def add(v: str | None):
        if v and v not in urls:
            urls.append(v)

    add(unpaywall_lookup(doi, email))
    for v in openalex_lookup(doi, email):
        add(v)
    for v in crossref_lookup(doi, email):
        add(v)
    add(f"https://doi.org/{doi}")
    return urls


def process_doi(doi: str, outdir: Path, email: str,
                pmid: str = "") -> str:
    """Try to download a PDF for one DOI. Returns status string."""
    safe_name = re.sub(r"[^\w\-.]", "_", doi)
    outpath = outdir / f"{safe_name}.pdf"

    if existing_pdf_ok(outpath):
        return "skip"

    # Remove stale stub
    if outpath.exists():
        outpath.unlink(missing_ok=True)

    # Step 1: Unpaywall direct PDF URL (fastest path)
    uw_url = unpaywall_lookup(doi, email)
    if uw_url and ".pdf" in uw_url.lower():
        if download_pdf(uw_url, outpath, email):
            return "oa"
        time.sleep(0.3)

    # Step 2: PMC (try before slow landing-page scraping)
    pmcid = id_to_pmcid(pmid, email) if pmid else None
    if not pmcid:
        pmcid = id_to_pmcid(doi, email)
    if pmcid and download_pmc_pdf(pmcid, outpath, email):
        return "pmc"

    # Step 3: OA candidates from OpenAlex, Crossref, landing pages
    candidates: list[str] = []
    if uw_url and uw_url not in candidates:
        candidates.append(uw_url)
    for v in openalex_lookup(doi, email):
        if v not in candidates:
            candidates.append(v)
    for v in crossref_lookup(doi, email):
        if v not in candidates:
            candidates.append(v)
    candidates.append(f"https://doi.org/{doi}")

    for url in candidates:
        if ".pdf" in url.lower():
            ok = download_pdf(url, outpath, email)
        else:
            ok = download_from_landing(url, outpath, email)
        if ok:
            return "oa"
        time.sleep(0.3)

    return "fail"


def read_doi_file(path: Path) -> list[dict]:
    """Read DOI list. Supports plain DOIs or TSV with DOI/PMID columns."""
    records = []
    with open(path, encoding="utf-8") as f:
        first_line = f.readline().strip()
        f.seek(0)

        # TSV with header containing DOI column
        if "\t" in first_line and "doi" in first_line.lower():
            import csv
            reader = csv.DictReader(f, delimiter="\t")
            for row in reader:
                doi = ""
                pmid = ""
                for k, v in row.items():
                    if k.lower().strip() == "doi":
                        doi = (v or "").strip()
                    elif k.lower().strip() == "pmid":
                        pmid = (v or "").strip()
                if doi:
                    records.append({"doi": doi, "pmid": pmid})
        else:
            # Plain text: one DOI per line
            for line in f:
                line = line.strip()
                if line and not line.startswith("#"):
                    records.append({"doi": line, "pmid": ""})
    return records


def main():
    parser = argparse.ArgumentParser(
        description="Batch download open-access PDFs by DOI.")
    parser.add_argument("input", type=Path,
                        help="File with DOIs (one per line, or TSV with DOI column)")
    parser.add_argument("-o", "--output", type=Path, default=Path("pdfs"),
                        help="Output directory (default: pdfs/)")
    parser.add_argument("-e", "--email", required=True,
                        help="Contact email (required by Unpaywall TOS)")
    parser.add_argument("-v", "--verbose", action="store_true",
                        help="Show debug messages")
    args = parser.parse_args()

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

    args.output.mkdir(parents=True, exist_ok=True)
    records = read_doi_file(args.input)
    print(f"Loaded {len(records)} DOIs from {args.input}")

    stats = {"oa": 0, "pmc": 0, "fail": 0, "skip": 0}

    for i, rec in enumerate(records, 1):
        doi = rec["doi"]
        pmid = rec.get("pmid", "")
        print(f"  [{i}/{len(records)}] {doi}", end=" … ", flush=True)

        status = process_doi(doi, args.output, args.email, pmid)
        stats[status] += 1

        labels = {"oa": "OK (OA)", "pmc": "OK (PMC)",
                  "fail": "FAIL", "skip": "SKIP"}
        print(labels[status])
        time.sleep(0.5)

    print(f"\n--- Summary ---")
    print(f"  OA:      {stats['oa']}")
    print(f"  PMC:     {stats['pmc']}")
    print(f"  Failed:  {stats['fail']}")
    print(f"  Skipped: {stats['skip']}")
    total = stats["oa"] + stats["pmc"] + stats["fail"]
    if total > 0:
        pct = (stats["oa"] + stats["pmc"]) / total * 100
        print(f"  Success: {pct:.0f}%")

    # Write failed DOIs for manual retrieval
    if stats["fail"] > 0:
        fail_path = args.output / "manual_needed.txt"
        with open(fail_path, "w") as f:
            f.write("# DOIs needing manual retrieval\n")
            f.write("# Options: institutional access, ILL\n\n")
            for rec in records:
                safe = re.sub(r"[^\w\-.]", "_", rec["doi"])
                pdf = args.output / f"{safe}.pdf"
                if not existing_pdf_ok(pdf):
                    f.write(f"{rec['doi']}\n")
        print(f"  Manual list: {fail_path}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Convert research paper PDFs to LLM-friendly Markdown.

Uses pymupdf4llm for high-quality extraction optimized for academic papers:
two-column layout handling, table preservation, header/footer removal.

Requires: pip install pymupdf4llm

Usage:
    python pdf_to_md.py pdfs/                   # convert all PDFs in directory
    python pdf_to_md.py paper.pdf               # convert single file
    python pdf_to_md.py pdfs/ -o markdown/       # custom output directory
    python pdf_to_md.py pdfs/ --pages 0-9        # first 10 pages only
    python pdf_to_md.py pdfs/ --force            # overwrite existing .md files
"""

import argparse
import re
import sys
from pathlib import Path

try:
    import pymupdf4llm
except ImportError:
    print("Error: pymupdf4llm is not installed.", file=sys.stderr)
    print("Install with: pip install pymupdf4llm", file=sys.stderr)
    sys.exit(1)


def parse_page_range(spec: str) -> list[int]:
    """Parse page range string like '0-9' or '0,2,5-7' into list of ints."""
    pages = []
    for part in spec.split(","):
        part = part.strip()
        if "-" in part:
            start, end = part.split("-", 1)
            pages.extend(range(int(start), int(end) + 1))
        else:
            pages.append(int(part))
    return pages


def clean_markdown(text: str) -> str:
    """Post-process pymupdf4llm output for cleaner LLM consumption."""
    # Collapse excessive blank lines (3+ → 2)
    text = re.sub(r"\n{4,}", "\n\n\n", text)
    # Strip trailing whitespace per line
    text = "\n".join(line.rstrip() for line in text.splitlines())
    return text.strip() + "\n"


def convert_pdf(pdf_path: Path, out_dir: Path, *,
                pages: list[int] | None = None,
                force: bool = False,
                verbose: bool = False) -> bool:
    """Convert a single PDF to Markdown. Returns True on success."""
    md_path = out_dir / pdf_path.with_suffix(".md").name

    if md_path.exists() and md_path.stat().st_size > 0 and not force:
        if verbose:
            print(f"  SKIP: {md_path.name} (exists, use --force to overwrite)")
        return True

    try:
        kwargs = {
            "show_progress": False,
            # Academic paper defaults: skip images (saves tokens),
            # strict table detection for grid-line tables
            "write_images": False,
            "ignore_images": True,
            "table_strategy": "lines_strict",
        }
        if pages is not None:
            kwargs["pages"] = pages

        # Suppress pymupdf's C-level OCR/parser messages (stdout + stderr)
        if not verbose:
            import os as _os
            _devnull = _os.open(_os.devnull, _os.O_WRONLY)
            _old_stdout = _os.dup(1)
            _old_stderr = _os.dup(2)
            _os.dup2(_devnull, 1)
            _os.dup2(_devnull, 2)
        try:
            md_text = pymupdf4llm.to_markdown(str(pdf_path), **kwargs)
        finally:
            if not verbose:
                _os.dup2(_old_stdout, 1)
                _os.dup2(_old_stderr, 2)
                _os.close(_devnull)
                _os.close(_old_stdout)
                _os.close(_old_stderr)
        md_text = clean_markdown(md_text)

        md_path.write_text(md_text, encoding="utf-8")
        if verbose:
            kb = len(md_text.encode("utf-8")) / 1024
            print(f"  OK: {md_path.name} ({kb:.1f} KB)")
        return True
    except Exception as e:
        print(f"  FAIL: {pdf_path.name}: {e}", file=sys.stderr)
        return False


def main():
    parser = argparse.ArgumentParser(
        description="Convert research PDFs to LLM-friendly Markdown "
                    "(via pymupdf4llm).")
    parser.add_argument("input", type=Path,
                        help="PDF file or directory containing PDFs")
    parser.add_argument("-o", "--output", type=Path, default=None,
                        help="Output directory (default: same as input)")
    parser.add_argument("--pages", type=str, default=None,
                        help="Page range, e.g. '0-9' for first 10 pages")
    parser.add_argument("--force", action="store_true",
                        help="Overwrite existing .md files")
    parser.add_argument("-v", "--verbose", action="store_true",
                        help="Show per-file progress")
    args = parser.parse_args()

    # Resolve input
    if args.input.is_file():
        pdfs = [args.input]
        default_out = args.input.parent
    elif args.input.is_dir():
        pdfs = sorted(args.input.glob("*.pdf"))
        default_out = args.input
    else:
        print(f"Error: {args.input} not found", file=sys.stderr)
        sys.exit(1)

    out_dir = args.output or default_out
    out_dir.mkdir(parents=True, exist_ok=True)

    if not pdfs:
        print("No PDF files found.")
        return

    # Parse pages
    pages = parse_page_range(args.pages) if args.pages else None

    print(f"Converting {len(pdfs)} PDF(s) → Markdown", flush=True)
    ok = 0
    fail = 0
    for pdf in pdfs:
        if convert_pdf(pdf, out_dir, pages=pages, force=args.force,
                       verbose=args.verbose):
            ok += 1
        else:
            fail += 1

    print(f"\n--- Summary ---")
    print(f"  Converted: {ok}")
    print(f"  Failed:    {fail}")
    print(f"  Total:     {len(pdfs)}")


if __name__ == "__main__":
    main()

schema_version: 2
name: fulltext-retrieval
layer: A
owner_domain: literature_discovery
maturity: official

when_to_use: "Batch-download open-access full-text PDFs from a DOI list; optionally convert them to Markdown for token-efficient analysis."
when_NOT_to_use: "Finding or verifying citations (use search-lit / verify-refs). Retrieving paywalled or non-open-access content."

inputs:
  - path: "DOI list (.txt, one per line, or .tsv with a DOI column)"
    schema: csv
    required: true
outputs:
  - path: "downloaded open-access PDFs (pdfs/)"
  - path: "optional PDF-to-Markdown conversions"

deterministic_scripts:
  - fetch_oa.py
  - pdf_to_md.py
side_effects:
  - downloads_files
  - network_access_oa_apis
downstream_consumers:
  - meta-analysis
  - obsidian-paper-vault
forbidden_actions:
  - download_paywalled_content
  - bypass_publisher_access_controls

# v2.1 quality card
purpose: "Resolve a DOI list to open-access full-text PDFs via legitimate OA APIs, with optional Markdown conversion."
safety_boundaries:
  - "Uses legitimate open-access sources only (Unpaywall, PMC / Europe PMC, OpenAlex, Crossref); never circumvents paywalls or access controls."
  - "Validates each download (>=10 KB and a %PDF- header) before accepting it."
known_limitations:
  - "Only open-access content is retrievable; non-OA DOIs fail by design rather than fetching from unauthorized sources."
  - "PDF-to-Markdown conversion requires the optional pymupdf4llm dependency (AGPL-3.0 or commercial license)."
validation_commands:
  - "python fetch_oa.py dois.txt -o pdfs/ -e <email> --verbose   # per-DOI source trace"
  - "verify each output begins with %PDF- and is at least 10 KB"
evidence_surface: bundled_script

#!/usr/bin/env python3
"""Regression test for fulltext-retrieval/pdf_to_md.py pure helpers.

pdf_to_md.py exits at import time if pymupdf4llm is unavailable, so we stub that
module before importing and exercise only the dependency-free, deterministic
helpers: parse_page_range (page-spec parsing) and clean_markdown (post-process).
This keeps CI free of the heavy PyMuPDF/pymupdf4llm dependency while still
gating the logic most prone to silent breakage. Stdlib-only, network-free.
"""
import importlib.util
import sys
import types
from pathlib import Path

HERE = Path(__file__).resolve().parent
MODULE_PATH = HERE.parent / "pdf_to_md.py"


def load_module():
    # Stub pymupdf4llm so the module-level import does not sys.exit(1).
    sys.modules.setdefault("pymupdf4llm", types.ModuleType("pymupdf4llm"))
    spec = importlib.util.spec_from_file_location("pdf_to_md", MODULE_PATH)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def main() -> int:
    assert MODULE_PATH.exists(), f"ENV-ERR: {MODULE_PATH} missing"
    mod = load_module()
    fails = []

    def check(label, cond):
        print(f"  {'PASS' if cond else 'FAIL'}  {label}")
        if not cond:
            fails.append(label)

    # --- parse_page_range ---
    check("range '0-9' -> 0..9", mod.parse_page_range("0-9") == list(range(0, 10)))
    check("list '0,2,5-7' -> [0,2,5,6,7]", mod.parse_page_range("0,2,5-7") == [0, 2, 5, 6, 7])
    check("single '3' -> [3]", mod.parse_page_range("3") == [3])
    check("whitespace ' 1 , 4 ' tolerated", mod.parse_page_range(" 1 , 4 ") == [1, 4])

    # --- clean_markdown ---
    out = mod.clean_markdown("a\n\n\n\n\nb   \n\n\n")
    check("collapses 4+ newlines to 3", "\n\n\n\n" not in out)
    check("rstrips line trailing spaces", "b   " not in out and "b" in out)
    check("ends with exactly one newline", out.endswith("\n") and not out.endswith("\n\n"))
    check("strips leading/trailing blank lines", out == "a\n\n\nb\n")
    # idempotent
    check("clean_markdown is idempotent", mod.clean_markdown(out) == out)

    if fails:
        print(f"FAILURES: {len(fails)}")
        return 1
    print("ALL PASS")
    return 0


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