#!/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"(? 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"(? 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()