PlantAI-rag-new/agent/legacy-rag/scripts/run_methodology_batch.py

#!/usr/bin/env python3
"""Run methodology extraction and bbav5 alignment for a paper CSV."""

from __future__ import annotations

import argparse
import csv
import json
import os
import re
import sys
from dataclasses import asdict, replace
from datetime import datetime
from pathlib import Path
from typing import Any

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from src.rag.unified.methodology import MethodEvidenceRef, MethodologyAligner, MethodObservation, ToolUsage
from src.rag.unified.methodology_extractor import ParsedDocumentMethodologyExtractor
from src.rag.unified.methodology_reviewer import MethodologyReviewer
from src.rag.unified.parsed_document import ParsedDocumentLoader
from src.rag.unified.bbav5_capability import BBAV5CapabilityCatalogBuilder
from src.rag.unified.bbav5_methodology_config import build_methodology_execution_plan


PAPER_FIELDS = [
    "paper_id",
    "family",
    "source_paper_role",
    "parsed_document_path",
    "pmid",
    "doi",
    "priority",
    "notes",
]

REVIEW_FIELDS = [
    "paper_id",
    "family",
    "observation_id",
    "normalized_stage",
    "method_type",
    "experiment_type",
    "validation_target",
    "biological_material",
    "treatment_or_condition",
    "manual_review_hint",
    "tools",
    "versions",
    "parameters",
    "inputs",
    "outputs",
    "bbav5_stage_id",
    "bbav5_default_tool",
    "replacement_tools",
    "support_status",
    "evidence_sentence",
    "evidence_sentences",
    "evidence_support_status",
    "unsupported_tools",
    "missing_tools",
    "missing_parameters",
    "evidence_repair_status",
    "evidence_quality_status",
    "evidence_quality_reason",
    "evidence_source_section",
    "v15_review_status",
    "v15_review_reason",
    "v15_review_suggestions",
    "section",
    "raw_method_text",
    "review_status",
    "stage_review",
    "tools_review",
    "versions_review",
    "parameters_review",
    "inputs_outputs_review",
    "bbav5_review",
    "bbav5_stage_review",
    "bbav5_tools_review",
    "support_status_review",
    "corrected_stage_value",
    "corrected_tools_value",
    "corrected_versions_value",
    "corrected_parameters_value",
    "corrected_evidence_value",
    "corrected_inputs_value",
    "corrected_outputs_value",
    "corrected_bbav5_stage_value",
    "corrected_bbav5_tools_value",
    "corrected_support_status_value",
    "corrected_stage",
    "corrected_tools",
    "corrected_parameters",
    "corrected_inputs",
    "corrected_outputs",
    "corrected_bbav5_stage",
    "corrected_support_status",
    "evidence_review",
    "manual_note",
]

STATUS_FIELDS = [
    "paper_id",
    "family",
    "source_paper_role",
    "parsed_document_path",
    "parsed_document_exists",
    "parser",
    "chunk_count",
    "methods_chunk_count",
    "methods_text_length",
    "observation_count",
    "ready_for_review",
    "failure_reason",
]

DISCOVERED_TOOL_FIELDS = [
    "tool_name",
    "paper_id",
    "family",
    "observation_id",
    "stage_guess",
    "role_guess",
    "version",
    "url",
    "evidence_sentence",
    "review_status",
    "accepted_stage",
    "accepted_role",
    "bbav5_stage",
    "support_status",
    "manual_note",
]

TOOL_AUDIT_FIELDS = [
    "paper_id",
    "family",
    "observation_id",
    "normalized_stage",
    "method_type",
    "tool_name",
    "version",
    "url",
    "role",
    "parameters",
    "bbav5_stage_id",
    "support_status",
    "evidence_sentence",
    "evidence_support_status",
    "section",
]

PARAMETER_AUDIT_FIELDS = [
    "paper_id",
    "family",
    "observation_id",
    "normalized_stage",
    "method_type",
    "component_type",
    "component_tool",
    "parameter_name",
    "parameter_value",
    "parameter_type",
    "evidence_sentence",
    "section",
]

URL_AUDIT_FIELDS = [
    "paper_id",
    "family",
    "observation_id",
    "normalized_stage",
    "source_name",
    "source_type",
    "url",
    "evidence_sentence",
    "section",
]

REVIEW_ISSUE_FIELDS = [
    "paper_id",
    "observation_id",
    "normalized_stage",
    "field",
    "severity",
    "issue_type",
    "message",
    "current_value",
    "suggested_value",
    "evidence_sentence",
    "section",
    "recommendation",
]

SPECIES_SUMMARY_FIELDS = [
    "paper_id",
    "family",
    "species_name",
    "species_role",
    "bbav5_role",
    "file_scope",
    "reference_use",
    "evidence_source",
    "observation_id",
    "normalized_stage",
    "evidence_sentence",
    "confidence",
    "actionable",
]

KNOWN_TOOL_NAMES = {
    "BLAST",
    "HMMER",
    "Pfam",
    "SMART",
    "NCBI CDD",
    "InterProScan",
    "ExPASy",
    "ProtParam",
    "ClustalX",
    "ClustalW",
    "MUSCLE",
    "MAFFT",
    "MEGA",
    "IQ-TREE",
    "RAxML",
    "PhyML",
    "FastTree",
    "MEME",
    "MCScanX",
    "JCVI",
    "TBtools",
    "GSDS",
    "qRT-PCR",
    "Primer3",
    "TopHat",
    "Cufflinks",
    "Cuffcompare",
    "HISAT2",
    "STAR",
    "Bowtie2",
    "DESeq",
    "DESeq2",
    "edgeR",
    "limma",
    "FastQC",
    "Trimmomatic",
    "Cutadapt",
    "featureCounts",
    "StringTie",
    "Salmon",
    "Kallisto",
    "HTSeq",
    "PlantCARE",
    "PLACE",
    "PlantPAN",
    "KOBAS",
    "GO",
    "KEGG",
    "STRING",
    "KaKs_Calculator",
    "ParaAT",
    "BLASTp",
    "HMM",
    "Hidden Markov Model",
    "UniProt",
}

SPECIES_ALIASES = {
    "broomcorn millet": "Panicum miliaceum",
    "common millet": "Panicum miliaceum",
    "hog millet": "Panicum miliaceum",
    "panicum miliaceum": "Panicum miliaceum",
    "proso millet": "Panicum miliaceum",
    "pmnac": "Panicum miliaceum",
    "tartary buckwheat": "Fagopyrum tataricum",
    "fagopyrum tataricum": "Fagopyrum tataricum",
    "ftspl": "Fagopyrum tataricum",
    "a. thaliana": "Arabidopsis thaliana",
    "atnac": "Arabidopsis thaliana",
    "atspl": "Arabidopsis thaliana",
    "arabidopsis": "Arabidopsis thaliana",
    "arabidopsis thaliana": "Arabidopsis thaliana",
    "b. vulgaris": "Beta vulgaris",
    "beta vulgaris": "Beta vulgaris",
    "g. max": "Glycine max",
    "glycine max": "Glycine max",
    "s. lycopersicum": "Solanum lycopersicum",
    "solanum lycopersicum": "Solanum lycopersicum",
    "tomato": "Solanum lycopersicum",
    "v. vinifera": "Vitis vinifera",
    "vitis vinifera": "Vitis vinifera",
    "grape": "Vitis vinifera",
    "o. sativa": "Oryza sativa",
    "oryza sativa": "Oryza sativa",
    "rice": "Oryza sativa",
    "osnac": "Oryza sativa",
    "pecan": "Carya illinoinensis",
    "carya illinoinensis": "Carya illinoinensis",
    "cispl": "Carya illinoinensis",
    "sunflower": "Helianthus annuus",
    "helianthus annuus": "Helianthus annuus",
    "citrus": "Citrus",
    "brassica rapa": "Brassica rapa",
    "foxtail millet": "Setaria italica",
    "setaria italica": "Setaria italica",
    "maize": "Zea mays",
    "zea mays": "Zea mays",
    "sorghum": "Sorghum bicolor",
    "sorghum bicolor": "Sorghum bicolor",
    "litchi": "Litchi chinensis",
    "litchi chinensis": "Litchi chinensis",
    "longan": "Dimocarpus longan",
    "dimocarpus longan": "Dimocarpus longan",
    "rambutan": "Nephelium lappaceum",
    "nephelium lappaceum": "Nephelium lappaceum",
}

TARGET_SPECIES_ALIAS_PATTERNS = re.compile(
    r"\b(tartary buckwheat|pecan|sunflower|foxtail millet|litchi|rambutan|longan|citrus|"
    r"brassica|broomcorn millet|common millet|hog millet|proso millet|panicum miliaceum)\b",
    re.I,
)

NON_TOOL_TERMS = {
    "Project",
    "Genome",
    "Tartary Buckwheat Genome",
    "SPL",
    "SBP",
    "PF03110",
    "Pinku1",
    "After",
    "Additional",
    "Figure",
    "Table",
    "Methods",
    "Results",
    "FtSPL",
    "Arabidopsis",
    "Abscisic",
    "Beta",
    "Capsicum",
    "Coding",
    "Multiple",
    "From",
    "DNA",
    "MVs",
    "PIs",
    "N-J",
    "The PIs",
    "The SPL",
    "Service",
    "Gene Structure Display",
    "CJ-Chen",
    "DEGs",
    "SUTs",
    "The SUTs",
}

KNOWN_NON_TOOL_PATTERNS = [
    r"^Ft[A-Z0-9]",
    r"^[A-Z][a-z]+[A-Z0-9]",
    r"^[A-Z][a-z]+$",
    r"^[A-Z]{2,4}$",
    r"^PF\d+",
    r"^\d",
    r".*\d+\.\d+$",
]


def resolve(path: str) -> Path:
    p = Path(path)
    return p if p.is_absolute() else PROJECT_ROOT / p


def load_or_build_capability_catalog(path: Path) -> dict[str, Any]:
    if path.exists():
        return json.loads(path.read_text(encoding="utf-8"))

    bbav5_root_value = os.environ.get("BBAV5_ROOT", "").strip()
    if bbav5_root_value:
        bbav5_root = Path(bbav5_root_value).expanduser()
        try:
            catalog = BBAV5CapabilityCatalogBuilder(bbav5_root).build()
            catalog_payload = catalog.to_dict()
            try:
                catalog.write_json(path)
            except OSError as write_exc:
                print(f"WARNING: capability catalog built but could not be cached at {path}: {write_exc}")
            return catalog_payload
        except Exception as exc:
            print(f"WARNING: failed to build bbav5 capability catalog from BBAV5_ROOT={bbav5_root}: {exc}")

    print(f"WARNING: capability catalog missing; continuing with empty catalog: {path}")
    return {"source_root": str(Path(bbav5_root_value).expanduser()) if bbav5_root_value else "", "stage_count": 0, "stages": {}}


def read_rows(path: Path) -> list[dict[str, str]]:
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        return [row for row in csv.DictReader(handle) if row.get("paper_id")]


def write_csv(path: Path, fields: list[str], rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        writer.writeheader()
        for row in rows:
            writer.writerow({field: row.get(field, "") for field in fields})


def dump_json(path: Path, value: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")


def observation_to_output_dict(obs: Any) -> dict[str, Any]:
    data = asdict(obs)
    data["parameters"] = clean_method_parameters(data.get("parameters") or {})
    for tool in data.get("tools") or []:
        if isinstance(tool, dict):
            tool["parameters"] = clean_method_parameters(tool.get("parameters") or {})
    evidence_support_status, unsupported_tools = evidence_support_for_obs(obs)
    data["evidence_sentences"] = evidence_sentences_for_obs(obs)
    data["evidence_support_status"] = evidence_support_status
    data["unsupported_tools"] = unsupported_tools
    data.setdefault("missing_tools", getattr(obs, "missing_tools", []))
    data.setdefault("missing_parameters", getattr(obs, "missing_parameters", []))
    data.setdefault("method_type", getattr(obs, "method_type", "computational_method"))
    data.setdefault("experiment_type", getattr(obs, "experiment_type", ""))
    data.setdefault("validation_target", getattr(obs, "validation_target", ""))
    data.setdefault("biological_material", getattr(obs, "biological_material", ""))
    data.setdefault("treatment_or_condition", getattr(obs, "treatment_or_condition", ""))
    data.setdefault("manual_review_hint", getattr(obs, "manual_review_hint", ""))
    data.setdefault("evidence_repair_status", getattr(obs, "evidence_repair_status", ""))
    data.setdefault("evidence_quality_status", getattr(obs, "evidence_quality_status", ""))
    data.setdefault("evidence_quality_reason", getattr(obs, "evidence_quality_reason", ""))
    data.setdefault("evidence_source_section", getattr(obs, "evidence_source_section", ""))
    data.setdefault("v15_review_status", getattr(obs, "v15_review_status", ""))
    data.setdefault("v15_review_reason", getattr(obs, "v15_review_reason", ""))
    data.setdefault("v15_review_suggestions", getattr(obs, "v15_review_suggestions", []))
    return data


def compact_json(value: Any) -> str:
    return json.dumps(value, ensure_ascii=False, sort_keys=True)


def timestamp() -> str:
    return datetime.now().isoformat(timespec="seconds")



def patch_external_job_status(
    status_json_path: str,
    substage: str,
    message: str,
    *,
    stage: str | None = None,
    extra: dict[str, Any] | None = None,
) -> None:
    """Merge progress and optional summary fields into upload-job status.json."""
    path_str = (status_json_path or "").strip()
    if not path_str:
        return
    path = Path(path_str)
    if not path.is_file():
        return
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        data["substage"] = substage
        data["message"] = message
        if stage is not None:
            data["stage"] = stage
        if extra:
            data.update(extra)
        data["updated_at"] = datetime.now().isoformat(timespec="seconds")
        path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    except Exception:
        pass


def append_log(path: Path, message: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(f"{timestamp()} {message}\n")
        handle.flush()


def append_jsonl(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    row = {"timestamp": timestamp(), **payload}
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")
        handle.flush()


def print_log(path: Path, message: str) -> None:
    print(message, flush=True)
    append_log(path, message)


def write_tool_lexicon(path: Path, rows: list[dict[str, Any]]) -> None:
    """Persist reviewed and high-confidence discovered tools for the next run."""
    import re

    lexicon = []
    seen = set()
    for row in rows:
        name = str(row.get("tool_name") or "").strip()
        if not name:
            continue
        status = str(row.get("review_status") or "").strip().lower()
        manual_stage = str(row.get("accepted_stage") or "").strip()
        stage = manual_stage or str(row.get("stage_guess") or row.get("normalized_stage") or row.get("bbav5_stage") or row.get("bbav5_stage_id") or "").strip()
        role = str(row.get("accepted_role") or row.get("role_guess") or row.get("role") or "").strip()
        support = str(row.get("support_status") or "").strip()
        norm = re.sub(r"[^a-z0-9]+", "", name.lower())
        if (
            not norm
            or norm in seen
            or len(name) < 3
            or len(name) > 80
            or re.search(r"^PF\d+|^\d|^(SPL|SBP|DNA|RNA|CDS|DEGs?|Methods?|Results?)$", name, re.I)
        ):
            continue
        accepted = status in {"accept", "accepted", "yes", "keep"}
        high_confidence = bool(stage and support in {"supported_default", "supported_alternative", "partially_supported", "pending", ""})
        if not accepted and not high_confidence:
            continue
        seen.add(norm)
        lexicon.append(
            {
                "tool_name": name,
                "stage": stage,
                "role": role,
                "bbav5_stage_id": str(row.get("bbav5_stage_id") or row.get("bbav5_stage") or "").strip(),
                "support_status": support,
                "source": "manual_review" if accepted or row.get("normalized_stage") else "auto_discovery",
            }
        )
    dump_json(path, {"tools": lexicon, "tool_count": len(lexicon)})


def load_llm_discoveries(path: Path | None) -> list[dict[str, Any]]:
    if not path:
        return []
    if not path.exists():
        raise FileNotFoundError(f"LLM discovery JSON not found: {path}")
    data = json.loads(path.read_text(encoding="utf-8"))
    rows = data.get("tools", data) if isinstance(data, dict) else data
    if not isinstance(rows, list):
        raise ValueError("LLM discovery JSON must be a list or an object with a 'tools' list")
    normalized = []
    for item in rows:
        if not isinstance(item, dict) or not item.get("tool_name"):
            continue
        normalized.append({field: item.get(field, "") for field in DISCOVERED_TOOL_FIELDS})
    return normalized


def is_methods_section(section: str) -> bool:
    if __import__("re").search(
        r"\b(availability of data|data availability|supplementary|additional files?|abbreviations|acknowledg|author'?s? contribution|funding|ethics|consent|competing interests?|references)\b",
        section or "",
        __import__("re").I,
    ):
        return False
    return bool(
        __import__("re").search(
            r"\b(methods?|materials|experimental|methodology|sequencing|analysis)\b",
            section or "",
            __import__("re").I,
        )
    )


def normalize_observations(observations: list[Any], family: str) -> list[Any]:
    """Remove non-method tail observations and correct obvious family-name bleed-through."""
    import dataclasses
    import re

    normalized = []
    current_family = (family or "").strip()
    wrong_family_pattern = re.compile(r"\b(AP2\s*/\s*ERF|AP2/ERF)\b", re.I)
    for obs in observations:
        evidence = obs.evidence[0] if obs.evidence else None
        section = evidence.section if evidence else ""
        text = " ".join(
            [
                section or "",
                evidence.evidence_sentence if evidence else "",
                obs.raw_method_text or "",
                " ".join(obs.outputs or []),
            ]
        )
        is_tail = re.search(
            r"\b(availability of data|data availability|datasets supporting|included in the article|authors'? contributions?|competing interests?|publisher'?s note)\b",
            text,
            re.I,
        )
        if is_tail:
            continue
        if current_family.upper() == "SPL" and wrong_family_pattern.search(text):
            fixed_tools = obs.tools
            fixed_parameters = _replace_family_terms(obs.parameters, wrong_family_pattern, "SPL")
            fixed_inputs = [_replace_family_terms(value, wrong_family_pattern, "SPL") for value in obs.inputs]
            fixed_outputs = [_replace_family_terms(value, wrong_family_pattern, "SPL") for value in obs.outputs]
            fixed_evidence = [
                dataclasses.replace(
                    ev,
                    evidence_sentence=_replace_family_terms(ev.evidence_sentence, wrong_family_pattern, "SPL"),
                )
                for ev in obs.evidence
            ]
            obs = dataclasses.replace(
                obs,
                tools=fixed_tools,
                parameters=fixed_parameters,
                inputs=fixed_inputs,
                outputs=fixed_outputs,
                evidence=fixed_evidence,
                raw_method_text=_replace_family_terms(obs.raw_method_text, wrong_family_pattern, "SPL"),
            )
        normalized.append(obs)
    return normalized


def _replace_family_terms(value: Any, pattern: Any, replacement: str) -> Any:
    if isinstance(value, str):
        return pattern.sub(replacement, value)
    if isinstance(value, list):
        return [_replace_family_terms(item, pattern, replacement) for item in value]
    if isinstance(value, dict):
        return {key: _replace_family_terms(item, pattern, replacement) for key, item in value.items()}
    return value



def clean_method_parameters(value: Any) -> Any:
    """Remove supplemental evidence payloads from parameter-like structures."""
    if isinstance(value, dict):
        cleaned: dict[str, Any] = {}
        for key, item in value.items():
            if str(key) == "supplemental_evidence" or _is_tool_version_parameter(str(key), item):
                continue
            cleaned_item = clean_method_parameters(item)
            if cleaned_item not in ("", None, [], {}):
                cleaned[key] = cleaned_item
        return cleaned
    if isinstance(value, list):
        cleaned_list = []
        for item in value:
            cleaned_item = clean_method_parameters(item)
            if cleaned_item not in ("", None, [], {}):
                cleaned_list.append(cleaned_item)
        return cleaned_list
    return value


def _is_tool_version_parameter(key: str, value: Any) -> bool:
    key_norm = _norm(key)
    if key_norm in {"version", "toolversion", "softwareversion", "programversion"}:
        return True
    if key_norm.endswith("version") and key_norm not in {"genomeversion", "annotationversion"}:
        return True
    value_text = str(value or "").strip()
    if re.fullmatch(r"v?\d+(?:\.\d+){0,4}[a-z]?", value_text, re.I) and "version" in key_norm:
        return True
    return False


def clean_observation_parameters(obs: Any) -> Any:
    tools = [replace(tool, parameters=clean_method_parameters(tool.parameters or {})) for tool in obs.tools]
    return replace(obs, tools=tools, parameters=clean_method_parameters(obs.parameters or {}))


def final_dedupe_observations(observations: list[Any]) -> tuple[list[Any], list[dict[str, Any]]]:
    """Final high-confidence same-paper/same-stage dedupe before batch files are written."""
    selected: list[Any] = []
    report: list[dict[str, Any]] = []
    for obs in observations:
        obs = clean_observation_parameters(obs)
        duplicate_index = _final_duplicate_target_index(obs, selected)
        if duplicate_index is None:
            selected.append(obs)
            continue
        primary = selected[duplicate_index]
        if _final_observation_rank(obs) > _final_observation_rank(primary):
            merged = _merge_final_duplicate(obs, primary)
            selected[duplicate_index] = merged
            primary_id = obs.observation_id
            removed_id = primary.observation_id
        else:
            merged = _merge_final_duplicate(primary, obs)
            selected[duplicate_index] = merged
            primary_id = primary.observation_id
            removed_id = obs.observation_id
        evidence_relation = _evidence_duplicate_relation(obs, primary) or _evidence_duplicate_relation(primary, obs) or "high_overlap"
        report.append(
            {
                "paper_id": obs.paper_id,
                "normalized_stage": obs.normalized_stage,
                "primary_observation_id": primary_id,
                "removed_observation_id": removed_id,
                "reason": evidence_relation,
                "evidence_overlap": f"{_token_overlap(_observation_text(obs), _observation_text(primary)):.3f}",
                "tool_overlap": f"{_tool_overlap(_tool_set(obs), _tool_set(primary)):.3f}",
            }
        )
    return selected, report


def _final_duplicate_target_index(obs: Any, selected: list[Any]) -> int | None:
    for index, primary in enumerate(selected):
        if obs.paper_id != primary.paper_id:
            continue
        if obs.normalized_stage != primary.normalized_stage:
            continue
        if _same_stage_duplicate_score(obs, primary) is None:
            continue
        return index
    return None


def _same_stage_duplicate_score(obs: Any, primary: Any) -> tuple[float, float] | None:
    left = _observation_text(obs)
    right = _observation_text(primary)
    relation = _evidence_duplicate_relation(obs, primary)
    evidence_overlap = _token_overlap(left, right)
    if relation is None and evidence_overlap < 0.72:
        return None
    left_tools = _tool_set(obs)
    right_tools = _tool_set(primary)
    tool_overlap = _tool_overlap(left_tools, right_tools)
    shorter, longer = (left_tools, right_tools) if len(left) <= len(right) else (right_tools, left_tools)
    tool_subset = bool(shorter) and shorter <= longer
    empty_weak = (not left_tools or not right_tools) and (_is_weak_context(obs) or _is_weak_context(primary))
    weak_same_text = (relation in {"contained_by_primary", "contains_primary"} or evidence_overlap >= 0.8) and (
        _is_weak_context(obs) or _is_weak_context(primary) or _is_rule_fallback(obs) or _is_rule_fallback(primary)
    )
    if tool_overlap >= 0.5 or tool_subset or empty_weak or weak_same_text:
        return (evidence_overlap, tool_overlap)
    return None


def _merge_final_duplicate(primary: Any, duplicate: Any) -> Any:
    tools = _merge_tools(primary.tools, duplicate.tools)
    params = clean_method_parameters(dict(primary.parameters or {}))
    for key, value in clean_method_parameters(duplicate.parameters or {}).items():
        if key not in params and _is_stage_parameter_like(primary.normalized_stage, key, value):
            params[key] = value
    inputs = _merge_scalar_values(primary.inputs, duplicate.inputs)
    outputs = _merge_scalar_values(primary.outputs, duplicate.outputs)
    evidence = list(primary.evidence)
    known = {(item.section, item.evidence_sentence, item.chunk_id) for item in evidence}
    for item in duplicate.evidence:
        key = (item.section, item.evidence_sentence, item.chunk_id)
        if key not in known:
            evidence.append(item)
            known.add(key)
    raw = primary.raw_method_text
    if len(duplicate.raw_method_text or "") > len(raw or "") and _looks_like_primary_context(duplicate):
        raw = duplicate.raw_method_text
    return replace(primary, raw_method_text=raw, tools=tools, parameters=params, inputs=inputs, outputs=outputs, evidence=evidence)


def _final_observation_rank(obs: Any) -> tuple[int, int, int, int, int, float, int]:
    return (
        1 if _looks_like_primary_context(obs) else 0,
        0 if _is_weak_context(obs) else 1,
        0 if _is_rule_fallback(obs) else 1,
        len(_tool_set(obs)),
        len(clean_method_parameters(obs.parameters or {})),
        float(getattr(obs, "confidence", 0) or 0),
        len(_observation_text(obs)),
    )


def _looks_like_primary_context(obs: Any) -> bool:
    section = " ".join(item.section or "" for item in obs.evidence)
    text = " ".join([section, obs.raw_method_text or ""])
    return bool(__import__("re").search(r"\b(methods?|materials|experimental|methodology|analysis)\b", text, __import__("re").I)) and not _is_weak_context(obs)


def _is_weak_context(obs: Any) -> bool:
    section = " ".join(item.section or "" for item in obs.evidence)
    text = " ".join([section, " ".join(item.evidence_sentence or "" for item in obs.evidence), obs.raw_method_text or ""])
    return bool(__import__("re").search(r"\b(results?|discussion|abstract|figure|fig\.?|table|supplement|additional file|abbreviations)\b", text, __import__("re").I))


def _is_rule_fallback(obs: Any) -> bool:
    return bool(__import__("re").search(r"_\d{3}$", obs.observation_id or ""))


def _observation_text(obs: Any) -> str:
    parts = [obs.raw_method_text or ""]
    parts.extend(item.evidence_sentence or "" for item in obs.evidence)
    parts.extend(obs.inputs or [])
    parts.extend(obs.outputs or [])
    return _compact_text(" ".join(parts))


def _evidence_duplicate_relation(left_obs: Any, right_obs: Any) -> str | None:
    left = _compact_text(_observation_text(left_obs)).lower()
    right = _compact_text(_observation_text(right_obs)).lower()
    if not left or not right:
        return None
    if len(left) <= len(right) and len(left) >= 80 and left in right:
        return "contained_by_primary"
    if len(right) < len(left) and len(right) >= 80 and right in left:
        return "contains_primary"
    if _token_overlap(left, right) >= 0.72:
        return "high_overlap"
    return None


def _token_overlap(left: str, right: str) -> float:
    tokens_left = set(__import__("re").findall(r"[a-zA-Z0-9]+", (left or "").lower()))
    tokens_right = set(__import__("re").findall(r"[a-zA-Z0-9]+", (right or "").lower()))
    if not tokens_left or not tokens_right:
        return 0.0
    return len(tokens_left & tokens_right) / len(tokens_left | tokens_right)


def _tool_overlap(left: set[str], right: set[str]) -> float:
    if not left and not right:
        return 0.0
    if not left or not right:
        return 0.0
    return len(left & right) / len(left | right)


def _tool_set(obs: Any) -> set[str]:
    return {_canonical_tool_name(tool.tool_name) for tool in obs.tools if _canonical_tool_name(tool.tool_name)}



def _norm(value: str) -> str:
    return "".join(ch.lower() for ch in str(value or "") if ch.isalnum())

def _canonical_tool_name(name: str) -> str:
    key = _norm(name)
    aliases = {
        "megax": "mega", "mega5": "mega", "mega6": "mega", "mega7": "mega", "mega11": "mega",
        "clustalomega": "clustal", "clustalw": "clustal", "clustalx": "clustal",
        "hmmer3": "hmmer", "hmmsearch": "hmmer", "pfamdatabase": "pfam",
        "stringdatabase": "string", "stringdb": "string", "plantcaredatabase": "plantcare",
        "gsds20": "gsds", "kakscalculator": "kaks_calculator",
    }
    return aliases.get(key, key)


def _tool_evidence_aliases(tool_name: str) -> list[str]:
    canonical = _canonical_tool_name(tool_name)
    aliases = {
        "gsds": ["GSDS", "Gene Structure Display Server"],
        "mega": ["MEGA", "MEGA X"],
        "clustal": ["ClustalW", "ClustalX", "Clustal Omega"],
        "hmmer": ["HMMER", "hmmsearch"],
        "pfam": ["Pfam", "PFAM"],
        "blast": ["BLAST", "BLASTP", "blastp", "blastn", "tblastn"],
        "meme": ["MEME", "MEME Suite"],
        "mcscanx": ["MCScanX"],
        "tbtools": ["TBtools", "TBtools-II"],
        "plantcare": ["PlantCARE"],
        "pal2nal": ["PAL2NAL", "pal2nal.pl"],
        "paml": ["PAML"],
        "codeml": ["codeml", "CodeML"],
        "muscle": ["MUSCLE"],
        "diverge": ["DIVERGE"],
        "genevestigator": ["GENEVESTIGATOR", "Genevestigator"],
    }
    values = [tool_name, canonical, *aliases.get(canonical, [])]
    result: list[str] = []
    seen: set[str] = set()
    for value in values:
        key = _norm(value)
        if key and key not in seen:
            result.append(str(value))
            seen.add(key)
    return result


def _text_mentions_tool(text: str, tool_name: str) -> bool:
    text = str(text or "")
    text_norm = _norm(text)
    for alias in _tool_evidence_aliases(tool_name):
        alias_norm = _norm(alias)
        if not alias_norm:
            continue
        if len(alias_norm) <= 2:
            if __import__("re").search(rf"(?<![A-Za-z0-9]){__import__('re').escape(alias)}(?![A-Za-z0-9])", text, __import__("re").I):
                return True
            continue
        if alias_norm in text_norm:
            return True
    return False


def evidence_sentences_for_obs(obs: Any) -> list[str]:
    for item in getattr(obs, "evidence", []) or []:
        sentence = _compact_text(getattr(item, "evidence_sentence", ""))
        if sentence:
            return [sentence]
    return []


def evidence_support_for_obs(obs: Any) -> tuple[str, list[str]]:
    evidence_text = " ".join(evidence_sentences_for_obs(obs) + [getattr(obs, "raw_method_text", "") or ""])
    unsupported = [
        tool.tool_name
        for tool in getattr(obs, "tools", []) or []
        if tool.tool_name and not _text_mentions_tool(evidence_text, tool.tool_name)
    ]
    tool_count = len(getattr(obs, "tools", []) or [])
    if not tool_count:
        return "supported", []
    if not unsupported:
        return "supported", []
    if len(unsupported) < tool_count:
        return "partially_supported", unsupported
    return "unresolved", unsupported


def evidence_sentence_for_tool(obs: Any, tool: Any) -> str:
    for sentence in evidence_sentences_for_obs(obs):
        if _text_mentions_tool(sentence, tool.tool_name):
            return sentence
    evidence = (getattr(obs, "evidence", []) or [None])[0]
    return getattr(evidence, "evidence_sentence", "") if evidence else ""


def section_for_tool_evidence(obs: Any, evidence_sentence: str) -> str:
    for item in getattr(obs, "evidence", []) or []:
        if getattr(item, "evidence_sentence", "") == evidence_sentence:
            return getattr(item, "section", "") or ""
    evidence = (getattr(obs, "evidence", []) or [None])[0]
    return getattr(evidence, "section", "") if evidence else ""


def _merge_tools(left: list[Any], right: list[Any]) -> list[Any]:
    result = list(left or [])
    by_name = {_canonical_tool_name(tool.tool_name): index for index, tool in enumerate(result)}
    for tool in right or []:
        key = _canonical_tool_name(tool.tool_name)
        if not key:
            continue
        cleaned = replace(tool, parameters=clean_method_parameters(tool.parameters or {}))
        if key not in by_name:
            by_name[key] = len(result)
            result.append(cleaned)
            continue
        current = result[by_name[key]]
        params = dict(current.parameters or {})
        for pkey, pvalue in clean_method_parameters(tool.parameters or {}).items():
            params.setdefault(pkey, pvalue)
        version = current.version or tool.version
        role = current.role or tool.role
        result[by_name[key]] = replace(current, version=version, role=role, parameters=params)
    return result


def _merge_scalar_values(left: list[Any], right: list[Any]) -> list[Any]:
    result = list(left or [])
    seen = {str(item) for item in result}
    for item in right or []:
        key = str(item)
        if key not in seen:
            result.append(item)
            seen.add(key)
    return result


def _is_stage_parameter_like(stage: str, key: str, value: Any) -> bool:
    if str(key) == "supplemental_evidence":
        return False
    value_text = json.dumps(value, ensure_ascii=False) if not isinstance(value, str) else value
    if len(value_text) > 600:
        return False
    if __import__("re").search(r"\b(chunk_id|evidence_sentence|observation_id)\b", f"{key} {value_text}", __import__("re").I):
        return False
    return True


def _compact_text(text: str) -> str:
    return " ".join(str(text or "").split())

def _species_regex(alias: str) -> str:
    escaped = re.escape(alias)
    if re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]*", alias or ""):
        return rf"(?<![A-Za-z0-9]){escaped}(?![A-Za-z0-9])"
    return rf"(?<![A-Za-z]){escaped}(?![A-Za-z])"


def _species_canonical_name(name: str) -> str:
    clean = re.sub(r"\s+", " ", str(name or "")).strip()
    clean = re.sub(r"\s+(?:L\.|L|Mill\.|Bunge|Moench|Merr\.)$", "", clean).strip()
    return clean


def _species_aliases_from_text(text: str) -> dict[str, str]:
    aliases = dict(SPECIES_ALIASES)
    pair_pattern = re.compile(
        r"\b([A-Za-z][A-Za-z-]*(?:\s+[A-Za-z][A-Za-z-]*){0,3})\s*"
        r"\(([A-Z][a-z]+\s+[a-z][a-z-]+(?:\s+(?:L\.|L|Mill\.|Bunge|Moench|Merr\.))?)\)",
    )
    bad_common = {"open access", "check for", "additional file", "figure", "table"}
    for common, latin in pair_pattern.findall(text or ""):
        common_clean = re.sub(r"\s+", " ", common).strip(" .,:;()[]").lower()
        canonical = _species_canonical_name(latin)
        if not common_clean or common_clean in bad_common:
            continue
        if len(common_clean) > 60:
            continue
        aliases[common_clean] = canonical
        aliases[canonical.lower()] = canonical
    return aliases


def _species_evidence_sources(observations: list[Any], document: Any | None) -> list[dict[str, str]]:
    sources: list[dict[str, str]] = []
    if document is not None:
        for chunk in (document.chunks or [])[:30]:
            clean = _compact_text(chunk.text)
            if not (20 <= len(clean) <= 260):
                continue
            if re.search(r"\b(abstract|background|methods?|results?|discussion|fig\.?|table)\b", clean, re.I):
                continue
            if re.search(r"\b(genome-wide|identification|analysis|expression analysis)\b.{0,160}\b(in|of|from)\b", clean, re.I):
                sources.append({"source": "title", "text": clean, "section": chunk.section or "front_matter", "observation_id": "", "stage": ""})
                break
        for chunk in document.chunks or []:
            clean = _compact_text(chunk.text)
            if not clean:
                continue
            section = str(chunk.section or "")
            if re.search(r"\b(abstract|background)\b", section, re.I):
                sources.append({"source": "abstract", "text": clean, "section": section, "observation_id": "", "stage": ""})
            elif is_methods_section(section):
                sources.append({"source": "methods", "text": clean, "section": section, "observation_id": "", "stage": ""})
    for obs in observations:
        evidence = obs.evidence[0] if obs.evidence else None
        text = _compact_text(
            " ".join(
                [
                    evidence.evidence_sentence if evidence else "",
                    " ".join(obs.inputs or []),
                    " ".join(obs.outputs or []),
                    compact_json(obs.parameters),
                    obs.raw_method_text or "",
                ]
            )
        )
        if text:
            sources.append(
                {
                    "source": "method_observation",
                    "text": text,
                    "section": evidence.section if evidence else "Methods",
                    "observation_id": obs.observation_id,
                    "stage": obs.normalized_stage,
                }
            )
    return sources


def _is_background_species_context(text: str) -> bool:
    return bool(
        re.search(
            r"\b(?:such as|including|many species|various plant species|studied in many species|has been widely studied)\b",
            text,
            re.I,
        )
    )


def _species_role_for_context(alias: str, canonical: str, text: str, source: str) -> str:
    lower = text.lower()
    alias_pattern = _species_regex(alias)
    alias_match = re.search(alias_pattern, text, re.I)
    if not alias_match:
        return ""
    mention = re.escape(alias)
    ref_context = (
        re.search(rf"\bdatabases?\s+of\s+{mention}", lower, re.I)
        or re.search(rf"\bclassification\s+(?:from|of)\s+{mention}", lower, re.I)
        or re.search(rf"{mention}.{{0,90}}\b(?:used as quer(?:y|ies)|as quer(?:y|ies)|classification|TAIR|Rice Genome Annotation Project)\b", text, re.I)
        or (re.search(r"\b(?:At[A-Z0-9]+|Os[A-Z0-9]+)\b", text) and canonical in {"Arabidopsis thaliana", "Oryza sativa"})
    )
    if ref_context:
        return "reference_species"

    reference_model_species = {"Arabidopsis thaliana", "Oryza sativa"}
    explicit_model_target = (
        source == "title"
        or re.search(rf"\b(?:genome|chromosomes|genes|proteins)\s+of\s+{mention}", lower, re.I)
        or re.search(rf"{mention}.{{0,60}}\b(?:genome|chromosomes|genes|proteins)\b", lower, re.I)
        or re.search(rf"\bidentified\b.{{0,80}}\b(?:in|from)\s+{mention}", lower, re.I)
    )
    if canonical in reference_model_species and not explicit_model_target:
        return ""

    explicit_target_context = (
        (source == "title" and re.search(rf"\b(?:in|from|of|for)\s+(?:the\s+)?{mention}\b", lower, re.I))
        or re.search(rf"\b(?:in|from|for)\s+(?:the\s+)?{mention}\b", lower, re.I)
        or re.search(rf"\b(?:genome|chromosomes|genes|proteins|seedlings|cultivar)\b.{{0,90}}{mention}", lower, re.I)
        or re.search(rf"{mention}.{{0,90}}\b(?:genome|chromosomes|genes|proteins|seedlings|cultivar|was downloaded|were identified)\b", lower, re.I)
        or (canonical not in {"Arabidopsis thaliana", "Oryza sativa"} and re.search(r"\b(?:identified|obtained|named|mapped|distributed)\b.{0,120}\b(?:PmNAC|CiSPL|FtSPL|SiHDZ)", text, re.I))
    )
    if explicit_target_context:
        return "target_species"
    if _is_background_species_context(text):
        return ""
    if TARGET_SPECIES_ALIAS_PATTERNS.search(alias):
        return "target_species"
    return ""


def infer_species_rows(paper_id: str, family: str, observations: list[Any], document: Any | None = None) -> list[dict[str, Any]]:
    sources = _species_evidence_sources(observations, document)
    corpus = " ".join(source["text"] for source in sources)
    aliases = _species_aliases_from_text(corpus)
    rows: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str, str]] = set()

    source_rank = {"title": 0, "abstract": 1, "methods": 2, "method_observation": 3}
    for source in sorted(sources, key=lambda item: source_rank.get(item["source"], 9)):
        text = source["text"]
        if not text:
            continue
        for alias, canonical_raw in aliases.items():
            canonical = _species_canonical_name(canonical_raw)
            if not canonical or not re.search(_species_regex(alias), text, re.I):
                continue
            role = _species_role_for_context(alias, canonical, text, source["source"])
            if not role:
                continue
            key = (canonical, role, source.get("observation_id", ""), source["source"])
            if key in seen:
                continue
            seen.add(key)
            rows.append(_species_summary_row(paper_id, family, canonical, role, source, text))

    rows = _resolve_species_role_conflicts(rows)
    if any(row["species_role"] == "target_species" for row in rows):
        return rows

    # Last non-species-specific rule: if the title contains exactly one common-name/Latin-name pair, it is the target species.
    for source in sources:
        if source["source"] != "title":
            continue
        title_aliases = _species_aliases_from_text(source["text"])
        pair_candidates = [name for alias, name in title_aliases.items() if alias not in SPECIES_ALIASES]
        unique = sorted({_species_canonical_name(name) for name in pair_candidates if name})
        if len(unique) == 1:
            rows.append(_species_summary_row(paper_id, family, unique[0], "target_species", {"source": "title", "observation_id": "", "stage": ""}, source["text"]))
            break
    return _resolve_species_role_conflicts(rows)


def _species_summary_row(paper_id: str, family: str, species_name: str, species_role: str, source: dict[str, str], evidence_text: str) -> dict[str, Any]:
    bbav5_role = _bbav5_role_for_species_context(species_role, evidence_text, source.get("source", ""), source.get("stage", ""))
    return {
        "paper_id": paper_id,
        "family": family,
        "species_name": species_name,
        "species_role": species_role,
        "bbav5_role": bbav5_role,
        "file_scope": _bbav5_role_file_scope(bbav5_role),
        "reference_use": _bbav5_role_reference_use(bbav5_role),
        "evidence_source": source.get("source", ""),
        "observation_id": source.get("observation_id", ""),
        "normalized_stage": source.get("stage", ""),
        "evidence_sentence": str(evidence_text or "")[:1000],
        "confidence": f"{_bbav5_role_confidence(bbav5_role, source.get('source', '')):.2f}",
        "actionable": "true" if bbav5_role not in {"background_only", "uncertain", ""} else "false",
    }


def _bbav5_role_for_species_context(species_role: str, text: str, source: str, stage: str) -> str:
    lower = " ".join([str(text or ""), str(stage or ""), str(source or "")]).lower()
    if _is_background_species_context(text):
        return "background_only"
    if species_role == "target_species":
        return "target"
    if species_role == "reference_species":
        if re.search(r"\bblastp?|quer(?:y|ies)|known .{0,40}protein|reference protein|identify", lower, re.I):
            return "identify_blast_reference"
        if re.search(r"\bsubfamily|classification|label|clade|phylogen", lower, re.I):
            return "primary_reference"
        return "primary_reference"
    return "uncertain"


def _bbav5_role_file_scope(role: str) -> str:
    if role in {"target", "comparative", "outgroup", "primary_reference"}:
        return "whole_genome"
    if role in {"identify_blast_reference", "seed_reference"}:
        return "known_family"
    return "none"


def _bbav5_role_reference_use(role: str) -> str:
    if role == "identify_blast_reference":
        return "identify_blast"
    if role == "seed_reference":
        return "both"
    if role == "primary_reference":
        return "downstream_primary"
    return "none"


def _bbav5_role_confidence(role: str, source: str) -> float:
    if role == "background_only":
        return 0.35
    if source == "title":
        return 0.90
    if source in {"methods", "method_observation"}:
        return 0.80
    return 0.60


def _resolve_species_role_conflicts(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    target_names = {row["species_name"] for row in rows if row.get("species_role") == "target_species"}
    non_model_targets = target_names - {"Arabidopsis thaliana", "Oryza sativa"}
    filtered: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str, str]] = set()
    for row in rows:
        name = row.get("species_name", "")
        role = row.get("species_role", "")
        if role == "reference_species" and name in target_names:
            continue
        if role == "target_species" and non_model_targets and name in {"Arabidopsis thaliana", "Oryza sativa"}:
            continue
        key = (name, role, row.get("observation_id", ""), row.get("evidence_source", ""))
        if key in seen:
            continue
        seen.add(key)
        filtered.append(row)
    return filtered


def tool_audit_rows(paper_id: str, family: str, observations: list[Any], alignments_by_id: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for obs in observations:
        alignment = alignments_by_id.get(obs.observation_id)
        evidence = obs.evidence[0] if obs.evidence else None
        for tool in obs.tools:
            params = tool.parameters or {}
            tool_evidence = evidence_sentence_for_tool(obs, tool)
            rows.append(
                {
                    "paper_id": paper_id,
                    "family": family,
                    "observation_id": obs.observation_id,
                    "normalized_stage": obs.normalized_stage,
                    "method_type": getattr(obs, "method_type", "computational_method"),
                    "tool_name": tool.tool_name,
                    "version": tool.version or "",
                    "url": params.get("url", ""),
                    "role": tool.role or "",
                    "parameters": compact_json(params),
                    "bbav5_stage_id": alignment.bbav5_stage_id if alignment else "",
                    "support_status": alignment.support_status if alignment else "",
                    "evidence_sentence": tool_evidence,
                    "evidence_support_status": "supported" if tool_evidence and _text_mentions_tool(tool_evidence, tool.tool_name) else "unresolved",
                    "section": section_for_tool_evidence(obs, tool_evidence) or (evidence.section if evidence else ""),
                }
            )
    return rows


def parameter_audit_rows(paper_id: str, family: str, observations: list[Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for obs in observations:
        evidence = obs.evidence[0] if obs.evidence else None
        if getattr(obs, "normalized_stage", "") == "gene_family_identification":
            rows.extend(_identify_parameter_audit_rows(paper_id, family, obs, evidence))
            continue

        for key, value in flatten_parameters(clean_method_parameters(obs.parameters or {})).items():
            rows.append(
                _parameter_audit_row(
                    paper_id,
                    family,
                    obs,
                    key,
                    value,
                    evidence.evidence_sentence if evidence else "",
                    evidence.section if evidence else "",
                )
            )
        for tool in obs.tools:
            for key, value in flatten_parameters(tool.parameters or {}).items():
                rows.append(
                    _parameter_audit_row(
                        paper_id,
                        family,
                        obs,
                        f"{tool.tool_name}.{key}",
                        value,
                        evidence.evidence_sentence if evidence else "",
                        evidence.section if evidence else "",
                    )
                )
    return rows


def _identify_parameter_audit_rows(paper_id: str, family: str, obs: Any, evidence: Any) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    evidence_sentence = evidence.evidence_sentence if evidence else ""
    section = evidence.section if evidence else ""
    emitted: set[tuple[str, str]] = set()

    for component in getattr(obs, "method_components", []) or []:
        if not isinstance(component, dict):
            continue
        component_type = str(component.get("component_type") or "").strip() or _component_type_for_tool_names(_component_tools(component))
        component_tool = _component_tool_display(component)
        component_evidence = str(component.get("evidence_sentence") or evidence_sentence)
        component_params = _component_parameters_for_audit(component)
        for key, value in flatten_parameters(component_params).items():
            parameter_name = _identify_component_parameter_name(component_type, component_tool, key)
            row_key = (parameter_name, compact_json(value) if isinstance(value, (dict, list)) else str(value))
            if row_key in emitted:
                continue
            emitted.add(row_key)
            rows.append(
                _parameter_audit_row(
                    paper_id,
                    family,
                    obs,
                    parameter_name,
                    value,
                    component_evidence,
                    section,
                    component_type=component_type,
                    component_tool=component_tool,
                )
            )

    for key, value in flatten_parameters(clean_method_parameters(obs.parameters or {})).items():
        parameter_name, component_type, component_tool = _identify_legacy_parameter_assignment(key)
        rows.append(
            _parameter_audit_row(
                paper_id,
                family,
                obs,
                parameter_name,
                value,
                evidence_sentence,
                section,
                component_type=component_type,
                component_tool=component_tool,
            )
        )

    for tool in obs.tools:
        component_type = _component_type_for_tool_names([tool.tool_name])
        component_tool = tool.tool_name if component_type else ""
        for key, value in flatten_parameters(tool.parameters or {}).items():
            parameter_name = _identify_component_parameter_name(component_type, component_tool, key) if component_type else f"{tool.tool_name}.{key}"
            rows.append(
                _parameter_audit_row(
                    paper_id,
                    family,
                    obs,
                    parameter_name,
                    value,
                    evidence_sentence,
                    section,
                    component_type=component_type,
                    component_tool=component_tool,
                )
            )
    return rows


def _parameter_audit_row(
    paper_id: str,
    family: str,
    obs: Any,
    parameter_name: str,
    value: Any,
    evidence_sentence: str,
    section: str,
    *,
    component_type: str = "",
    component_tool: str = "",
) -> dict[str, Any]:
    return {
        "paper_id": paper_id,
        "family": family,
        "observation_id": obs.observation_id,
        "normalized_stage": obs.normalized_stage,
        "method_type": getattr(obs, "method_type", "computational_method"),
        "component_type": component_type,
        "component_tool": component_tool,
        "parameter_name": parameter_name,
        "parameter_value": compact_json(value) if isinstance(value, (dict, list)) else value,
        "parameter_type": type(value).__name__,
        "evidence_sentence": evidence_sentence,
        "section": section,
    }


def _component_parameters_for_audit(component: dict[str, Any]) -> dict[str, Any]:
    params = clean_method_parameters(component.get("parameters") or {})
    params = dict(params) if isinstance(params, dict) else {}
    for source_key, target_key in (
        ("reference_species", "reference_species"),
        ("referenceSpecies", "reference_species"),
        ("reference_sets", "reference_sets"),
        ("referenceSets", "reference_sets"),
        ("target_database", "target_database"),
        ("targetDatabase", "target_database"),
    ):
        value = component.get(source_key)
        if value not in (None, "", [], {}):
            params.setdefault(target_key, value)
    return params


def _component_tools(component: dict[str, Any]) -> list[str]:
    tools: list[str] = []
    if component.get("tool"):
        tools.append(str(component.get("tool")))
    if component.get("tool_name"):
        tools.append(str(component.get("tool_name")))
    if isinstance(component.get("tools"), list):
        tools.extend(str(item) for item in component.get("tools") if str(item).strip())
    return tools


def _component_tool_display(component: dict[str, Any]) -> str:
    tools = []
    for tool in _component_tools(component):
        if tool and tool not in tools:
            tools.append(tool)
    return ";".join(tools)


def _component_type_for_tool_names(tools: list[str]) -> str:
    joined = " ".join(tools)
    if re.search(r"\bBLASTP?|blastp|blastn|tblastn\b", joined, re.I):
        return "blast_search"
    if re.search(r"\bHMMER|hmmsearch|Pfam\b", joined, re.I):
        return "hmm_search"
    if re.search(r"\bSMART|CDD|CD-Search|InterProScan\b", joined, re.I):
        return "domain_confirmation"
    return ""


def _identify_component_parameter_name(component_type: str, component_tool: str, key: str) -> str:
    text = str(key or "").strip()
    if not text:
        return text
    prefix, _, suffix = text.partition(".")
    prefix_norm = _norm(prefix)
    if prefix_norm in {"blast", "hmmer", "cdd", "confirmation", "domainconfirmation", "unassigned"} and suffix:
        return _normalize_scoped_identify_parameter(prefix, suffix)

    suffix = _normalize_identify_parameter_suffix(text)
    if component_type == "blast_search":
        return f"blast.{suffix}"
    if component_type == "hmm_search":
        return f"hmmer.{suffix}"
    if component_type == "domain_confirmation":
        if _norm(component_tool) in {"ncbicdd", "cdd", "cdsearch"} or suffix.lower() in {"evalue", "e_value"}:
            return f"cdd.{suffix}"
        return f"confirmation.{suffix}"
    if component_type == "unassigned":
        return f"unassigned.{text}"
    return text


def _identify_legacy_parameter_assignment(key: str) -> tuple[str, str, str]:
    text = str(key or "").strip()
    norm = _norm(text)
    if "." in text:
        prefix, suffix = text.split(".", 1)
        scoped = _normalize_scoped_identify_parameter(prefix, suffix)
        component_type = {
            "blast": "blast_search",
            "hmmer": "hmm_search",
            "cdd": "domain_confirmation",
            "domainConfirmation": "domain_confirmation",
            "confirmation": "domain_confirmation",
            "unassigned": "unassigned",
        }.get(scoped.split(".", 1)[0], "")
        return scoped, component_type, ""
    if norm in {"evalue", "evaluecutoff"} or text in {"e_value", "e-value", "E-value"}:
        return "unassigned.e_value", "unassigned", ""
    if norm in {"blastidentitythreshold", "blastidentity", "identitythreshold"}:
        return "blast.identityThreshold", "blast_search", ""
    if norm in {"blastevalue", "blastevaluecutoff"}:
        return "blast.evalue", "blast_search", ""
    if norm in {"hmmerevalue", "hmmerevaluecutoff", "hmmerthreshold", "hmmerethreshold"}:
        return "hmmer.evalue", "hmm_search", ""
    if norm in {"hmmercoverage", "hmmcoverage"}:
        return "hmmer.coverage", "hmm_search", ""
    if norm in {"cddevalue", "cddthreshold"}:
        return "cdd.evalue", "domain_confirmation", ""
    if norm in {"domain", "pfamdomain", "pfamdomains", "hmmerdomains", "domains"}:
        return "hmmer.domains", "hmm_search", ""
    return text, "", ""


def _normalize_scoped_identify_parameter(prefix: str, suffix: str) -> str:
    normalized_prefix = {
        "blast": "blast",
        "hmmer": "hmmer",
        "cdd": "cdd",
        "domainconfirmation": "domainConfirmation",
        "confirmation": "confirmation",
        "unassigned": "unassigned",
    }.get(_norm(prefix), prefix)
    return f"{normalized_prefix}.{_normalize_identify_parameter_suffix(suffix)}"


def _normalize_identify_parameter_suffix(value: str) -> str:
    norm = _norm(value)
    if norm in {"evalue", "evaluecutoff"}:
        return "evalue"
    if norm in {"identity", "identitythreshold", "percentidentity", "percentidentitythreshold"}:
        return "identityThreshold"
    if norm in {"referencespecies", "referencespecieslist"}:
        return "reference_species"
    if norm in {"referencesets", "referenceset", "referencequeries"}:
        return "reference_sets"
    if norm in {"targetdatabase", "targetdb"}:
        return "target_database"
    if norm in {"domains", "domain", "pfamdomains", "pfamdomain"}:
        return "domains"
    if norm in {"cddevalue", "cddevaluecutoff"}:
        return "evalue"
    return value


def flatten_parameters(value: Any, prefix: str = "") -> dict[str, Any]:
    if not isinstance(value, dict):
        return {prefix or "value": value} if value not in ("", None, [], {}) else {}
    rows: dict[str, Any] = {}
    for key, item in value.items():
        if str(key) == "supplemental_evidence":
            continue
        name = f"{prefix}.{key}" if prefix else str(key)
        if isinstance(item, dict):
            rows.update(flatten_parameters(item, name))
        else:
            rows[name] = item
    return rows


def url_audit_rows(paper_id: str, family: str, observations: list[Any]) -> list[dict[str, Any]]:
    import re

    rows: list[dict[str, Any]] = []
    seen = set()
    for obs in observations:
        evidence = obs.evidence[0] if obs.evidence else None
        text = " ".join([evidence.evidence_sentence if evidence else "", obs.raw_method_text or ""])
        for tool in obs.tools:
            url = (tool.parameters or {}).get("url", "")
            if url:
                key = (obs.observation_id, tool.tool_name, url)
                if key not in seen:
                    seen.add(key)
                    rows.append(
                        {
                            "paper_id": paper_id,
                            "family": family,
                            "observation_id": obs.observation_id,
                            "normalized_stage": obs.normalized_stage,
                            "source_name": tool.tool_name,
                            "source_type": "tool_parameter",
                            "url": url,
                            "evidence_sentence": evidence.evidence_sentence if evidence else "",
                            "section": evidence.section if evidence else "",
                        }
                    )
        for match in re.finditer(r"https?://[^\s,)]+|www\.[^\s,)]+", text):
            url = match.group(0).rstrip(".,;")
            key = (obs.observation_id, "", url)
            if key in seen:
                continue
            seen.add(key)
            rows.append(
                {
                    "paper_id": paper_id,
                    "family": family,
                    "observation_id": obs.observation_id,
                    "normalized_stage": obs.normalized_stage,
                    "source_name": "",
                    "source_type": "evidence_url",
                    "url": url,
                    "evidence_sentence": evidence.evidence_sentence if evidence else "",
                    "section": evidence.section if evidence else "",
                }
            )
    return rows


def build_review_rows(
    paper_id: str,
    family: str,
    observations: list[Any],
    alignments_by_id: dict[str, Any],
    catalog: dict[str, Any],
) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for obs in observations:
        alignment = alignments_by_id.get(obs.observation_id)
        evidence = obs.evidence[0] if obs.evidence else None
        evidence_sentences = evidence_sentences_for_obs(obs)
        evidence_support_status, unsupported_tools = evidence_support_for_obs(obs)
        default_tools = []
        if alignment and alignment.bbav5_stage_id:
            default_tools = catalog.get("stages", {}).get(alignment.bbav5_stage_id, {}).get("default_tools", [])
        rows.append(
            {
                "paper_id": paper_id,
                "family": family,
                "observation_id": obs.observation_id,
                "normalized_stage": obs.normalized_stage,
                "method_type": getattr(obs, "method_type", "computational_method"),
                "experiment_type": getattr(obs, "experiment_type", ""),
                "validation_target": getattr(obs, "validation_target", ""),
                "biological_material": getattr(obs, "biological_material", ""),
                "treatment_or_condition": getattr(obs, "treatment_or_condition", ""),
                "manual_review_hint": getattr(obs, "manual_review_hint", ""),
                "tools": ";".join(tool.tool_name for tool in obs.tools),
                "versions": ";".join(f"{tool.tool_name}:{tool.version}" for tool in obs.tools if tool.version),
                "parameters": compact_json(clean_method_parameters(obs.parameters or {})),
                "inputs": ";".join(obs.inputs),
                "outputs": ";".join(obs.outputs),
                "bbav5_stage_id": alignment.bbav5_stage_id if alignment else "",
                "bbav5_default_tool": ";".join(default_tools),
                "replacement_tools": ";".join(alignment.replacement_tools if alignment else []),
                "support_status": alignment.support_status if alignment else "",
                "evidence_sentence": evidence.evidence_sentence if evidence else "",
                "evidence_sentences": compact_json(evidence_sentences),
                "evidence_support_status": evidence_support_status,
                "unsupported_tools": ";".join(unsupported_tools),
                "missing_tools": ";".join(getattr(obs, "missing_tools", []) or []),
                "missing_parameters": ";".join(getattr(obs, "missing_parameters", []) or []),
                "evidence_repair_status": getattr(obs, "evidence_repair_status", ""),
                "evidence_quality_status": getattr(obs, "evidence_quality_status", ""),
                "evidence_quality_reason": getattr(obs, "evidence_quality_reason", ""),
                "evidence_source_section": getattr(obs, "evidence_source_section", ""),
                "v15_review_status": getattr(obs, "v15_review_status", ""),
                "v15_review_reason": getattr(obs, "v15_review_reason", ""),
                "v15_review_suggestions": compact_json(getattr(obs, "v15_review_suggestions", []) or []),
                "section": evidence.section if evidence else "",
                "raw_method_text": obs.raw_method_text[:1200],
                "review_status": "",
                "stage_review": "",
                "tools_review": "",
                "versions_review": "",
                "parameters_review": "",
                "inputs_outputs_review": "",
                "bbav5_review": "",
                "bbav5_stage_review": "",
                "bbav5_tools_review": "",
                "support_status_review": "",
                "corrected_stage_value": "",
                "corrected_tools_value": "",
                "corrected_versions_value": "",
                "corrected_parameters_value": "",
                "corrected_evidence_value": "",
                "corrected_inputs_value": "",
                "corrected_outputs_value": "",
                "corrected_bbav5_stage_value": "",
                "corrected_bbav5_tools_value": "",
                "corrected_support_status_value": "",
                "corrected_stage": "",
                "corrected_tools": "",
                "corrected_parameters": "",
                "corrected_inputs": "",
                "corrected_outputs": "",
                "corrected_bbav5_stage": "",
                "corrected_support_status": "",
                "evidence_review": "",
                "manual_note": "",
            }
        )
    return rows



def sanitize_review_text(value: Any) -> str:
    text = str(value or "")
    if "supplemental_evidence" not in text:
        return text
    text = __import__("re").sub(r'"?supplemental_evidence"?\s*:\s*\[.*?\]', '[supplemental evidence removed]', text, flags=__import__("re").S)
    return text.replace("supplemental_evidence", "supplemental evidence")


def sanitize_review_row(row: dict[str, Any]) -> dict[str, Any]:
    cleaned = dict(row)
    for key in ("parameters", "corrected_parameters_value", "corrected_parameters", "manual_note"):
        if key in cleaned:
            cleaned[key] = sanitize_review_text(cleaned.get(key, ""))
    return cleaned

def write_per_paper_checkpoint(
    output_dir: Path,
    paper_id: str,
    observations: list[Any],
    package: Any,
    rows: dict[str, list[dict[str, Any]]],
    review_report: dict[str, Any],
    duplicate_merge_rows: list[dict[str, Any]] | None = None,
) -> None:
    paper_dir = output_dir / "per_paper" / paper_id
    paper_dir.mkdir(parents=True, exist_ok=True)
    dump_json(paper_dir / "observations.json", [observation_to_output_dict(obs) for obs in observations])
    dump_json(paper_dir / "alignment.json", package.to_dict())
    dump_json(paper_dir / "methodology_review_report.json", review_report)
    write_csv(paper_dir / "review_template.csv", REVIEW_FIELDS, [sanitize_review_row(row) for row in rows.get("review", [])])
    write_csv(paper_dir / "methodology_tool_audit.csv", TOOL_AUDIT_FIELDS, rows.get("tool_audit", []))
    write_csv(paper_dir / "methodology_parameter_audit.csv", PARAMETER_AUDIT_FIELDS, rows.get("parameter_audit", []))
    write_csv(paper_dir / "methodology_database_url_audit.csv", URL_AUDIT_FIELDS, rows.get("url_audit", []))
    write_csv(paper_dir / "species_summary.csv", SPECIES_SUMMARY_FIELDS, rows.get("species", []))
    write_csv(paper_dir / "methodology_review_issues.csv", REVIEW_ISSUE_FIELDS, review_report.get("issues", []))
    dump_json(paper_dir / "duplicate_merge_report.json", {"merges": duplicate_merge_rows or [], "merge_count": len(duplicate_merge_rows or [])})


def discover_tool_candidates(paper_id: str, family: str, obs: Any, alignment: Any, evidence: Any) -> list[dict[str, Any]]:
    import re

    known_norm = {re.sub(r"[^a-z0-9]+", "", tool.lower()) for tool in KNOWN_TOOL_NAMES}
    observed_norm = {re.sub(r"[^a-z0-9]+", "", tool.tool_name.lower()) for tool in obs.tools}
    candidates: list[dict[str, Any]] = []
    text = obs.raw_method_text
    patterns = [
        r"\b[A-Z][A-Za-z0-9_.+-]{2,}(?:\s+[A-Z][A-Za-z0-9_.+-]{2,}){0,1}\b",
        r"\b[a-z]+[A-Z][A-Za-z0-9_.+-]{2,}\b",
    ]
    seen = set()
    for pattern in patterns:
        for match in re.finditer(pattern, text):
            name = match.group(0).strip(" ,.;:()[]")
            norm = re.sub(r"[^a-z0-9]+", "", name.lower())
            norm_no_version = re.sub(r"\d+(?:\d+)*$", "", norm)
            if (
                not norm
                or norm in seen
                or norm in known_norm
                or norm_no_version in known_norm
                or norm in observed_norm
                or norm_no_version in observed_norm
                or any(norm in known or known in norm for known in observed_norm if len(known) >= 5)
            ):
                continue
            if name in NON_TOOL_TERMS or name.lower() in {"the", "and", "for", "using", "additional", "figure", "table", "methods"}:
                continue
            if any(re.search(pattern, name) for pattern in KNOWN_NON_TOOL_PATTERNS):
                continue
            window_start = max(0, match.start() - 180)
            window_end = min(len(text), match.end() + 220)
            window = " ".join(text[window_start:window_end].split())
            if not re.search(r"\b(software|program|tool|package|database|server|website|platform|using|used|performed with|analyzed with|conducted by|calculated by|constructed with)\b", window, re.I):
                continue
            if not re.search(re.escape(name) + r".{0,80}\b(software|program|tool|package|database|server|website|platform)\b|\b(software|program|tool|package|database|server|website|platform)\b.{0,80}" + re.escape(name), window, re.I):
                continue
            if re.search(r"\b(gene|genes|protein|proteins|domain|domains|sequence|sequences|species|sample|samples|value|method|analysis)\b", name, re.I):
                continue
            if len(name) < 3 or len(name) > 60:
                continue
            seen.add(norm)
            version_match = re.search(r"(?:v|version)?\s*(\d+(?:\.\d+){0,3})", window, re.I)
            url_match = re.search(r"https?://[^\s,)]+", window)
            candidates.append(
                {
                    "tool_name": name,
                    "paper_id": paper_id,
                    "family": family,
                    "observation_id": obs.observation_id,
                    "stage_guess": obs.normalized_stage,
                    "role_guess": "",
                    "version": version_match.group(1) if version_match else "",
                    "url": url_match.group(0) if url_match else "",
                    "evidence_sentence": evidence.evidence_sentence if evidence else window[:500],
                    "review_status": "",
                    "accepted_stage": "",
                    "accepted_role": "",
                    "bbav5_stage": alignment.bbav5_stage_id if alignment else "",
                    "support_status": alignment.support_status if alignment else "",
                    "manual_note": "",
                }
            )
    return candidates


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--papers", default="data/evaluation/methodology_papers.csv")
    parser.add_argument("--catalog", default="data/bbav5_capability_catalog.json")
    parser.add_argument("--output-dir", default="data/evaluation/methodology_batch")
    parser.add_argument("--paper-id", default="", help="Optional single paper_id to process.")
    parser.add_argument("--max-papers", type=int, default=0, help="Optional cap for smoke tests.")
    parser.add_argument(
        "--reference",
        default="data/evaluation/bbav5_methodology_reference/bbav5_tools_reviewed.json",
        help="Reviewed bbav5 methodology reference JSON.",
    )
    parser.add_argument(
        "--llm-discovery-json",
        default="",
        help="Optional external LLM tool discovery JSON to merge into discovered_tools.csv.",
    )
    parser.add_argument(
        "--use-llm",
        action="store_true",
        help="Use LLM-based extraction instead of rule-based extraction. Requires LLM_API_KEY, DEEPSEEK_API_KEY, or DASHSCOPE_API_KEY.",
    )
    parser.add_argument(
        "--llm-model",
        default="deepseek-v4-pro",
        help="LLM model to use for extraction (default: deepseek-v4-pro)",
    )
    parser.add_argument(
        "--llm-provider",
        default="deepseek",
        choices=["deepseek", "kimi", "openai_compatible"],
        help="LLM provider profile family (default: deepseek).",
    )
    parser.add_argument(
        "--llm-profile",
        default="",
        choices=["", "deepseek", "kimi", "openai_compatible"],
        help="Optional provider profile override for prompt/output behavior.",
    )
    parser.add_argument(
        "--llm-base-url",
        default="",
        help="OpenAI-compatible LLM base URL. Defaults to LLM_BASE_URL/DEEPSEEK_BASE_URL or https://api.deepseek.com/v1.",
    )
    parser.add_argument(
        "--llm-api-key-env",
        default="",
        help="Environment variable name containing the LLM API key. Explicit env selection overrides provider defaults.",
    )
    parser.add_argument(
        "--llm-disable-repair",
        action="store_true",
        help="Disable JSON repair retry and use rule fallback on invalid model JSON.",
    )
    parser.add_argument(
        "--llm-request-timeout",
        type=int,
        default=75,
        help="Per-request LLM timeout in seconds. Logged, not used as a whole-batch limit.",
    )
    parser.add_argument(
        "--external-job-status-json",
        default="",
        help="Optional path to methodology upload job status.json to patch with substage progress.",
    )
    args = parser.parse_args()

    papers_path = resolve(args.papers)
    catalog = load_or_build_capability_catalog(resolve(args.catalog))
    aligner = MethodologyAligner(catalog)
    output_dir = resolve(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    log_path = output_dir / "methodology_batch_run.log"
    progress_path = output_dir / "progress.jsonl"
    llm_discovery_path = resolve(args.llm_discovery_json) if args.llm_discovery_json else None
    reference_path = resolve(args.reference) if args.reference else None
    reviewer = MethodologyReviewer.from_reference_path(reference_path)

    print_log(log_path, f"START batch papers={papers_path} output_dir={output_dir} use_llm={args.use_llm}")
    append_jsonl(progress_path, {"event": "batch_start", "papers": str(papers_path), "use_llm": args.use_llm})

    # Initialize LLM extractor if requested
    llm_extractor = None
    if args.use_llm:
        from src.rag.unified.llm_methodology_extractor import LLMMethodologyExtractor
        llm_extractor = LLMMethodologyExtractor(
            model=args.llm_model,
            base_url=args.llm_base_url or None,
            provider=args.llm_provider,
            provider_profile=args.llm_profile,
            api_key_env=args.llm_api_key_env,
            disable_repair=args.llm_disable_repair,
            request_timeout=args.llm_request_timeout,
        )
        print_log(
            log_path,
            "Using LLM extraction "
            f"provider={args.llm_provider} profile={args.llm_profile or args.llm_provider} "
            f"model={llm_extractor.model} base_url={llm_extractor.base_url} "
            f"request_timeout={args.llm_request_timeout}s repair_enabled={not args.llm_disable_repair}",
        )

    all_observations = []
    all_alignments = []
    review_rows: list[dict[str, Any]] = []
    summary_rows: list[dict[str, Any]] = []
    status_rows: list[dict[str, Any]] = []
    tool_audit: list[dict[str, Any]] = []
    parameter_audit: list[dict[str, Any]] = []
    url_audit: list[dict[str, Any]] = []
    species_summary: list[dict[str, Any]] = []
    review_issue_rows: list[dict[str, Any]] = []
    review_reports: list[dict[str, Any]] = []
    discovered_tool_rows: list[dict[str, Any]] = load_llm_discoveries(llm_discovery_path)
    duplicate_merge_rows: list[dict[str, Any]] = []
    llm_provider_reports: list[dict[str, Any]] = []

    ext_job_status = str(args.external_job_status_json or "").strip()

    paper_rows = read_rows(papers_path)
    if args.paper_id:
        paper_rows = [row for row in paper_rows if row["paper_id"].strip() == args.paper_id]
    if args.max_papers:
        paper_rows = paper_rows[: args.max_papers]

    for index, row in enumerate(paper_rows, start=1):
        paper_id = row["paper_id"].strip()
        family = row.get("family", "").strip()
        role = row.get("source_paper_role", "primary_gene_family_paper").strip() or "primary_gene_family_paper"
        parsed_path = resolve(row["parsed_document_path"].strip())
        paper_duplicate_merges: list[dict[str, Any]] = []
        paper_llm_provider_reports: list[dict[str, Any]] = []
        print_log(log_path, f"PAPER_START {index}/{len(paper_rows)} paper_id={paper_id} path={parsed_path}")
        append_jsonl(progress_path, {"event": "paper_start", "paper_id": paper_id, "index": index, "total": len(paper_rows)})

        try:
            document = ParsedDocumentLoader.from_path(parsed_path, document_id=paper_id)
            patch_external_job_status(ext_job_status, "load_document", "已载入解析文档，准备抽取方法学…")

            # Use LLM extraction if requested, otherwise use rule-based
            if llm_extractor:
                print_log(log_path, f"EXTRACT_START paper_id={paper_id} mode=llm")
                patch_external_job_status(ext_job_status, "extract", "正在进行方法学抽取（LLM）…")
                observations = llm_extractor.extract(document, paper_id, role)
                paper_llm_provider_reports = list(getattr(llm_extractor, "provider_report", []))
                llm_provider_reports.extend(paper_llm_provider_reports)
            else:
                print_log(log_path, f"EXTRACT_START paper_id={paper_id} mode=rule")
                patch_external_job_status(ext_job_status, "extract", "正在进行方法学抽取（规则）…")
                observations = ParsedDocumentMethodologyExtractor(paper_id, role).extract(document)

            patch_external_job_status(ext_job_status, "normalize", "抽取完成，正在规范化 observation…")
            observations = normalize_observations(observations, family)
            patch_external_job_status(ext_job_status, "align", "正在进行 bbav5 目录对齐与审计表生成…")
            package = aligner.build_package(paper_id, observations)
            error = ""
            print_log(log_path, f"EXTRACT_DONE paper_id={paper_id} observations={len(observations)}")
        except Exception as exc:
            observations = []
            package = aligner.build_package(paper_id, [])
            error = f"{type(exc).__name__}: {exc}"
            document = None
            print_log(log_path, f"EXTRACT_ERROR paper_id={paper_id} error={error}")

        methods_chunks = []
        if document is not None:
            methods_chunks = [chunk for chunk in document.chunks if is_methods_section(chunk.section)]
        methods_text_length = sum(len(chunk.text or "") for chunk in methods_chunks)
        if methods_text_length < 500 and observations:
            observed_texts = {
                (obs.raw_method_text or "").strip()
                for obs in observations
                if getattr(obs, "raw_method_text", "")
            }
            methods_text_length = max(methods_text_length, sum(len(text) for text in observed_texts))
        ready_for_review = not error and methods_text_length >= 500 and len(observations) >= 3

        alignments_by_id = {alignment.observation_id: alignment for alignment in package.alignments}
        paper_tool_audit = tool_audit_rows(paper_id, family, observations, alignments_by_id)
        paper_parameter_audit = parameter_audit_rows(paper_id, family, observations)
        paper_url_audit = url_audit_rows(paper_id, family, observations)
        paper_species_summary = infer_species_rows(paper_id, family, observations, document)
        paper_review_rows = build_review_rows(paper_id, family, observations, alignments_by_id, catalog)
        patch_external_job_status(ext_job_status, "review", "正在执行自动审核（规则 / 交叉验证）…")
        review_report = reviewer.review(paper_id, family, observations, package.alignments, document)

        tool_audit.extend(paper_tool_audit)
        parameter_audit.extend(paper_parameter_audit)
        url_audit.extend(paper_url_audit)
        species_summary.extend(paper_species_summary)
        review_rows.extend(paper_review_rows)
        review_issue_rows.extend(review_report.get("issues", []))
        review_reports.append(review_report)
        all_observations.extend(observation_to_output_dict(observation) for observation in observations)
        all_alignments.extend(asdict(alignment) for alignment in package.alignments)

        stages = sorted({obs.normalized_stage for obs in observations})
        tools = sorted({tool.tool_name for obs in observations for tool in obs.tools})
        support_counts: dict[str, int] = {}
        for alignment in package.alignments:
            support_counts[alignment.support_status] = support_counts.get(alignment.support_status, 0) + 1

        summary_rows.append(
            {
                "paper_id": paper_id,
                "family": family,
                "source_paper_role": role,
                "parsed_document_path": row.get("parsed_document_path", ""),
                "observation_count": len(observations),
                "stage_count": len(stages),
                "stages_detected": ";".join(stages),
                "tools_detected": ";".join(tools),
                "support_counts": compact_json(support_counts),
                "error": error,
            }
        )
        status_rows.append(
            {
                "paper_id": paper_id,
                "family": family,
                "source_paper_role": role,
                "parsed_document_path": row.get("parsed_document_path", ""),
                "parsed_document_exists": "yes" if parsed_path.exists() else "no",
                "parser": document.parser if document else "",
                "chunk_count": len(document.chunks) if document else 0,
                "methods_chunk_count": len(methods_chunks),
                "methods_text_length": methods_text_length,
                "observation_count": len(observations),
                "ready_for_review": "yes" if ready_for_review else "no",
                "failure_reason": error
                or (
                    "methods_text_too_short"
                    if methods_text_length < 500
                    else ("too_few_observations" if len(observations) < 3 else "")
                ),
            }
        )

        for obs in observations:
            alignment = alignments_by_id.get(obs.observation_id)
            evidence = obs.evidence[0] if obs.evidence else None
            if ready_for_review:
                discovered_tool_rows.extend(discover_tool_candidates(paper_id, family, obs, alignment, evidence))

        patch_external_job_status(ext_job_status, "persist", "审核完成，正在写入本轮检查点…")
        write_per_paper_checkpoint(
            output_dir,
            paper_id,
            observations,
            package,
            {
                "review": paper_review_rows,
                "tool_audit": paper_tool_audit,
                "parameter_audit": paper_parameter_audit,
                "url_audit": paper_url_audit,
                "species": paper_species_summary,
            },
            review_report,
            paper_duplicate_merges,
        )
        if paper_llm_provider_reports:
            dump_json(
                output_dir / "per_paper" / paper_id / "llm_provider_report.json",
                {"windows": paper_llm_provider_reports, "window_count": len(paper_llm_provider_reports)},
            )
        append_jsonl(
            progress_path,
            {
                "event": "paper_done" if not error else "paper_failed",
                "paper_id": paper_id,
                "observation_count": len(observations),
                "tool_count": len(paper_tool_audit),
                "parameter_count": len(paper_parameter_audit),
                "species_count": len(paper_species_summary),
                "review_status": review_report.get("review_status"),
                "review_score": review_report.get("score"),
                "error": error,
            },
        )
        print_log(
            log_path,
            f"PAPER_DONE paper_id={paper_id} observations={len(observations)} "
            f"tools={len(paper_tool_audit)} species={len(paper_species_summary)} "
            f"review={review_report.get('review_status')} score={review_report.get('score')}",
        )

    final_stage_names = sorted({
        stage
        for row in summary_rows
        for stage in str(row.get("stages_detected") or "").split(";")
        if stage
    })
    final_stage_count = len(final_stage_names)
    patch_external_job_status(
        ext_job_status,
        "done",
        "方法学批量提取完成",
        extra={
            "stage_count": final_stage_count,
            "species_count": len(species_summary),
            "methodology_observation_count": len(all_observations),
            "methodology_tool_count": len(tool_audit),
            "methodology_stage_names": final_stage_names,
        },
    )
    dump_json(output_dir / "observations.json", all_observations)
    dump_json(output_dir / "alignment.json", {"observations": all_observations, "alignments": all_alignments})
    dump_json(output_dir / "methodology_review_report.json", {"reports": review_reports})
    dump_json(output_dir / "duplicate_merge_report.json", {"merges": duplicate_merge_rows, "merge_count": len(duplicate_merge_rows)})
    dump_json(output_dir / "llm_provider_report.json", {"windows": llm_provider_reports, "window_count": len(llm_provider_reports)})
    write_csv(output_dir / "summary.csv", list(summary_rows[0].keys()) if summary_rows else [], summary_rows)
    write_csv(output_dir / "processing_status.csv", STATUS_FIELDS, status_rows)
    write_csv(output_dir / "review_template.csv", REVIEW_FIELDS, [sanitize_review_row(row) for row in review_rows])
    write_csv(output_dir / "discovered_tools.csv", DISCOVERED_TOOL_FIELDS, discovered_tool_rows)
    write_csv(output_dir / "methodology_tool_audit.csv", TOOL_AUDIT_FIELDS, tool_audit)
    write_csv(output_dir / "methodology_parameter_audit.csv", PARAMETER_AUDIT_FIELDS, parameter_audit)
    write_csv(output_dir / "methodology_database_url_audit.csv", URL_AUDIT_FIELDS, url_audit)
    write_csv(output_dir / "species_summary.csv", SPECIES_SUMMARY_FIELDS, species_summary)
    write_csv(output_dir / "methodology_review_issues.csv", REVIEW_ISSUE_FIELDS, review_issue_rows)
    try:
        dump_json(output_dir / "methodology_plan.json", build_methodology_execution_plan(output_dir))
    except Exception as exc:
        dump_json(output_dir / "methodology_plan_error.json", {"error": str(exc)})
    write_tool_lexicon(PROJECT_ROOT / "data/evaluation/methodology_tool_lexicon.json", discovered_tool_rows + tool_audit)

    append_jsonl(progress_path, {"event": "batch_done", "papers": len(summary_rows), "observations": len(review_rows)})
    print_log(log_path, f"Wrote {output_dir / 'observations.json'}")
    print_log(log_path, f"Wrote {output_dir / 'alignment.json'}")
    print_log(log_path, f"Wrote {output_dir / 'summary.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'processing_status.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'review_template.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'discovered_tools.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'methodology_tool_audit.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'methodology_parameter_audit.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'methodology_database_url_audit.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'species_summary.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'methodology_review_issues.csv'}")
    print_log(log_path, f"Wrote {output_dir / 'methodology_review_report.json'}")
    print_log(log_path, f"Wrote {output_dir / 'methodology_plan.json'}")
    print_log(log_path, f"Wrote {PROJECT_ROOT / 'data/evaluation/methodology_tool_lexicon.json'}")
    print_log(log_path, f"DONE papers={len(summary_rows)} observations={len(review_rows)}")


if __name__ == "__main__":
    main()