#!/usr/bin/env python3 """Run and validate the maintained V16 extraction baseline on a real PDF.""" from __future__ import annotations import argparse import csv import importlib.util import json import os import sys from collections import Counter from pathlib import Path from typing import Any, Dict, Iterable, List PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) PC_REQUIRED_FILES = ( "scientific_chunks.json", "gene_alias.json", "all_dic.json", "relationships_by_category_final.json", ) TABLE_READER_MODULES = { "pandas": "pandas", "xlsx": "openpyxl", "xls": "xlrd", "docx": "docx", } CPU_MINERU_ENV = { "CUDA_VISIBLE_DEVICES": "", "MINERU_MODEL_SOURCE": "modelscope", "TORCH_CUDNN_V8_API_DISABLED": "1", "V15_MINERU_BACKEND": "pipeline", "V15_MINERU_METHOD": "txt", "V15_MINERU_LANG": "en", "V15_MINERU_USE_API": "0", "V15_MINERU_FORMULA": "0", "V15_MINERU_TABLE": "0", } def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run V16Extractor on a real PDF and validate baseline invariants." ) parser.add_argument("--pdf", required=True, type=Path, help="Input PDF path.") parser.add_argument("--pmid", default=None, help="PMID passed to the extractor.") parser.add_argument("--doi", default=None, help="Optional DOI override.") parser.add_argument("--output-dir", required=True, type=Path, help="Extraction output directory.") parser.add_argument( "--cpu-mineru", action="store_true", help="Force low-memory MinerU pipeline settings and disable CUDA visibility.", ) parser.add_argument("--no-llm", action="store_true", help="Keep LLM relation extraction off.") parser.add_argument("--no-vlm", action="store_true", help="Keep VLM image extraction off.") parser.add_argument("--with-llm", action="store_true", help="Enable LLM relation extraction when keys are configured.") parser.add_argument( "--with-llm-definitions", action="store_true", help="Also generate per-entity LLM definitions. Off by default for regression stability.", ) parser.add_argument("--with-vlm", action="store_true", help="Enable VLM image extraction.") parser.add_argument("--no-pc-data", action="store_true", help="Disable PC data layer.") parser.add_argument( "--extractor-version", choices=("v16_1_2", "v16_1_1"), default="v16_1_2", help="Extractor package to validate. Defaults to the latest maintained version.", ) parser.add_argument( "--pc-data-root", type=Path, default=None, help="Override PC_DATA_ROOT for this validation run without editing .env.", ) parser.add_argument( "--split-data-root", type=Path, default=None, help="Record/check the user-managed split-data root for this run. No data processing is performed.", ) parser.add_argument( "--require-pc-data", action="store_true", help="Fail if PC required files are missing or PC loader counts are zero.", ) return parser.parse_args(list(argv) if argv is not None else None) def load_env_file(env_path: Path = PROJECT_ROOT / ".env") -> None: if not env_path.exists(): return for raw_line in env_path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value def apply_runtime_environment(cpu_mineru: bool = False) -> Dict[str, str]: applied: Dict[str, str] = {} if cpu_mineru: for key, value in CPU_MINERU_ENV.items(): os.environ[key] = value applied[key] = value mineru_cmd = os.environ.get("MINERU_CMD", "").strip() if mineru_cmd and not os.environ.get("V15_MINERU_CMD"): os.environ["V15_MINERU_CMD"] = mineru_cmd applied["V15_MINERU_CMD"] = mineru_cmd return applied def _resolve_path(value: str | Path | None, default: str | Path | None = None) -> Path | None: raw_value = str(value).strip() if value is not None else "" raw_default = str(default).strip() if default is not None else "" chosen = raw_value or raw_default if not chosen: return None path = Path(chosen) if not path.is_absolute(): path = PROJECT_ROOT / path return path.resolve() def apply_external_data_overrides(args: argparse.Namespace) -> Dict[str, str]: applied: Dict[str, str] = {} if args.pc_data_root is not None: pc_root = _resolve_path(args.pc_data_root) if pc_root is not None: os.environ["PC_DATA_ROOT"] = str(pc_root) applied["PC_DATA_ROOT"] = str(pc_root) if args.split_data_root is not None: split_root = _resolve_path(args.split_data_root) if split_root is not None: os.environ["PLANTAI_SPLIT_DATA_ROOT"] = str(split_root) applied["PLANTAI_SPLIT_DATA_ROOT"] = str(split_root) return applied def inspect_data_root(root: Path | None, required_files: Iterable[str] = ()) -> Dict[str, Any]: required = list(required_files) if root is None: return { "configured": False, "root": "", "exists": False, "is_dir": False, "readable": False, "required_files": {name: False for name in required}, "missing_required_files": required, "available": False, } file_status = {name: (root / name).is_file() for name in required} missing = [name for name, exists in file_status.items() if not exists] exists = root.exists() is_dir = root.is_dir() readable = os.access(root, os.R_OK) if exists else False return { "configured": True, "root": str(root), "exists": exists, "is_dir": is_dir, "readable": readable, "required_files": file_status, "missing_required_files": missing, "available": exists and is_dir and readable and not missing, } def inspect_external_data(args: argparse.Namespace) -> Dict[str, Any]: pc_root = _resolve_path(args.pc_data_root, os.environ.get("PC_DATA_ROOT") or "data/pc") split_root = _resolve_path(args.split_data_root, os.environ.get("PLANTAI_SPLIT_DATA_ROOT")) return { "pc": inspect_data_root(pc_root, PC_REQUIRED_FILES), "split": inspect_data_root(split_root), } def inspect_table_reader_dependencies() -> Dict[str, Any]: modules = { label: importlib.util.find_spec(module_name) is not None for label, module_name in TABLE_READER_MODULES.items() } return { "modules": modules, "csv_available": modules["pandas"], "tsv_available": modules["pandas"], "xlsx_available": modules["pandas"] and modules["xlsx"], "xls_available": modules["pandas"] and modules["xls"], "docx_available": modules["pandas"] and modules["docx"], "missing_modules": [label for label, available in modules.items() if not available], } def preflight_configuration(args: argparse.Namespace, external_data: Dict[str, Any]) -> List[str]: failures: List[str] = [] if args.with_llm and not args.no_llm and not os.environ.get("DEEPSEEK_API_KEY", "").strip(): failures.append("DEEPSEEK_API_KEY is required when --with-llm is used") if args.with_vlm and not args.no_vlm and not os.environ.get("DASHSCOPE_API_KEY", "").strip(): failures.append("DASHSCOPE_API_KEY is required when --with-vlm is used") if args.require_pc_data: pc_info = external_data.get("pc") or {} if not pc_info.get("available"): failures.append(f"PC data is required but unavailable: {pc_info.get('root', '')}") for file_name in pc_info.get("missing_required_files") or []: failures.append(f"PC required file missing: {file_name}") return failures def _count_items(result: Dict[str, Any], key: str) -> int: value = result.get(key) or [] return len(value) if isinstance(value, list) else 0 def _resolve_audit_path(path_value: str | Path | None, output_dir: Path) -> Path | None: if not path_value: return None path = Path(str(path_value)) if not path.is_absolute(): path = output_dir / path return path def _count_jsonl_rows(path: Path) -> int: if not path.is_file(): return 0 with path.open("r", encoding="utf-8") as handle: return sum(1 for line in handle if line.strip()) def _count_csv_data_rows(path: Path) -> int: if not path.is_file(): return 0 with path.open("r", encoding="utf-8", newline="") as handle: return sum(1 for _row in csv.DictReader(handle)) def summarize_run_request(args: argparse.Namespace) -> Dict[str, Any]: pdf_path = args.pdf.resolve() output_dir = args.output_dir.resolve() use_llm = bool(args.with_llm and not args.no_llm) use_vlm = bool(args.with_vlm and not args.no_vlm) return { "scope": "single_real_pdf", "pdf_path": str(pdf_path), "pdf_exists": pdf_path.is_file(), "pdf_size_bytes": pdf_path.stat().st_size if pdf_path.is_file() else 0, "pmid": args.pmid or "", "doi": args.doi or "", "output_dir": str(output_dir), "extractor_version": args.extractor_version, "cpu_mineru": bool(args.cpu_mineru), "use_llm_relations": use_llm, "use_llm_definitions": bool(args.with_llm_definitions and use_llm), "use_vlm": use_vlm, "use_pc_data": not args.no_pc_data, "require_pc_data": bool(args.require_pc_data), } def _read_llm_raw_response_rows(path: Path) -> List[Dict[str, Any]]: if not path.is_file(): return [] rows: List[Dict[str, Any]] = [] with path.open("r", encoding="utf-8") as handle: for line in handle: raw = line.strip() if not raw: continue try: row = json.loads(raw) except json.JSONDecodeError: rows.append( { "window_id": "", "section": "", "status": "parse_error", "error": "raw response audit row is not valid JSON", } ) continue if isinstance(row, dict): rows.append(row) return rows def collect_llm_runtime_diagnostics( output_dir: Path, llm_metadata: Dict[str, Any] | None = None, ) -> Dict[str, Any]: """Summarize LLM runtime stability without copying prompt/response text.""" llm_metadata = llm_metadata or {} audit_files = llm_metadata.get("audit_files") or {} raw_candidates = [ _resolve_audit_path(audit_files.get("llm1_raw_response"), output_dir), _resolve_audit_path(audit_files.get("llm_relation_raw_responses"), output_dir), output_dir / "llm1_raw_response.jsonl", output_dir / "llm_relation_raw_responses.jsonl", ] raw_response_path = next((path for path in raw_candidates if path and path.is_file()), None) raw_rows = _read_llm_raw_response_rows(raw_response_path) if raw_response_path else [] status_counts = Counter(str(row.get("status") or "unknown") for row in raw_rows) error_rows = [row for row in raw_rows if str(row.get("status") or "").lower() != "ok"] timeout_rows = [ row for row in error_rows if "timed out" in str(row.get("error") or "").lower() or "timeout" in str(row.get("error") or "").lower() ] error_windows = [ { "window_id": str(row.get("window_id") or ""), "section": str(row.get("section") or ""), "error": str(row.get("error") or "")[:240], } for row in error_rows[:50] ] candidate_files = sorted(output_dir.glob("*.llm1_candidate_relations.jsonl")) rejected_files = sorted(output_dir.glob("*.llm2_rejected_relations.csv")) return { "enabled": bool(llm_metadata.get("enabled", False)), "client_available": bool(llm_metadata.get("client_available", False)), "metadata_llm_requests": int(llm_metadata.get("llm_requests") or 0), "metadata_llm_failures": int(llm_metadata.get("llm_failures") or 0), "raw_response_file": str(raw_response_path) if raw_response_path else "", "raw_response_rows": len(raw_rows), "raw_response_status_counts": dict(status_counts), "raw_response_error_count": len(error_rows), "timeout_error_count": len(timeout_rows), "error_windows": error_windows, "error_windows_truncated": len(error_rows) > len(error_windows), "candidate_relation_files": [str(path) for path in candidate_files], "candidate_relation_rows": sum(_count_jsonl_rows(path) for path in candidate_files), "rejected_candidate_files": [str(path) for path in rejected_files], "rejected_candidate_rows": sum(_count_csv_data_rows(path) for path in rejected_files), } def summarize_result( result: Dict[str, Any], output_dir: Path, external_data: Dict[str, Any] | None = None, table_readers: Dict[str, Any] | None = None, ) -> Dict[str, Any]: relations = result.get("relations") or [] entities = result.get("entities") or [] sections = result.get("sections") or [] chunks = result.get("chunks") or [] metadata = result.get("metadata") or {} missing_source = sum(1 for rel in relations if not str(rel.get("source", "")).strip()) missing_target = sum(1 for rel in relations if not str(rel.get("target", "")).strip()) missing_evidence = sum(1 for rel in relations if not str(rel.get("evidence", "")).strip()) relation_types = Counter( str(rel.get("relationship_resolved") or rel.get("relation") or rel.get("relation_type") or "unknown") for rel in relations ) output_layers = Counter(str(rel.get("output_layer") or "") for rel in relations) external_data = external_data or {} table_readers = table_readers or inspect_table_reader_dependencies() pc_data = external_data.get("pc") or {} split_data = external_data.get("split") or {} llm_relation_extraction = metadata.get("llm_relation_extraction", {}) llm_runtime_diagnostics = collect_llm_runtime_diagnostics( output_dir, llm_relation_extraction, ) return { "output_dir": str(output_dir), "result_json": str(output_dir / "result.json"), "entities": len(entities), "relations": len(relations), "sections": len(sections), "chunks": len(chunks), "missing_source": missing_source, "missing_target": missing_target, "missing_evidence": missing_evidence, "relation_types_top": relation_types.most_common(20), "output_layers": dict(output_layers), "pc_data_stats": metadata.get("pc_data_stats", {}), "relation_block_extraction": metadata.get("relation_block_extraction", {}), "llm_relation_extraction": llm_relation_extraction, "llm_runtime_diagnostics": llm_runtime_diagnostics, "pc_data_root": pc_data.get("root", ""), "pc_data_available": bool(pc_data.get("available", False)), "split_data_root": split_data.get("root", ""), "split_data_available": bool(split_data.get("available", False)), "external_data": external_data, "supplementary_table_readers": table_readers, "version": metadata.get("version"), "extractor_version": metadata.get("extractor_version"), } def validate_summary( summary: Dict[str, Any], require_pc_data: bool = False, require_llm_relations: bool = False, require_run_request: bool = False, ) -> List[str]: failures: List[str] = [] for key in ("sections", "entities", "relations", "chunks"): if int(summary.get(key) or 0) <= 0: failures.append(f"{key} must be > 0") for key in ("missing_source", "missing_target", "missing_evidence"): if int(summary.get(key) or 0) != 0: failures.append(f"{key} must be 0") result_json = Path(str(summary.get("result_json") or "")) if not result_json.exists(): failures.append(f"result.json not found: {result_json}") if require_run_request: request = summary.get("run_request") or {} if not isinstance(request, dict) or not request: failures.append("run_request must be recorded for real validation runs") else: if request.get("scope") != "single_real_pdf": failures.append("run_request.scope must be single_real_pdf") if not str(request.get("pdf_path") or "").strip(): failures.append("run_request.pdf_path must be recorded") if not request.get("pdf_exists"): failures.append("run_request.pdf_exists must be true") if int(request.get("pdf_size_bytes") or 0) <= 0: failures.append("run_request.pdf_size_bytes must be > 0") if request.get("extractor_version") not in {"v16_1_1", "v16_1_2"}: failures.append("run_request.extractor_version must be a maintained V16 version") if not str(request.get("output_dir") or "").strip(): failures.append("run_request.output_dir must be recorded") if require_pc_data: pc_info = (summary.get("external_data") or {}).get("pc") or {} missing_files = pc_info.get("missing_required_files") or [] if not pc_info.get("available"): failures.append(f"PC data is required but unavailable: {pc_info.get('root', '')}") for file_name in missing_files: failures.append(f"PC required file missing: {file_name}") pc_stats = summary.get("pc_data_stats") or {} for key in ( "alias_map_loaded", "gene_aliases_loaded", "relationship_patterns_loaded", "entity_dict_loaded", ): if int(pc_stats.get(key) or 0) <= 0: failures.append(f"pc_data_stats.{key} must be > 0 when --require-pc-data is used") if require_llm_relations: block = summary.get("relation_block_extraction") or {} if not block.get("enabled"): failures.append("LLM relation block extraction must be enabled when --with-llm is used") if int(block.get("chunk_count") or 0) <= 0: failures.append("relation_block_extraction.chunk_count must be > 0 when --with-llm is used") if int(block.get("relation_count_before_gate") or 0) <= 0: failures.append( "relation_block_extraction.relation_count_before_gate must be > 0 when --with-llm is used" ) return failures def run_extraction(args: argparse.Namespace) -> Dict[str, Any]: if args.extractor_version == "v16_1_1": from src.rag.extractors.v16_1_1 import V16Extractor else: from src.rag.extractors.v16_1_2 import V16Extractor output_dir = args.output_dir.resolve() output_dir.mkdir(parents=True, exist_ok=True) use_llm = bool(args.with_llm and not args.no_llm) use_llm_definitions = bool(args.with_llm_definitions and use_llm) use_vlm = bool(args.with_vlm and not args.no_vlm) extractor = V16Extractor( use_vlm=use_vlm, use_mineru=True, output_dir=output_dir, use_pc_data=not args.no_pc_data, use_disambiguation=True, use_llm_definitions=use_llm_definitions, use_llm_relations=use_llm, use_rule_relation_fallback=args.extractor_version == "v16_1_1", use_neo4j=False, use_unified_storage=False, ) return extractor.extract(args.pdf.resolve(), pmid=args.pmid, doi=args.doi) def main(argv: Iterable[str] | None = None) -> int: args = parse_args(argv) load_env_file() external_overrides = apply_external_data_overrides(args) applied_env = apply_runtime_environment(cpu_mineru=args.cpu_mineru) applied_env.update(external_overrides) if not args.pdf.exists(): print(f"ERROR: input PDF does not exist: {args.pdf}", file=sys.stderr) return 2 if args.no_pc_data and args.require_pc_data: print("ERROR: --no-pc-data conflicts with --require-pc-data", file=sys.stderr) return 2 external_data = inspect_external_data(args) table_readers = inspect_table_reader_dependencies() preflight_failures = preflight_configuration(args, external_data) if preflight_failures: print("VALIDATION_PREFLIGHT_FAILED") for failure in preflight_failures: print(f"- {failure}") return 2 result = run_extraction(args) summary = summarize_result( result, args.output_dir.resolve(), external_data=external_data, table_readers=table_readers, ) summary["run_request"] = summarize_run_request(args) summary["applied_env"] = { key: ("" if value == "" else value) for key, value in applied_env.items() if "KEY" not in key and "PASSWORD" not in key and "TOKEN" not in key } summary_path = args.output_dir.resolve() / "validation_summary.json" summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") require_llm_relations = bool( args.with_llm and not args.no_llm and args.extractor_version == "v16_1_2" ) failures = validate_summary( summary, require_pc_data=args.require_pc_data, require_llm_relations=require_llm_relations, require_run_request=True, ) print(json.dumps(summary, ensure_ascii=False, indent=2)) print(f"VALIDATION_SUMMARY_JSON {summary_path}") if failures: print("VALIDATION_FAILED") for failure in failures: print(f"- {failure}") return 2 print("VALIDATION_PASSED") return 0 if __name__ == "__main__": raise SystemExit(main())