find_massive_gnps.py

"""
Script to find GNPS task IDs, Zenodo record IDs, and MassIVE accessions in
PDF/TXT papers located in the data_retriever/papers/ directory.

Uses docling with ThreadedPdfPipelineOptions for fast PDF conversion.
Saves converted papers as markdown in data_retriever/papers_md/.
Updates data_retriever/papers_summary.csv with columns:
  - converted     : yes / no
  - gnps_task_ids : pipe-separated list of unique GNPS 32-char task IDs found
  - gnps_methods  : pipe-separated list of detection methods used (aligned with gnps_task_ids)
  - zenodo_ids    : pipe-separated list of unique Zenodo record IDs found, or empty
  - MASSIVE       : pipe-separated list of unique MassIVE accessions (e.g. MSV000087728)

GNPS detection is performed by four ordered methods from hex_finder.py:
  1. regex_strict      – full canonical URL with task=<hex32>
  2. regex_task_id     – "task [sep] hex32" or "ID [sep] hex32"
  3. regex_domain_near – gnps.ucsd.edu (OCR-tolerant) followed by hex32 within 300 chars
  4. fuzzy_domain      – broad fuzzy domain window + rapidfuzz scoring

No LLM is used. Only hits where the detected raw ID is already a clean 32-char
hexadecimal string are kept.

Dependencies:
    pip install docling rapidfuzz

Usage:
    python data_retriever/find_massive_gnps.py
    python data_retriever/find_massive_gnps.py --output results.json
    python data_retriever/find_massive_gnps.py --papers-dir /path/to/papers
    python data_retriever/find_massive_gnps.py --only-matches
    python data_retriever/find_massive_gnps.py --no-csv-update
"""

import re
import json
import argparse
import csv
import os
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import Optional

from docling.pipeline.threaded_standard_pdf_pipeline import ThreadedStandardPdfPipeline
from docling.datamodel.accelerator_options import AcceleratorDevice, AcceleratorOptions
from docling.datamodel.base_models import ConversionStatus, InputFormat
from docling.datamodel.pipeline_options import ThreadedPdfPipelineOptions
from docling.document_converter import DocumentConverter, PdfFormatOption

# Import hex detection methods (no LLM used)
from data_retriever.hex_finder import (
    regex_strict,
    regex_task_id,
    regex_domain_near,
    fuzzy_domain,
)

_HEX32_FULL = re.compile(r'^[a-f0-9]{32}$', re.IGNORECASE)

# Ordered detection pipeline: fastest / most precise first, fuzzy last
_GNPS_METHODS = [regex_strict, regex_task_id, regex_domain_near, fuzzy_domain]

# ---------------------------------------------------------------------------
# Zenodo & MassIVE patterns
# ---------------------------------------------------------------------------

# Zenodo record IDs — three forms, each tolerant of spaces between characters
# injected by PDF text extraction:
#   1. https://doi.org/10.5281/zenodo.1234567       (DOI)
#      including spaced:  10 . 5 2 8 1 / z e n o d o . 1 7 2 4 3 8 4 9
#   2. https://zenodo.org/record/1234567             (direct record URL)
#   3. zenodo.1234567  or  zenodo. 17243849          (bare reference)
#
# Strategy: use a broad spaced pattern, then normalise + extract the numeric ID.
RE_ZENODO = re.compile(
    r"(?:"
    # Form 1 – DOI: 10.5281/zenodo.<digits>  (spaces allowed anywhere)
    r"10\s*\.\s*5\s*2\s*8\s*1\s*/\s*z\s*e\s*n\s*o\s*d\s*o\s*\.\s*\d[\d\s]*"
    # Form 2 – zenodo.org URL
    r"|https?\s*://\s*z\s*e\s*n\s*o\s*d\s*o\s*\.\s*o\s*r\s*g\s*/\s*"
    r"(?:r\s*e\s*c\s*o\s*r\s*d|d\s*e\s*p\s*o\s*s\s*i\s*t)\s*/\s*\d[\d\s]*"
    # Form 3 – bare "zenodo.<digits>" (not preceded by / . or word char)
    r"|(?<![./\w])z\s*e\s*n\s*o\s*d\s*o\s*\.\s*\d[\d\s]*"
    r")",
    re.IGNORECASE,
)

CONTEXT_WINDOW = 120  # characters on each side of a match

# MassIVE accessions: MSV followed by 9 digits, tolerant of spaced OCR artifacts.
# Uses negative lookbehind to avoid consuming preceding alphanumerics and
# negative lookahead to avoid an extra digit.
RE_MASSIVE = re.compile(
    r"(?<![A-Z0-9])M\s*S\s*V\s*\d(?:\s*\d){8}(?!\d)",
    re.IGNORECASE,
)


def extract_zenodo_id(raw: str) -> Optional[str]:
    """
    From a raw Zenodo regex match (may contain spaces), extract just the
    numeric record ID (all digits, no spaces).  Returns ``None`` if no
    digits are found.
    """
    normalised = re.sub(r"\s+", "", raw)
    m = re.search(r"(\d+)\s*$", normalised)
    return m.group(1) if m else None


def extract_massive_id(raw: str) -> Optional[str]:
    """Normalise spaced MassIVE accession text to canonical MSV#########."""
    normalised = re.sub(r"\s+", "", raw).upper()
    return normalised if re.fullmatch(r"MSV\d{9}", normalised) else None


def get_context(text: str, match: re.Match, window: int = CONTEXT_WINDOW) -> str:
    """Return surrounding text of a regex match, with the match highlighted."""
    start = max(0, match.start() - window)
    end = min(len(text), match.end() + window)
    snippet = text[start:end].replace("\n", " ").strip()
    rel_start = match.start() - start
    rel_end = match.end() - start
    return snippet[:rel_start] + ">>>" + snippet[rel_start:rel_end] + "<<<" + snippet[rel_end:]


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------

@dataclass
class ZenodoMatch:
    matched_text: str
    context: str
    page: Optional[int]


@dataclass
class GnpsTaskHit:
    task_id: str       # 32-char lowercase hex
    method: str        # name of detection function
    context: str       # raw matched context from the method


@dataclass
class MassiveMatch:
    matched_text: str
    context: str
    page: Optional[int]


@dataclass
class PaperResult:
    filename: str
    path: str
    converted: bool = False
    gnps_hits: list[GnpsTaskHit] = field(default_factory=list)
    zenodo_matches: list[ZenodoMatch] = field(default_factory=list)
    massive_matches: list[MassiveMatch] = field(default_factory=list)
    error: Optional[str] = None

    @property
    def has_matches(self) -> bool:
        return bool(self.gnps_hits) or bool(self.zenodo_matches) or bool(self.massive_matches)

    # ── GNPS ──────────────────────────────────────────────────────────────────

    @property
    def all_gnps_task_ids(self) -> list[str]:
        """Return unique GNPS task IDs preserving first-found order."""
        seen: set[str] = set()
        ids: list[str] = []
        for hit in self.gnps_hits:
            tid = hit.task_id.lower()
            if tid not in seen:
                seen.add(tid)
                ids.append(tid)
        return ids

    @property
    def gnps_task_ids_str(self) -> str:
        """Pipe-separated string of unique GNPS task IDs (for CSV storage)."""
        return " | ".join(self.all_gnps_task_ids)

    @property
    def all_gnps_methods(self) -> list[str]:
        """
        Return the detection method name for each unique task ID
        (aligned with all_gnps_task_ids).
        """
        seen: set[str] = set()
        methods: list[str] = []
        for hit in self.gnps_hits:
            tid = hit.task_id.lower()
            if tid not in seen:
                seen.add(tid)
                methods.append(hit.method)
        return methods

    @property
    def gnps_methods_str(self) -> str:
        """Pipe-separated string of detection methods (aligned with gnps_task_ids_str)."""
        return " | ".join(self.all_gnps_methods)

    # ── Zenodo ────────────────────────────────────────────────────────────────

    @property
    def all_zenodo_ids(self) -> list[str]:
        """Return all unique Zenodo record IDs (numeric) found in the paper."""
        seen: set[str] = set()
        ids: list[str] = []
        for m in self.zenodo_matches:
            zid = extract_zenodo_id(m.matched_text)
            if zid and zid not in seen:
                seen.add(zid)
                ids.append(zid)
        return ids

    @property
    def zenodo_ids_str(self) -> str:
        """Pipe-separated string of unique Zenodo record IDs (for CSV storage)."""
        return " | ".join(self.all_zenodo_ids)

    # ── MassIVE ───────────────────────────────────────────────────────────────

    @property
    def all_massive_ids(self) -> list[str]:
        """Return all unique MassIVE accessions (MSV#########) found in the paper."""
        seen: set[str] = set()
        ids: list[str] = []
        for m in self.massive_matches:
            mid = extract_massive_id(m.matched_text)
            if mid and mid not in seen:
                seen.add(mid)
                ids.append(mid)
        return ids

    @property
    def massive_ids_str(self) -> str:
        """Pipe-separated string of unique MassIVE accessions (for CSV storage)."""
        return " | ".join(self.all_massive_ids)


# ---------------------------------------------------------------------------
# GNPS task detection on text
# ---------------------------------------------------------------------------

def find_gnps_tasks_in_text(text: str, workers: int = 4) -> list[GnpsTaskHit]:
    """
    Run the ordered hex_finder detection pipeline on ``text``.

    For each method, all hits are collected (not just the first one).
    Only hits where the captured raw ID is already a clean 32-char hex string
    are kept — the LLM fallback is intentionally skipped.

    Deduplication by task ID is applied: the first method to find a given ID
    wins (methods are run in parallel but results are merged in canonical order).

    ``workers`` controls the thread-pool size for concurrent method execution.
    """
    # Run all methods concurrently; results keyed by method index to preserve order
    method_results: dict[int, list[tuple[str, str]]] = {}

    def _run_method(idx: int, method):
        return idx, list(method(text))

    with ThreadPoolExecutor(max_workers=min(workers, len(_GNPS_METHODS))) as pool:
        futures = {pool.submit(_run_method, i, m): i for i, m in enumerate(_GNPS_METHODS)}
        for future in as_completed(futures):
            idx, pairs = future.result()
            method_results[idx] = pairs

    # Merge in canonical order (strict → task_id → domain_near → fuzzy)
    hits: list[GnpsTaskHit] = []
    seen_ids: set[str] = set()

    for i, method in enumerate(_GNPS_METHODS):
        for context, raw_id in method_results.get(i, []):
            if not _HEX32_FULL.match(raw_id):
                # Noisy / incomplete ID — skip without LLM
                continue
            task_id = raw_id.lower()
            if task_id not in seen_ids:
                seen_ids.add(task_id)
                hits.append(GnpsTaskHit(
                    task_id=task_id,
                    method=method.__name__,
                    context=context[:240],
                ))

    return hits


def find_massive_in_text(text: str) -> list[MassiveMatch]:
    """Find MassIVE accessions in raw text (no threading needed)."""
    return search_massive(text, page_num=None)


# ---------------------------------------------------------------------------
# Zenodo search
# ---------------------------------------------------------------------------

def search_zenodo(text: str, page_num: Optional[int]) -> list[ZenodoMatch]:
    """Search a text block for Zenodo record IDs."""
    found: list[ZenodoMatch] = []
    seen: set[str] = set()
    for m in RE_ZENODO.finditer(text):
        key = m.group().lower()
        if key not in seen:
            seen.add(key)
            found.append(ZenodoMatch(
                matched_text=m.group(),
                context=get_context(text, m),
                page=page_num,
            ))
    return found


def search_massive(text: str, page_num: Optional[int]) -> list[MassiveMatch]:
    """Search a text block for MassIVE accessions (MSV#########)."""
    found: list[MassiveMatch] = []
    seen: set[str] = set()
    for m in RE_MASSIVE.finditer(text):
        key = m.group().lower()
        if key not in seen:
            seen.add(key)
            found.append(MassiveMatch(
                matched_text=m.group(),
                context=get_context(text, m),
                page=page_num,
            ))
    return found


# ---------------------------------------------------------------------------
# Docling helpers
# ---------------------------------------------------------------------------

def build_converter() -> DocumentConverter:
    """Build a docling DocumentConverter using ThreadedPdfPipelineOptions."""
    pipeline_options = ThreadedPdfPipelineOptions(
        accelerator_options=AcceleratorOptions(
            device=AcceleratorDevice.CUDA,
        ),
        ocr_batch_size=100,
        layout_batch_size=100,
        table_batch_size=4,
    )
    # Disable OCR and table structure analysis — we only need embedded text
    pipeline_options.do_ocr = False
    pipeline_options.do_table_structure = False

    converter = DocumentConverter(
        format_options={
            InputFormat.PDF: PdfFormatOption(
                pipeline_cls=ThreadedStandardPdfPipeline,
                pipeline_options=pipeline_options,
            )
        }
    )
    return converter


# ---------------------------------------------------------------------------
# Per-paper processing
# ---------------------------------------------------------------------------

def _process_text(text: str, result: PaperResult, workers: int = 4) -> None:
    """
    Apply all detectors to ``text`` and accumulate results into ``result``.

    GNPS task detection, Zenodo search, and MassIVE search run concurrently.
    ``workers`` is forwarded to :func:`find_gnps_tasks_in_text`.
    """
    with ThreadPoolExecutor(max_workers=3) as pool:
        future_gnps    = pool.submit(find_gnps_tasks_in_text, text, workers)
        future_zenodo  = pool.submit(search_zenodo, text, None)
        future_massive = pool.submit(search_massive, text, None)
        result.gnps_hits.extend(future_gnps.result())
        result.zenodo_matches.extend(future_zenodo.result())
        result.massive_matches.extend(future_massive.result())


def process_pdf(
    pdf_path: Path,
    converter: Optional[DocumentConverter],
    md_dir: Path,
    *,
    workers: int = 4,
    converter_lock: Optional[threading.Lock] = None,
) -> PaperResult:
    """
    Convert a PDF with docling (or reuse existing markdown), then search for patterns.

    ``workers``        – thread-pool size forwarded to :func:`_process_text`.
    ``converter_lock`` – optional lock serialising docling conversion calls so that
                         concurrent threads do not call the converter simultaneously.
    """
    result = PaperResult(filename=pdf_path.name, path=str(pdf_path))
    md_path = md_dir / (pdf_path.stem + ".md")

    try:
        if md_path.exists():
            result.converted = True
            md_content = md_path.read_text(encoding="utf-8")
            _process_text(md_content, result, workers=workers)
        else:
            if converter is None:
                result.error = "Converter not initialised but markdown cache is missing."
                return result

            # Serialise calls to the docling converter (not thread-safe for
            # concurrent PDF conversions, but its internal pipeline is already
            # threaded, so one call at a time is the correct usage).
            lock_ctx = converter_lock if converter_lock is not None else threading.Lock()
            with lock_ctx:
                conv_result = converter.convert(str(pdf_path))

            if conv_result.status not in (
                ConversionStatus.SUCCESS,
                ConversionStatus.PARTIAL_SUCCESS,
            ):
                result.error = f"Conversion failed with status: {conv_result.status}"
                return result

            doc = conv_result.document
            result.converted = True

            md_content = doc.export_to_markdown()
            md_path.write_text(md_content, encoding="utf-8")

            # Search the full markdown (simpler and more reliable than per-item)
            _process_text(md_content, result, workers=workers)

    except Exception as exc:
        result.error = str(exc)

    return result


def process_txt(txt_path: Path, md_dir: Path, *, workers: int = 4) -> PaperResult:
    """Read a plain-text paper (or reuse existing markdown), and search patterns."""
    result = PaperResult(filename=txt_path.name, path=str(txt_path))
    md_path = md_dir / (txt_path.stem + ".md")
    try:
        if md_path.exists():
            text = md_path.read_text(encoding="utf-8", errors="replace")
        else:
            text = txt_path.read_text(encoding="utf-8", errors="replace")
            md_path.write_text(text, encoding="utf-8")

        result.converted = True
        _process_text(text, result, workers=workers)

    except Exception as exc:
        result.error = str(exc)

    return result


# ---------------------------------------------------------------------------
# CSV update helper
# ---------------------------------------------------------------------------

CSV_NEW_COLS = ["converted", "gnps_task_ids", "gnps_methods", "zenodo_ids", "MASSIVE"]


def update_csv(csv_path: Path, results: list[PaperResult]) -> None:
    """
    Read papers_summary.csv, add/update columns:
      - converted      : yes / no
      - gnps_task_ids  : unique GNPS task IDs (32-char hex), " | "-separated
      - gnps_methods   : detection method name per task ID, " | "-separated
      - zenodo_ids     : unique Zenodo record IDs (numeric), " | "-separated

    Legacy columns (gnps_url, gnps_urls, massive_datasets, massive_dataset) are
    removed from the output.
    """
    if not csv_path.exists():
        print(f"  [CSV] File not found, skipping update: {csv_path}", file=sys.stderr)
        return

    result_by_filename: dict[str, PaperResult] = {r.filename: r for r in results}

    rows: list[dict] = []
    with csv_path.open(newline="", encoding="utf-8") as fh:
        reader = csv.DictReader(fh)
        existing_fields: list[str] = list(reader.fieldnames or [])
        for row in reader:
            rows.append(row)

    # Build final fieldnames: keep existing ones except legacy GNPS/massive cols,
    # then append new cols that aren't already present.
    _legacy_cols = {"gnps_url", "gnps_urls", "massive_datasets", "massive_dataset"}
    final_fields = [f for f in existing_fields if f not in _legacy_cols]
    for col in CSV_NEW_COLS:
        if col not in final_fields:
            final_fields.append(col)

    # Patch rows
    for row in rows:
        # Remove legacy columns
        for col in _legacy_cols:
            row.pop(col, None)

        fname = row.get("filename", "")
        res = result_by_filename.get(fname)
        if res is not None:
            row["converted"] = "yes" if res.converted else "no"
            row["gnps_task_ids"] = res.gnps_task_ids_str
            row["gnps_methods"] = res.gnps_methods_str
            row["zenodo_ids"] = res.zenodo_ids_str
            row["MASSIVE"] = res.massive_ids_str
        else:
            for col in CSV_NEW_COLS:
                row.setdefault(col, "")

    with csv_path.open("w", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(fh, fieldnames=final_fields)
        writer.writeheader()
        writer.writerows(rows)

    print(f"  [CSV] Updated {csv_path} ({len(rows)} rows)")


# ---------------------------------------------------------------------------
# Output helpers
# ---------------------------------------------------------------------------

def print_summary(results: list[PaperResult]) -> None:
    papers_with_matches = [r for r in results if r.has_matches]
    papers_with_errors = [r for r in results if r.error]

    print(f"\n{'=' * 70}")
    print("SUMMARY")
    print(f"{'=' * 70}")
    print(f"Total papers scanned : {len(results)}")
    print(f"Papers converted     : {sum(1 for r in results if r.converted)}")
    print(f"Papers with matches  : {len(papers_with_matches)}")
    print(f"Papers with errors   : {len(papers_with_errors)}")

    if papers_with_matches:
        print(f"\n{'─' * 70}")
        print("PAPERS WITH MATCHES")
        print(f"{'─' * 70}")
        for result in papers_with_matches:
            print(f"\n📄  {result.filename}")
            if result.gnps_hits:
                print(f"  [gnps_task_ids]  ({len(result.all_gnps_task_ids)} unique task(s))")
                for hit in result.gnps_hits:
                    print(f"    • [{hit.method}]  {hit.task_id}")
                    print(f"           …{hit.context[:120]}…")
            if result.zenodo_matches:
                print(f"  [zenodo_ids]  ({len(result.all_zenodo_ids)} unique ID(s))")
                for zm in result.zenodo_matches:
                    page_str = f"p.{zm.page:>3}" if zm.page is not None else "p.???"
                    print(f"    • {page_str}  {zm.matched_text}")
                    print(f"           …{zm.context}…")
            if result.massive_matches:
                print(f"  [MASSIVE]  ({len(result.all_massive_ids)} unique accession(s))")
                for mm in result.massive_matches:
                    page_str = f"p.{mm.page:>3}" if mm.page is not None else "p.???"
                    print(f"    • {page_str}  {mm.matched_text}")
                    print(f"           …{mm.context}…")

    if papers_with_errors:
        print(f"\n{'─' * 70}")
        print("PAPERS WITH ERRORS")
        print(f"{'─' * 70}")
        for result in papers_with_errors:
            print(f"  ✗  {result.filename}: {result.error}")

    print(f"\n{'=' * 70}\n")


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Find GNPS task IDs and Zenodo record IDs in PDF/TXT papers."
    )
    parser.add_argument(
        "--papers-dir",
        type=Path,
        default=Path(__file__).parent / "papers",
        help="Directory containing PDF/TXT papers (default: papers/ next to this script).",
    )
    parser.add_argument(
        "--md-dir",
        type=Path,
        default=Path(__file__).parent / "papers_md",
        help="Directory to save converted Markdown files (default: papers_md/ next to this script).",
    )
    parser.add_argument(
        "--csv",
        type=Path,
        default=Path(__file__).parent / "papers_summary.csv",
        help="Path to papers_summary.csv to update (default: papers_summary.csv next to this script).",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=None,
        help="Optional path to write JSON results.",
    )
    parser.add_argument(
        "--only-matches",
        action="store_true",
        help="Only include papers with at least one match in the output.",
    )
    parser.add_argument(
        "--no-csv-update",
        action="store_true",
        help="Skip updating papers_summary.csv.",
    )
    parser.add_argument(
        "--workers",
        type=int,
        default=min(8, (os.cpu_count() or 4)),
        metavar="N",
        help=(
            "Maximum number of worker threads for parallel file processing and "
            "within-paper extraction (default: min(8, cpu_count))."
        ),
    )
    return parser.parse_args()


def _print_result(result: PaperResult, cached: bool, file_type: str, print_lock: threading.Lock) -> None:
    """Thread-safe progress line for a processed paper."""
    if file_type == "pdf":
        status_ok = "reused ✓" if cached else "converted ✓"
    else:
        status_ok = "reused ✓" if cached else "saved ✓"

    if result.error:
        line = f"  Processing ({'cached MD' if cached else file_type.upper()}): {result.filename} … ERROR – {result.error}"
    elif result.has_matches:
        n_gnps = len(result.all_gnps_task_ids)
        n_zen  = len(result.all_zenodo_ids)
        line = (
            f"  Processing ({'cached MD' if cached else file_type.upper()}): {result.filename} … "
            f"{status_ok}  |  {n_gnps} GNPS task(s), {n_zen} Zenodo ID(s)"
        )
    else:
        line = f"  Processing ({'cached MD' if cached else file_type.upper()}): {result.filename} … {status_ok}  |  no matches"

    with print_lock:
        print(line, flush=True)


def main() -> None:
    args = parse_args()
    papers_dir: Path = args.papers_dir
    md_dir: Path     = args.md_dir
    workers: int     = args.workers

    if not papers_dir.exists():
        raise FileNotFoundError(f"Papers directory not found: {papers_dir}")

    md_dir.mkdir(parents=True, exist_ok=True)

    pdf_files = sorted(papers_dir.glob("*.pdf"))
    txt_files = sorted(papers_dir.glob("*.txt"))
    all_files = pdf_files + txt_files

    if not all_files:
        print(f"No PDF or TXT files found in {papers_dir}")
        return

    print(f"Scanning directory : {papers_dir}")
    print(f"Markdown output    : {md_dir}")
    print(f"Parallel workers   : {workers}")
    print(f"Found {len(pdf_files)} PDF file(s) and {len(txt_files)} TXT file(s).\n")

    results: list[PaperResult] = []
    print_lock = threading.Lock()

    # --- Process PDFs with docling ---
    if pdf_files:
        needs_conversion = [p for p in pdf_files if not (md_dir / (p.stem + ".md")).exists()]
        converter: Optional[DocumentConverter] = None
        # Single lock so concurrent threads never call converter.convert() simultaneously
        converter_lock = threading.Lock()

        if needs_conversion:
            print(
                f"Initialising docling converter for {len(needs_conversion)} uncached PDF(s)"
                " (may download model weights on first run)…"
            )
            converter = build_converter()
            print("Converter ready.\n")
        else:
            print("All PDFs already converted — reusing cached markdown.\n")

        def _do_pdf(pdf_path: Path) -> PaperResult:
            cached = (md_dir / (pdf_path.stem + ".md")).exists()
            result = process_pdf(
                pdf_path, converter, md_dir,
                workers=workers,
                converter_lock=converter_lock,
            )
            _print_result(result, cached, "pdf", print_lock)
            return result

        # Use parallel workers for cached PDFs; conversion is serialised internally
        # via converter_lock so the thread count only helps for extraction on cached files.
        with ThreadPoolExecutor(max_workers=workers) as pool:
            futures = {pool.submit(_do_pdf, p): p for p in pdf_files}
            # Collect in submission order to keep results list deterministic
            ordered = {p: f for f, p in futures.items()}
            for pdf_path in pdf_files:
                results.append(ordered[pdf_path].result())

    # --- Process TXT files in parallel ---
    def _do_txt(txt_path: Path) -> PaperResult:
        cached = (md_dir / (txt_path.stem + ".md")).exists()
        result = process_txt(txt_path, md_dir, workers=workers)
        _print_result(result, cached, "txt", print_lock)
        return result

    if txt_files:
        with ThreadPoolExecutor(max_workers=workers) as pool:
            futures = {pool.submit(_do_txt, p): p for p in txt_files}
            ordered = {p: f for f, p in futures.items()}
            for txt_path in txt_files:
                results.append(ordered[txt_path].result())

    if args.only_matches:
        results_to_report = [r for r in results if r.has_matches]
    else:
        results_to_report = results

    print_summary(results_to_report)

    # --- Update CSV ---
    if not args.no_csv_update:
        update_csv(args.csv, results)

    # --- Write JSON ---
    if args.output:
        output_data = [
            {
                "filename": r.filename,
                "path": r.path,
                "converted": r.converted,
                "gnps_task_ids": r.all_gnps_task_ids,
                "gnps_methods": r.all_gnps_methods,
                "zenodo_ids": r.all_zenodo_ids,
                "massive_ids": r.all_massive_ids,
                "error": r.error,
                "gnps_hits": [
                    {"task_id": h.task_id, "method": h.method, "context": h.context}
                    for h in r.gnps_hits
                ],
                "zenodo_matches": [
                    {"matched_text": z.matched_text, "context": z.context, "page": z.page}
                    for z in r.zenodo_matches
                ],
                "massive_matches": [
                    {"matched_text": m.matched_text, "context": m.context, "page": m.page}
                    for m in r.massive_matches
                ],
            }
            for r in results_to_report
        ]
        args.output.write_text(json.dumps(output_data, indent=2, ensure_ascii=False))
        print(f"JSON results written to: {args.output}\n")


if __name__ == "__main__":
    main()