From 65f79207f050a89d8a3ae3adedb270927bdaa358 Mon Sep 17 00:00:00 2001 From: wtx Date: Sun, 9 Aug 2026 01:37:00 +0800 Subject: [PATCH] feat: add evidence-bound RNA-seq result reports --- apps/web/vue3/main.py | 138 +- gfa_core/mas/utils/intent_parser.py | 49 + gfa_core/rnaseq_report.py | 1304 ++++++++++++++++ tests/unit/test_rnaseq_report.py | 2131 +++++++++++++++++++++++++++ 4 files changed, 3620 insertions(+), 2 deletions(-) create mode 100644 gfa_core/rnaseq_report.py create mode 100644 tests/unit/test_rnaseq_report.py diff --git a/apps/web/vue3/main.py b/apps/web/vue3/main.py index b727ff36f..79490f6d9 100644 --- a/apps/web/vue3/main.py +++ b/apps/web/vue3/main.py @@ -6911,6 +6911,10 @@ class ExecuteRequest(BaseModel): speciesCode: str = "" rnaseqPrefix: str = "" rnaseqOutputPrefix: str = "" + rnaseqScopeId: str = "" + rnaseqExpectedStages: List[str] = Field(default_factory=list) + rnaseqReportAction: str = "generate" + rnaseqReportUseModel: bool = True # RNA-seq QC 参数 rnaseqDataMode: str = "upload" rnaseqDataPath: str = "" @@ -7794,6 +7798,10 @@ def run_stage_sync(task_id: str, stage: str, job_id: str, params: dict): wgcna_powers=params.get("wgcnaPowers", "1,2,3,4,5,6,7,8,9,10,12,14,16,18,20"), candidate_max_deg_proteins=params.get("candidateMaxDegProteins", 500), candidate_run_interproscan=params.get("candidateRunInterproscan", "auto"), + rnaseq_scope_id=params.get("rnaseqScopeId", ""), + rnaseq_expected_stages=params.get("rnaseqExpectedStages") or params.get("expectedStages") or [], + rnaseq_report_action=params.get("rnaseqReportAction", "generate"), + rnaseq_report_use_model=_bool_param("rnaseqReportUseModel", "report_use_model", default=True), # PPI参数 ppi_species=params.get("ppiSpecies") or params.get("ppi_species") or "auto", normalization_species_label=params.get("normalizationSpeciesLabel") or params.get("normalization_species_label") or "", @@ -12093,6 +12101,10 @@ async def execute_stage(req: ExecuteRequest): "wgcnaPowers": getattr(req, 'wgcnaPowers', "1,2,3,4,5,6,7,8,9,10,12,14,16,18,20"), "candidateMaxDegProteins": getattr(req, 'candidateMaxDegProteins', 500), "candidateRunInterproscan": getattr(req, 'candidateRunInterproscan', "auto"), + "rnaseqScopeId": getattr(req, 'rnaseqScopeId', ""), + "rnaseqExpectedStages": getattr(req, 'rnaseqExpectedStages', []), + "rnaseqReportAction": getattr(req, 'rnaseqReportAction', "generate"), + "rnaseqReportUseModel": getattr(req, 'rnaseqReportUseModel', True), "ncbiQuery": getattr(req, 'ncbiQuery', ''), "ncbiGeneIds": getattr(req, 'ncbiGeneIds', ''), "ncbiMode": getattr(req, 'ncbiMode', 'gene'), @@ -13679,6 +13691,7 @@ async def get_results( "rnaseq_fgsea": {"images": [], "files": [], "elapsed_time": 0}, "rnaseq_wgcna": {"images": [], "files": [], "elapsed_time": 0}, "rnaseq_candidate_family": {"images": [], "files": [], "elapsed_time": 0}, + "rnaseq_report": {"images": [], "files": [], "elapsed_time": 0}, "genome_prepare": {"images": [], "files": [], "elapsed_time": 0}, "prepare_organize": {"images": [], "files": [], "elapsed_time": 0}, "prepare_normalize_gff": {"images": [], "files": [], "elapsed_time": 0}, @@ -13780,6 +13793,7 @@ async def get_results( "rnaseq_fgsea", "rnaseq_wgcna", "rnaseq_candidate_family", + "rnaseq_report", "expression", } RNASEQ_FILE_TOKENS = ( @@ -14801,6 +14815,7 @@ async def get_results( "genome_prepare": ["genome_prepare", "species_config"], "msa_region_plot": ["msa_region", "region.tsv", "conservation_profile"], "subfamily_confirm": ["subfamily_confirm", "confirmed_subfamily"], + "rnaseq_report": ["rnaseq_report"], "rnaseq_fgsea": ["rnaseq_fgsea", "fgsea"], "rnaseq_go_enrich": ["rnaseq_go_enrich", ".go_enrich", "_go_"], "rnaseq_kegg_enrich": ["rnaseq_kegg_enrich", ".kegg_enrich", "_kegg_", "kegg_pathway", "kegg_pathview"], @@ -16029,6 +16044,7 @@ def parse_natural_language(message: str) -> dict: "rnaseq_align": ["rnaseq比对", "rna比对", "hisat2", "比对"], "rnaseq_quant": ["rnaseq定量", "rna定量", "stringtie", "定量"], "rnaseq_sample_qc": ["样本质量", "样本检测", "sample qc", "sample quality", "pca", "样本相关性"], + "rnaseq_report": ["rnaseq结果报告", "rna-seq结果报告", "转录组结果报告", "rnaseq报告"], "rnaseq_wgcna": ["wgcna", "共表达", "加权共表达", "模块性状", "module trait"], "rnaseq_de": ["rnaseq差异", "rna差异", "差异表达", "de分析"], "rnaseq_fgsea": ["fgsea", "gsea", "基因集富集", "排序富集"], @@ -16838,7 +16854,10 @@ async def _run_agent_chat( ) router_context_terms = dict(terms) - from gfa_core.mas.utils.intent_parser import parse_sequence_extract_request + from gfa_core.mas.utils.intent_parser import ( + deterministic_agent_chat_intent, + parse_sequence_extract_request, + ) utility_params = parse_sequence_extract_request(request.message) if utility_params: @@ -16884,6 +16903,58 @@ async def _run_agent_chat( "session_id": session_id, } + deterministic_intent = deterministic_agent_chat_intent( + request.message, + rnaseq_context=any( + str(stage).startswith("rnaseq_") for stage in completed_stages + ), + ) + if deterministic_intent is not None: + report_action = str( + (deterministic_intent.get("params") or {}).get("rnaseqReportAction") + or "generate" + ) + report_message = ( + "已识别为打开 RNA-seq 结果报告。" + if report_action == "open" + else "已识别为生成 RNA-seq 结果报告。" + ) + report_intent = { + **deterministic_intent, + "stage": "rnaseq_report", + "primary_stage": "rnaseq_report", + "taskType": "analysis", + "task_kind": "analysis", + "job_id": request.job_id, + "workspace": ( + str(PROJECT_ROOT / "workspace" / request.job_id) + if request.job_id + else "" + ), + "message": report_message, + } + if conversation_turn is not None: + active_store.finish_turn( + conversation_turn, + assistant_message=report_message, + interaction_kind="independent", + action="propose_analysis", + intent=report_intent, + ) + return { + "success": True, + "status": "ok", + "action": "propose_analysis", + "intent": report_intent, + "sources": [], + "tool_trace": [], + "mode": "analysis", + "completed_stages": completed_stages, + "model_used": "rule", + "session_id": session_id, + "handoff_request": request.message, + } + global llm_manager active_llm_manager = llm_manager if active_llm_manager is None: @@ -20012,6 +20083,63 @@ async def agent_plan(request: AgentPlanRequest): request.intent["params"] = dict(request.intent.get("params") or {}) _apply_reconciled_species_to_intent(request.intent, plan_role_resolution) query_lower_for_context = (request.message or "").lower() + result_report_query = ( + "结果报告" in query_lower_for_context + and any(action in query_lower_for_context for action in ["生成", "打开"]) + ) + if result_report_query: + from gfa_core.rnaseq_evidence import is_rnaseq_evidence_stage, manifest_identity_value + + current_intent = dict(request.intent or {}) + current_params = dict(current_intent.get("params") or {}) + current_stages = [str(stage) for stage in (current_intent.get("stages") or [])] + explicit_rnaseq_report = any( + marker in query_lower_for_context + for marker in ["rnaseq", "rna-seq", "rna seq", "转录组"] + ) + rnaseq_context = any(is_rnaseq_evidence_stage(stage) for stage in current_stages) + latest_rnaseq_manifest = None + if request.job_id: + report_final_dir = WORKSPACE_ROOT / request.job_id / "final" + report_prefix = _nonempty_prefix(request.prefix or current_intent.get("prefix") or current_params.get("prefix")) + report_pattern = f"{report_prefix}.*.manifest.json" if report_prefix else "*.manifest.json" + candidates = [] + for candidate in report_final_dir.glob(report_pattern) if report_final_dir.exists() else []: + try: + payload = json.loads(candidate.read_text(encoding="utf-8")) + except Exception: + continue + if ( + is_rnaseq_evidence_stage(str(payload.get("stage") or "")) + and str(payload.get("status") or "") in {"success", "partial_success"} + ): + candidates.append((candidate.stat().st_mtime, candidate, payload)) + if candidates: + _mtime, _path, latest_rnaseq_manifest = max(candidates, key=lambda item: item[0]) + rnaseq_context = True + if explicit_rnaseq_report or rnaseq_context: + current_params["rnaseqReportAction"] = "open" if "打开" in query_lower_for_context else "generate" + if latest_rnaseq_manifest: + current_params.setdefault( + "rnaseqScopeId", + manifest_identity_value(latest_rnaseq_manifest, "rnaseq_scope_id"), + ) + report_prefix = _nonempty_prefix( + request.prefix + or current_intent.get("prefix") + or current_params.get("prefix") + or (latest_rnaseq_manifest or {}).get("prefix") + ) + request.intent = { + **current_intent, + "stages": ["rnaseq_report"], + "primary_stage": "rnaseq_report", + "task_type": "report", + "task_kind": "pipeline", + "prefix": report_prefix, + "params": {**current_params, "prefix": report_prefix}, + "confidence": 0.99, + } subfamily_confirm_query_for_context = any( kw in query_lower_for_context for kw in ["确认亚家族", "亚家族确认", "确认亚家族标签", "人工确认亚家族", "subfamily confirm", "confirmed subfamily"] @@ -20907,7 +21035,10 @@ async def agent_plan(request: AgentPlanRequest): "manifest_path": str(status.manifest_path) if status.manifest_path else "", "missing_count": len(status.missing_artifacts or []) + len(status.errors or []), } - if status.completed: + # Report requests are actions, not a request to skip an + # already completed computation. The CLI decides whether + # an "open" request can reuse the fresh snapshot. + if status.completed and stage_id != "rnaseq_report": completed_rnaseq_stages.append(stage_id) else: remaining_rnaseq_stages.append(stage_id) @@ -20923,6 +21054,7 @@ async def agent_plan(request: AgentPlanRequest): "rnaseq_fgsea": "RNA-seq fgsea", "rnaseq_wgcna": "RNA-seq WGCNA", "rnaseq_candidate_family": "RNA-seq Candidate Family", + "rnaseq_report": "RNA-seq Result Report", } plan_prefix = effective_plan_prefix or request.prefix or incoming_intent.get("prefix", "") final_dir_for_rnaseq = PROJECT_ROOT / "workspace" / request.job_id / "final" @@ -20942,6 +21074,7 @@ async def agent_plan(request: AgentPlanRequest): "rnaseq_fgsea": ["rnaseq_de"], "rnaseq_wgcna": ["rnaseq_sample_qc"], "rnaseq_candidate_family": ["rnaseq_de", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"], + "rnaseq_report": [], } if ( "rnaseq_fgsea" in remaining_rnaseq_stages @@ -20965,6 +21098,7 @@ async def agent_plan(request: AgentPlanRequest): "rnaseq_fgsea": 10, "rnaseq_wgcna": 20, "rnaseq_candidate_family": 10, + "rnaseq_report": 2, } stages_detail = [] for order, stage_id in enumerate(remaining_rnaseq_stages, start=1): diff --git a/gfa_core/mas/utils/intent_parser.py b/gfa_core/mas/utils/intent_parser.py index f75329b21..48cc0f25a 100755 --- a/gfa_core/mas/utils/intent_parser.py +++ b/gfa_core/mas/utils/intent_parser.py @@ -87,6 +87,7 @@ STAGE_KEYWORDS = { "rnaseq_kegg_enrich": ["rnaseq kegg", "rnaseq_kegg", "rna kegg", "kegg富集", "kegg分析", "kegg", "富集分析", "enrich"], "rnaseq_enrich": ["富集分析", "enrich", "go分析", "kegg", "clusterprofiler"], "rnaseq_candidate_family": ["候选家族", "候选基因家族", "家族排名", "tf注释", "tf family", "tffamily", "pfam注释", "deg注释", "差异基因注释", "差异基因家族", "candidate family"], + "rnaseq_report": ["rnaseq结果报告", "rna-seq结果报告", "转录组结果报告", "rnaseq报告", "rna-seq报告"], "ncbi_fetch_sequences": ["ncbi下载", "ncbi序列", "下载序列", "下载蛋白", "获取蛋白", "下载protein", "下载cds", "protein和cds", "蛋白和cds", "基因list", "gene list", "gene_list", "id列表", ".ids"], } @@ -256,6 +257,7 @@ RNASEQ_STAGES = { "rnaseq_kegg_enrich", "rnaseq_enrich", "rnaseq_candidate_family", + "rnaseq_report", } # 物种关键词映射 @@ -732,6 +734,10 @@ def parse_intent_from_query(query: str) -> Dict: "clarify_type": result.get("clarify_type"), "message": result.get("message") } + if parsed["primary_stage"] == "rnaseq_report": + parsed["params"] = { + "rnaseqReportAction": "open" if "打开" in query_lower else "generate" + } if ncbi_params: parsed["params"] = ncbi_params parsed["prefix"] = ncbi_params.get("prefix", "") @@ -781,6 +787,43 @@ def parse_intent_from_query(query: str) -> Dict: return parsed +def deterministic_agent_chat_intent( + query: str, + *, + rnaseq_context: bool = False, +) -> Optional[Dict]: + """Return explicit chat intents that must not depend on the Router model.""" + intent = parse_intent_from_query(query) + if intent.get("stages") == ["rnaseq_report"]: + return intent + query_lower = str(query or "").lower() + query_without_space = re.sub(r"\s+", "", query_lower) + is_generic_report_action = ( + rnaseq_context + and "结果报告" in query_without_space + and any(action in query_without_space for action in ("生成", "打开")) + ) + if is_generic_report_action: + return { + "gene_family": "", + "species": "", + "stages": ["rnaseq_report"], + "primary_stage": "rnaseq_report", + "task_type": "report", + "task_kind": "pipeline", + "confidence": 0.99, + "need_clarify": False, + "clarify_type": None, + "message": None, + "params": { + "rnaseqReportAction": ( + "open" if "打开" in query_without_space else "generate" + ) + }, + } + return None + + def parse_msa_region_plot_request(query: str) -> Dict: """Extract parameters for drawing a selected MSA alignment region.""" text = str(query or "").strip() @@ -1367,6 +1410,12 @@ def _extract_stages(query: str, query_lower: str) -> List[str]: stages.append("gwas_fmt") matched_keywords = [] # 记录匹配到的关键词长度，用于优先级判断 has_rnaseq_marker = any(marker in query_lower for marker in RNASEQ_MARKERS) + requests_result_report = any( + keyword in query_lower + for keyword in ["生成结果报告", "打开结果报告", "结果报告", "生成报告", "打开报告"] + ) + if has_rnaseq_marker and requests_result_report: + return ["rnaseq_report"] extraction_context = any(kw in query_lower for kw in ["提取", "extract", "gffread", "筛选", "subseq"]) explicit_gene_family_context = any(kw in query_lower for kw in ["基因家族", "家族表达", "表达量", "热图", "heatmap"]) curated_member_context = ( diff --git a/gfa_core/rnaseq_report.py b/gfa_core/rnaseq_report.py new file mode 100644 index 000000000..e7f42adf0 --- /dev/null +++ b/gfa_core/rnaseq_report.py @@ -0,0 +1,1304 @@ +"""Evidence-bound RNA-seq result report generation. + +This module deliberately keeps the formal report independent from the live +frontend interpretation cache. It reuses the same pure evidence builder, but +materializes a deterministic Markdown report and provenance artifacts under a +job's ``final/`` directory. +""" +from __future__ import annotations + +import hashlib +import ast +import csv +import json +import math +import os +import re +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, Iterable + +from gfa_core.rnaseq_evidence import ( + RNASEQ_STAGE_NAMES, + build_rnaseq_evidence_snapshot, + resolve_artifact_path, + manifest_identity_value, + select_rnaseq_manifests, +) + + +REPORT_STAGE = "rnaseq_report" +REPORT_RENDERER_VERSION = "4" +SUCCESS_STATUSES = {"success", "partial_success"} +DEFAULT_EXPECTED_STAGES = [ + "rnaseq_qc", + "rnaseq_align", + "rnaseq_quant", + "rnaseq_sample_qc", + "rnaseq_de", +] +PREFERRED_STAGE_ORDER = [ + "rnaseq_qc", + "rnaseq_align", + "rnaseq_quant", + "rnaseq_sample_qc", + "rnaseq_de", + "annotation", + "annotation_go", + "annotation_kegg", + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_enrich", + "rnaseq_fgsea", + "rnaseq_wgcna", + "rnaseq_candidate_family", + "rnaseq_family_score", +] +IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".svg", ".webp"} +STAGE_NAMES = RNASEQ_STAGE_NAMES + + +class NoSuccessfulRNASeqEvidence(RuntimeError): + """Raised when a report would otherwise be empty.""" + + +@dataclass(frozen=True) +class RNASeqReportResult: + report_path: Path + evidence_path: Path + model_path: Path + manifest_path: Path + coverage_status: str + included_stages: tuple[str, ...] + rnaseq_scope_id: str + reused: bool = False + + +def _now() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +def _read_json(path: Path) -> dict[str, Any]: + try: + data = json.loads(path.read_text(encoding="utf-8")) + return data if isinstance(data, dict) else {} + except (OSError, json.JSONDecodeError): + return {} + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _write_text_atomic(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") + temporary.write_text(text, encoding="utf-8") + temporary.replace(path) + + +def _identity(manifest: dict[str, Any], key: str) -> str: + return manifest_identity_value(manifest, key) + + +def is_rnaseq_report_source(stage: str) -> bool: + stage = str(stage or "").strip() + return stage != REPORT_STAGE and ( + stage.startswith("rnaseq_") or stage in {"annotation", "annotation_go", "annotation_kegg"} + ) + + +def _stage_sort_key(stage: str) -> tuple[int, str]: + try: + return (PREFERRED_STAGE_ORDER.index(stage), stage) + except ValueError: + return (len(PREFERRED_STAGE_ORDER), stage) + + +def _normalize_expected_stages(stages: Iterable[str] | None) -> list[str]: + raw = list(stages) if stages is not None else list(DEFAULT_EXPECTED_STAGES) + result: list[str] = [] + for stage in raw: + value = str(stage or "").strip() + if value and is_rnaseq_report_source(value) and value not in result: + result.append(value) + return result + + +def _source_selection( + final_dir: Path, + prefix: str, + rnaseq_scope_id: str, +) -> tuple[str, list[str], dict[str, dict[str, Any]]]: + selection = select_rnaseq_manifests( + final_dir, + prefix, + rnaseq_scope_id=rnaseq_scope_id, + ) + return selection.rnaseq_scope_id, list(selection.discovered_scope_ids), selection.manifests + + +def _source_records(manifests: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for stage in sorted(manifests, key=_stage_sort_key): + manifest = manifests[stage] + path = Path(str(manifest.get("_manifest_path") or "")) + records.append( + { + "stage": stage, + "status": str(manifest.get("status") or "unknown"), + "rnaseq_scope_id": _identity(manifest, "rnaseq_scope_id"), + "path": str(path), + "sha256": _sha256(path) if path.is_file() else "", + "mtime": datetime.fromtimestamp(path.stat().st_mtime).astimezone().isoformat(timespec="seconds") + if path.is_file() + else "", + } + ) + return records + + +def _fingerprint( + records: list[dict[str, Any]], expected_stages: list[str], use_model: bool +) -> tuple[str, str]: + source_payload = json.dumps(records, ensure_ascii=False, sort_keys=True).encode("utf-8") + source_fingerprint = hashlib.sha256(source_payload).hexdigest() + request_payload = json.dumps( + { + "source_fingerprint": source_fingerprint, + "expected_stages": expected_stages, + "use_model": use_model, + "report_renderer_version": REPORT_RENDERER_VERSION, + }, + ensure_ascii=False, + sort_keys=True, + ).encode("utf-8") + return source_fingerprint, hashlib.sha256(request_payload).hexdigest() + + +def _artifact_path(workspace: Path, manifest: dict[str, Any], item: dict[str, Any]) -> Path | None: + return resolve_artifact_path(workspace / "final", item, require_exists=True) + + +def _enrichment_figure_identity(stage: str, path: Path) -> tuple[str, str, str] | None: + name = path.name + lower = name.lower() + comparison = "" + category = "" + pathway_id = "" + if stage == "rnaseq_go_enrich" or (stage == "rnaseq_enrich" and ".go_enrich" in lower): + match = re.search(r"rnaseq_(?:go_enrich|enrich)\.(.+?)\.go_enrich", name, flags=re.IGNORECASE) + comparison = match.group(1) if match else "" + if ".go_enrich.ontology_barplot." in lower: + category = "go_ontology" + elif ".go_enrich.barplot." in lower: + category = "go_bar" + elif ".go_enrich.dotplot." in lower: + category = "go_dot" + elif re.search(r"\.go_enrich\.svg$", lower): + category = "go_bar" + elif stage == "rnaseq_kegg_enrich" or (stage == "rnaseq_enrich" and "kegg" in lower): + match = re.search(r"rnaseq_(?:kegg_enrich|enrich)\.(.+?)(?:\.kegg|_kegg)", name, flags=re.IGNORECASE) + comparison = match.group(1) if match else "" + pathway_match = re.search(r"\.kegg_pathway_map\.(ko\d+)\.", lower) + if pathway_match: + category = "kegg_pathway" + pathway_id = pathway_match.group(1) + elif re.search(r"(?:\.|_)kegg_enrich\.(?:png|jpe?g|webp|svg|pdf)$", lower): + category = "kegg_bar" + elif re.search(r"_kegg_dotplot\.(?:png|jpe?g|webp|svg|pdf)$", lower): + category = "kegg_dot" + elif stage == "rnaseq_fgsea": + match = re.search(r"rnaseq_fgsea\.(.+?)\.fgsea", name, flags=re.IGNORECASE) + comparison = match.group(1) if match else "" + if ".fgsea_kegg_dotplot." in lower: + category = "fgsea_dot" + elif ".fgsea_kegg_top_enrichment." in lower: + category = "fgsea_curve" + if not category: + return None + identity = f"{category}:{comparison}:{pathway_id}" if category == "kegg_pathway" else category + return identity, comparison, pathway_id + + +def _artifact_comparison(stage: str, path: Path, category: str = "") -> str: + patterns = { + "rnaseq_go_enrich": r"rnaseq_go_enrich\.(.+?)\.go_enrich", + "rnaseq_kegg_enrich": r"rnaseq_kegg_enrich\.(.+?)(?:\.kegg|_kegg)", + "rnaseq_fgsea": r"rnaseq_fgsea\.(.+?)\.fgsea", + } + if stage == "rnaseq_enrich": + pattern = ( + r"rnaseq_enrich\.(.+?)\.go_enrich" + if category.startswith("go_") + else r"rnaseq_enrich\.(.+?)(?:\.kegg|_kegg)" + ) + else: + pattern = patterns.get(stage) + match = re.search(pattern, path.name, flags=re.IGNORECASE) if pattern else None + return match.group(1) if match else "" + + +def _table_rows(path: Path) -> list[dict[str, str]]: + try: + with path.open("r", encoding="utf-8", errors="replace", newline="") as handle: + delimiter = "," if path.suffix.lower() == ".csv" else "\t" + return [ + {str(key): str(value or "") for key, value in row.items() if key is not None} + for row in csv.DictReader(handle, delimiter=delimiter) + ] + except OSError: + return [] + + +def _supporting_tables( + workspace: Path, + manifest: dict[str, Any], + comparison: str, + category: str = "", +) -> list[tuple[Path, list[dict[str, str]]]]: + tables: list[tuple[Path, list[dict[str, str]]]] = [] + for item in (manifest.get("artifacts") or {}).values(): + if not isinstance(item, dict): + continue + path = _artifact_path(workspace, manifest, item) + if ( + not path + or not path.is_file() + or path.suffix.lower() not in {".txt", ".tsv", ".csv"} + or ( + comparison + and _artifact_comparison(str(manifest.get("stage") or ""), path, category) != comparison + ) + ): + continue + lower_name = path.name.lower() + if category.startswith("go_") and not lower_name.endswith(".go_enrich.txt"): + continue + if category in {"kegg_bar", "kegg_dot"} and not lower_name.endswith(".kegg_enrich.txt"): + continue + if category == "kegg_pathway" and not ( + lower_name.endswith(".kegg_enrich.txt") + or ".kegg_pathway_gene_hits." in lower_name + or ".kegg_pathway_mapper." in lower_name + or ".kegg_pathway_map.ko_log2fc." in lower_name + ): + continue + if category.startswith("fgsea_") and not lower_name.endswith(".fgsea_kegg.tsv"): + continue + rows = _table_rows(path) + if rows: + tables.append((path, rows)) + return tables + + +def _best_adjusted_p(tables: list[tuple[Path, list[dict[str, str]]]]) -> float: + values: list[float] = [] + for _, rows in tables: + for row in rows: + raw = row.get("p.adjust") or row.get("padj") or row.get("FDR") or "" + try: + value = float(raw) + if math.isfinite(value) and 0.0 <= value <= 1.0: + values.append(value) + except (TypeError, ValueError): + continue + return min(values, default=float("inf")) + + +def _fgsea_selection_score(tables: list[tuple[Path, list[dict[str, str]]]]) -> tuple[float, float]: + scores: list[tuple[float, float]] = [] + for _, rows in tables: + for row in rows: + try: + padj = float(row.get("padj") or row.get("p.adjust") or "") + nes = float(row.get("NES") or "") + except (TypeError, ValueError): + continue + if math.isfinite(padj) and 0.0 <= padj <= 1.0 and math.isfinite(nes): + scores.append((padj, -abs(nes))) + return min(scores, default=(float("inf"), float("inf"))) + + +def _support_labels(value: str) -> list[str]: + """Parse both the real semicolon serialization and legacy list encodings.""" + raw = str(value or "").strip() + if not raw: + return [] + if raw[:1] in {"[", "("}: + try: + parsed = json.loads(raw) if raw.startswith("[") else ast.literal_eval(raw) + except (ValueError, SyntaxError, json.JSONDecodeError): + try: + parsed = ast.literal_eval(raw) + except (ValueError, SyntaxError): + parsed = None + if isinstance(parsed, (list, tuple)): + return list(dict.fromkeys(str(item).strip() for item in parsed if str(item).strip())) + return list(dict.fromkeys(item.strip() for item in raw.split(";") if item.strip())) + + +def _support_identity(label: str) -> tuple[str, str]: + raw = str(label or "").strip() + go_match = re.match(r"(?i)^GO:(GO:\d+)(?::.*)?$", raw) + if go_match: + return "go", go_match.group(1) + match = re.match(r"(?i)^(KEGG|fgsea):([^:]+)(?::.*)?$", raw) + if match: + return match.group(1).lower(), match.group(2).lower() + return "", "" + + +def _candidate_rows(workspace: Path, manifests: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: + manifest = manifests.get("rnaseq_family_score") or {} + item = (manifest.get("artifacts") or {}).get("analysis_ready_family") + if not isinstance(item, dict): + return [] + path = _artifact_path(workspace, manifest, item) + if not path or not path.is_file(): + return [] + rows: list[dict[str, Any]] = [] + for index, row in enumerate(_table_rows(path), 1): + family = str(row.get("family_id") or row.get("family_name") or "").strip() + comparison = str(row.get("best_comparison") or "").strip() + if not family or not comparison: + continue + try: + rank = int(str(row.get("rank") or index)) + except ValueError: + rank = index + rows.append( + { + "family": family, + "family_name": str(row.get("family_name") or "").strip(), + "rank": rank, + "comparison": comparison, + "support_labels": _support_labels(str(row.get("best_support_labels") or "")), + "source_table": str(path), + } + ) + return sorted(rows, key=lambda row: (row["rank"], row["family"])) + + +def _pathway_adjusted_p( + tables: list[tuple[Path, list[dict[str, str]]]], pathway_id: str +) -> float: + row = _first_matching_row(tables, pathway_id=pathway_id) + try: + value = float(row.get("p.adjust") or row.get("padj") or row.get("FDR") or "") + except (TypeError, ValueError): + return float("inf") + return value if math.isfinite(value) and 0.0 <= value <= 1.0 else float("inf") + + +def _figures( + workspace: Path, manifests: dict[str, dict[str, Any]] +) -> tuple[list[dict[str, Any]], list[dict[str, Any]], str]: + candidates: list[dict[str, Any]] = [] + seen: set[str] = set() + for stage in sorted(manifests, key=_stage_sort_key): + artifacts = manifests[stage].get("artifacts") or {} + if not isinstance(artifacts, dict): + continue + for key, item in sorted(artifacts.items()): + if not isinstance(item, dict): + continue + path = _artifact_path(workspace, manifests[stage], item) + if not path or not path.is_file(): + continue + classified = _enrichment_figure_identity(stage, path) + if path.suffix.lower() not in IMAGE_SUFFIXES and not (path.suffix.lower() == ".pdf" and classified): + continue + identity = str(path.resolve()) + if identity in seen: + continue + seen.add(identity) + row = {"stage": stage, "artifact": str(key), "path": str(path)} + if classified: + identity, comparison, pathway_id = classified + row.update( + { + "selection_identity": identity, + "category": identity.split(":", 1)[0], + "comparison": comparison, + "pathway_id": pathway_id, + } + ) + candidates.append(row) + + selected: list[dict[str, Any]] = [] + enrichment_candidates: list[dict[str, Any]] = [] + for row in candidates: + if not row.get("selection_identity"): + row.update( + candidate_families=[], support_labels=[], selection_reason="all_stage_figures", + source_tables=[], + ) + selected.append(row) + else: + enrichment_candidates.append(row) + + formal_candidates = _candidate_rows(workspace, manifests) + format_rank = {".png": 0, ".jpg": 1, ".jpeg": 1, ".webp": 2, ".svg": 3, ".pdf": 4} + table_cache: dict[tuple[str, str, str], list[tuple[Path, list[dict[str, str]]]]] = {} + + def tables_for(row: dict[str, Any]) -> list[tuple[Path, list[dict[str, str]]]]: + cache_key = (row["stage"], row.get("comparison", ""), row.get("category", "")) + if cache_key not in table_cache: + table_cache[cache_key] = _supporting_tables( + workspace, manifests[row["stage"]], cache_key[1], cache_key[2] + ) + return table_cache[cache_key] + + def table_contains(category: str, identifier: str, row: dict[str, Any]) -> bool: + target_identifier = identifier.lower() + if category == "kegg_pathway": + return True + tables = tables_for(row) + if category == "fgsea_curve": + top = _best_fgsea_row(tables) + top_identifier = str(top.get("pathway") or top.get("PathwayID") or top.get("ID") or "") + return bool(top_identifier) and top_identifier.lower() == target_identifier + for _, rows in tables: + for table_row in rows: + value = str( + table_row.get("ID") + or table_row.get("PathwayID") + or table_row.get("pathway") + or "" + ) + if value.lower() == target_identifier: + return True + return False + + def row_support(row: dict[str, Any]) -> tuple[list[str], list[str], int, list[str]]: + category = row.get("category", "") + wanted_kind = ( + "go" if category.startswith("go_") else "fgsea" if category.startswith("fgsea_") else "kegg" + ) + pathway_id = str(row.get("pathway_id") or "").lower() + matched: list[tuple[int, str, str, str]] = [] + for candidate in formal_candidates: + if candidate["comparison"] != row.get("comparison"): + continue + for label in candidate["support_labels"]: + kind, identifier = _support_identity(label) + if ( + kind != wanted_kind + or (category == "kegg_pathway" and identifier != pathway_id) + or not table_contains(category, identifier, row) + ): + continue + matched.append( + (candidate["rank"], candidate["family"], label, candidate["source_table"]) + ) + matched.sort(key=lambda value: (value[0], value[1], value[2])) + return ( + list(dict.fromkeys(value[1] for value in matched)), + list(dict.fromkeys(value[2] for value in matched)), + matched[0][0] if matched else 10**9, + list(dict.fromkeys(value[3] for value in matched)), + ) + + def evidence_score(row: dict[str, Any]) -> tuple[Any, ...]: + tables = tables_for(row) + if row.get("category") == "fgsea_curve": + evidence: tuple[Any, ...] = _fgsea_selection_score(tables) + elif row.get("category") == "kegg_pathway": + evidence = (_pathway_adjusted_p(tables, row.get("pathway_id", "")),) + else: + evidence = (_best_adjusted_p(tables),) + return (*evidence, format_rank.get(Path(row["path"]).suffix.lower(), 9), Path(row["path"]).name) + + for row in enrichment_candidates: + families, labels, candidate_rank, candidate_tables = row_support(row) + row["candidate_families"] = families + row["support_labels"] = labels + row["candidate_rank"] = candidate_rank + row["source_tables"] = candidate_tables + [str(path) for path, _ in tables_for(row)] + + selection_mode = ( + "candidate_guided" + if any(row.get("support_labels") for row in enrichment_candidates) + else "representative_fallback" + ) + if selection_mode == "representative_fallback": + for row in enrichment_candidates: + row["candidate_families"] = [] + row["support_labels"] = [] + row["candidate_rank"] = 10**9 + row["source_tables"] = [str(path) for path, _ in tables_for(row)] + + groups: dict[str, list[dict[str, Any]]] = {} + for row in enrichment_candidates: + category = row.get("category", "") + if selection_mode == "candidate_guided" and not row.get("support_labels"): + continue + key = ( + f"{category}:{row.get('comparison', '')}:{row.get('pathway_id', '')}" + if selection_mode == "candidate_guided" and category == "kegg_pathway" + else f"{category}:{row.get('pathway_id', '')}" + if category == "kegg_pathway" + else category + ) + groups.setdefault(key, []).append(row) + + for rows in groups.values(): + rows.sort( + key=lambda row: ( + row.get("candidate_rank", 10**9) if selection_mode == "candidate_guided" else 0, + *evidence_score(row), + ) + ) + chosen = rows[0] + chosen["selection_reason"] = ( + "candidate_family_support" if selection_mode == "candidate_guided" else "representative_fallback" + ) + selected.append(chosen) + + selected_paths = {row["path"] for row in selected} + omitted: list[dict[str, Any]] = [] + for row in enrichment_candidates: + if row["path"] in selected_paths: + continue + row.setdefault("candidate_families", []) + row.setdefault("support_labels", []) + row.setdefault("source_tables", []) + row["selection_reason"] = ( + "not_candidate_supported" + if selection_mode == "candidate_guided" and not row["support_labels"] + else "lower_ranked_representative" + ) + omitted.append(row) + stage_rank = {stage: index for index, stage in enumerate(PREFERRED_STAGE_ORDER)} + ordered = sorted( + selected, + key=lambda row: (stage_rank.get(row["stage"], len(stage_rank)), Path(row["path"]).name), + ) + omitted.sort(key=lambda row: (stage_rank.get(row["stage"], len(stage_rank)), Path(row["path"]).name)) + return ordered, omitted, selection_mode + + +def _first_matching_row( + tables: list[tuple[Path, list[dict[str, str]]]], + *, + pathway_id: str = "", +) -> dict[str, str]: + rows = [row for _, table_rows in tables for row in table_rows] + if pathway_id: + for row in rows: + identifier = row.get("ID") or row.get("PathwayID") or row.get("pathway") or "" + if identifier.lower() == pathway_id.lower(): + return row + return {} + return rows[0] if rows else {} + + +def _best_fgsea_row(tables: list[tuple[Path, list[dict[str, str]]]]) -> dict[str, str]: + ranked: list[tuple[float, float, int, dict[str, str]]] = [] + for _, rows in tables: + for index, row in enumerate(rows): + try: + padj = float(row.get("padj") or row.get("p.adjust") or "") + nes = float(row.get("NES") or "") + except (TypeError, ValueError): + continue + if math.isfinite(padj) and 0.0 <= padj <= 1.0 and math.isfinite(nes): + ranked.append((padj, -abs(nes), index, row)) + return min(ranked, default=(float("inf"), float("inf"), 0, {}))[-1] + + +def _value(row: dict[str, str], *keys: str) -> str: + for key in keys: + value = str(row.get(key) or "").strip() + if value: + return value + return "未登记" + + +def _probability_value(row: dict[str, str], *keys: str) -> str: + raw = _value(row, *keys) + try: + value = float(raw) + except (TypeError, ValueError): + return "未登记" + return raw if math.isfinite(value) and 0.0 <= value <= 1.0 else "未登记" + + +def _field_items(row: dict[str, str], *keys: str) -> set[str]: + value = next((str(row.get(key) or "").strip() for key in keys if str(row.get(key) or "").strip()), "") + return {item for item in re.split(r"[/,;\s]+", value) if item} + + +def _ordered_field_items(row: dict[str, str], *keys: str) -> list[str]: + value = next((str(row.get(key) or "").strip() for key in keys if str(row.get(key) or "").strip()), "") + return list(dict.fromkeys(item for item in re.split(r"[/,;\s]+", value) if item)) + + +def _item_summary(label: str, items: set[str]) -> str: + ordered = sorted(items) + preview = "/".join(ordered[:12]) if ordered else "未登记" + return f"{label}总数={len(ordered)}，前12项={preview}" + + +def _source_table_links(report_path: Path, tables: list[tuple[Path, list[dict[str, str]]]], markers: tuple[str, ...]) -> str: + links = [ + f"[{path.name}]({_relative_link(report_path, path)})" + for path, _ in tables + if any(marker in path.name.lower() for marker in markers) + ] + return "；完整源表：" + "、".join(links) if links else "" + + +def _figure_interpretation( + figure: dict[str, str], workspace: Path, manifests: dict[str, dict[str, Any]], report_path: Path +) -> list[str]: + category = figure.get("category", "") + if not category: + return [f"该图来自 `{figure['stage']}` 阶段。具体趋势应与该阶段的数值表和 manifest 指标联合判断。"] + comparison = figure.get("comparison", "") + tables = _supporting_tables(workspace, manifests[figure["stage"]], comparison, category) + pathway_id = figure.get("pathway_id", "") + row = ( + _best_fgsea_row(tables) + if category.startswith("fgsea_") + else _first_matching_row(tables, pathway_id=pathway_id) + ) + if not row: + result = f"比较组 `{comparison or '未识别'}` 证据不足：缺少可匹配的数值表，无法可靠判断本次数据的具体趋势。" + elif category == "go_ontology": + ontology_rows: dict[str, dict[str, str]] = {} + for _, table_rows in tables: + for candidate in table_rows: + ontology_value = str(candidate.get("Ontology") or "").strip().lower() + ontology = { + "bp": "BP", + "biological process": "BP", + "cc": "CC", + "cellular component": "CC", + "mf": "MF", + "molecular function": "MF", + }.get(ontology_value, "") + if ontology in {"BP", "CC", "MF"} and ontology not in ontology_rows: + ontology_rows[ontology] = candidate + summaries = [ + f"{ontology}={_value(ontology_rows[ontology], 'Description')}" + for ontology in ("BP", "CC", "MF") + if ontology in ontology_rows + ] + result = ( + f"比较组 `{comparison}` 的三类本体代表条目为" + f"{'；'.join(summaries) if summaries else '未登记'}。" + ) + elif category.startswith("go_"): + result = ( + f"比较组 `{comparison}` 的代表条目为 {_value(row, 'ID')}（{_value(row, 'Description')}），" + f"Ontology={_value(row, 'Ontology')}，GeneRatio={_value(row, 'GeneRatio')}，" + f"Count={_value(row, 'Count')}，p.adjust={_probability_value(row, 'p.adjust')}。" + ) + elif category in {"kegg_bar", "kegg_dot"}: + genes = _field_items(row, "geneID", "Genes") + kos = set(re.findall(r"K\d+", " ".join(str(row.get(key) or "") for key in ("KO", "KOs", "ko")))) + result = ( + f"比较组 `{comparison}` 的代表通路为 {_value(row, 'ID')}（{_value(row, 'Description')}），" + f"GeneRatio={_value(row, 'GeneRatio')}，Count={_value(row, 'Count')}，" + f"p.adjust={_probability_value(row, 'p.adjust', 'FDR')}，" + f"{_item_summary('命中基因', genes)}，{_item_summary('KO', kos)}" + f"{_source_table_links(report_path, tables, ('.kegg_enrich.txt',))}。" + ) + elif category == "kegg_pathway": + pathway_genes: set[str] = set() + pathway_kos: set[str] = set() + for _, table_rows in tables: + for candidate in table_rows: + identifier = candidate.get("PathwayID") or candidate.get("ID") or candidate.get("pathway") or "" + if identifier.lower() == pathway_id.lower(): + pathway_genes.update( + gene + for gene in re.split(r"[/,;\s]+", candidate.get("Genes") or candidate.get("geneID") or "") + if gene + ) + pathway_kos.update(re.findall(r"K\d+", candidate.get("KO") or "")) + if not pathway_genes: + pathway_genes.update( + gene for gene in re.split(r"[/,;\s]+", _value(row, "Genes", "geneID")) if gene != "未登记" + ) + ko_changes: list[float] = [] + if pathway_kos: + for table_path, table_rows in tables: + if "ko_log2fc" not in table_path.name.lower(): + continue + for candidate in table_rows: + ko = candidate.get("KO") or candidate.get("ko") or "" + if ko not in pathway_kos: + continue + try: + ko_changes.append(float(candidate.get("log2FoldChange") or candidate.get("log2FC") or "")) + except (TypeError, ValueError): + continue + direction = "" + if ko_changes: + direction = f"，KO变化中上调{sum(value > 0 for value in ko_changes)}个、下调{sum(value < 0 for value in ko_changes)}个" + elif not pathway_kos: + direction = "，KO对应关系证据不足，未汇总KO表达方向" + result = ( + f"比较组 `{comparison}` 的通路 {pathway_id}（{_value(row, 'PathwayName', 'Description', 'pathway_name')}），" + f"p.adjust={_probability_value(row, 'p.adjust', 'padj', 'FDR')}，" + f"{_item_summary('命中基因', pathway_genes)}，{_item_summary('KO', pathway_kos)}" + f"{direction}" + f"{_source_table_links(report_path, tables, ('kegg_pathway_gene_hits', 'ko_log2fc'))}。" + ) + else: + leading_edge = _ordered_field_items(row, "leadingEdge") + leading_edge_preview = "/".join(leading_edge[:12]) if leading_edge else "未登记" + result = ( + f"比较组 `{comparison}` 的通路 {_value(row, 'pathway', 'ID')}（{_value(row, 'pathway_name', 'Description')}），" + f"NES={_value(row, 'NES')}，padj={_probability_value(row, 'padj', 'p.adjust')}，" + f"leadingEdge总数={len(leading_edge)}，原顺序前12项={leading_edge_preview}" + f"{_source_table_links(report_path, tables, ('.fgsea_kegg.tsv',))}。" + ) + + meanings = { + "go_bar": "GO富集条形图概括显著功能条目及其富集强度。", + "go_dot": "GO富集气泡图同时呈现功能条目、富集比例、基因数和显著性。", + "go_ontology": "GO本体分面图并列展示BP、CC和MF三类功能条目。", + "kegg_bar": "KEGG富集条形图概括显著通路及其富集比例。", + "kegg_dot": "KEGG富集气泡图同时呈现通路富集比例、命中基因数和显著性。", + "kegg_pathway": "KEGG Pathway图将该比较组的KO表达变化映射到具体通路节点。", + "fgsea_dot": "FGSEA气泡图汇总基于全基因排序的KEGG通路NES及显著性。", + "fgsea_curve": "FGSEA富集曲线展示代表通路成员在全基因排序中的累积富集位置。", + } + readings = { + "go_bar": "结合条目名称、柱长、Count及校正后P值阅读。", + "go_dot": "结合横轴富集比例、点大小所示Count及颜色所示显著性阅读。", + "go_ontology": "分别比较BP、CC和MF分面中的条目、比例和显著性。", + "kegg_bar": "结合通路名称、柱长、GeneRatio、Count及校正后P值阅读。", + "kegg_dot": "结合GeneRatio、点大小所示Count及颜色所示显著性阅读。", + "kegg_pathway": "按图例读取节点颜色，并结合命中基因、KO及log2FoldChange表判断变化方向。", + "fgsea_dot": "NES正负表示富集方向，点大小和颜色需结合size与padj阅读。", + "fgsea_curve": "结合运行富集分数、成员命中位置和leadingEdge基因阅读。", + } + boundary = ( + "富集结果表示统计关联，不足以单独证明功能或通路被激活、抑制或具有因果调控关系。" + if not category.startswith("fgsea_") + else "FGSEA基于全基因排序，不能与仅使用显著差异基因的ORA结果混为一谈，也不证明因果关系。" + ) + return [ + f"图形含义：{meanings[category]}", + f"阅读方法：{readings[category]}", + f"本次结果：{result}", + f"解释边界：{boundary}", + ] + + +def _relative_link(report_path: Path, target: Path) -> str: + try: + return os.path.relpath(target, report_path.parent).replace(os.sep, "/") + except ValueError: + return str(target) + + +def _render_deterministic_markdown( + *, + prefix: str, + scope_id: str, + evidence: dict[str, Any], + manifests: dict[str, dict[str, Any]], + expected_stages: list[str], + stage_statuses: dict[str, str], + coverage_status: str, + report_path: Path, + workspace: Path, + generated_at: str, + figures: list[dict[str, Any]], + model_summary: str = "", +) -> str: + metadata = evidence.get("metadata") or {} + included = list(metadata.get("included_stages") or []) + flow_rows = {row.get("id"): row for row in (evidence.get("stage_flow") or {}).get("stages", [])} + findings = {row.get("stage"): row for row in evidence.get("key_findings") or []} + reviews = {row.get("stage"): row for row in evidence.get("stage_reviews") or []} + evidence_cards = evidence.get("evidence_cards") or [] + risks = evidence.get("risk_flags") or [] + missing = [stage for stage in expected_stages if stage_statuses.get(stage) not in SUCCESS_STATUSES] + stage_status_text = "; ".join( + f"{stage}={stage_statuses.get(stage, 'missing')}" for stage in expected_stages + ) + + lines = [ + f"# {prefix} RNA-seq 结果报告", + "", + "> 报告生成原则：结果判断只来自脚本、Manifest 和结果表；API 仅参与已有证据语句的排序和组织，不参与结果分析或推断。", + "", + "## 0. 报告摘要", + "", + f"- 生成时间：{generated_at}", + f"- RNA-seq scope：{scope_id or 'legacy/unscoped'}", + f"- 证据覆盖：{coverage_status}（成功/可用阶段 {len(included)} 个）", + f"- 已纳入阶段：{', '.join(included)}", + f"- 当前缺失或未成功阶段：{', '.join(missing) if missing else '无'}", + "", + str(evidence.get("biological_story") or "现有证据不足以形成跨阶段生物学叙事，请查看分阶段结果。"), + ] + major_claims = evidence.get("major_claims") or [] + if major_claims: + lines.extend(["", "### 主要证据结论", ""]) + for claim in major_claims[:10]: + text = claim.get("claim") or claim.get("text") or claim.get("title") or str(claim) + lines.append(f"- {text}") + lines.extend( + [ + "", + "### 模型辅助语言润色（不作为新增证据）", + "", + "> API 仅参与已有证据语句的排序和组织，不参与结果分析或推断；不能新增数字、基因、通路、显著性、变化方向、机制或因果结论。", + "", + model_summary or "本次未采用模型组织结果；正式结论以脚本生成的确定性正文和可追溯结果数据为准。", + ] + ) + lines.extend( + [ + "", + "## 1. 分析范围与证据边界", + "", + f"- 计划覆盖阶段：{', '.join(expected_stages)}", + f"- 阶段状态：{stage_status_text}", + f"- 解读边界：{metadata.get('interpretation_boundary') or '仅陈述可由当前产物支持的结果，不推断调控因果。'}", + "- API 边界：只允许排序和分组脚本批准的证据语句；任何自由文本、未知证据编号或无依据内容均被拒绝并回退。", + "- 本报告不读取前端会话缓存，也不调用“深度解读”接口；二者只共享纯证据构建逻辑。", + "", + "## 2. 数据质量与流程概览", + "", + (evidence.get("stage_flow") or {}).get("visualization", " -> ".join(included)), + ] + ) + for stage in included: + flow = flow_rows.get(stage) or {} + finding = findings.get(stage) or {} + review = reviews.get(stage) or {} + stage_cards = [item for item in evidence_cards if item.get("stage") == stage] + metric_cards = [item for item in stage_cards if str(item.get("claim") or "").startswith("指标 ")] + artifact_cards = [item for item in stage_cards if str(item.get("claim") or "").startswith("产物 ")] + displayed_cards = (metric_cards[:8] + artifact_cards[:4])[:12] + lines.extend(["", f"## 3.{included.index(stage) + 1} {STAGE_NAMES.get(stage, stage)}", ""]) + lines.append(f"- 状态：{flow.get('status') or manifests.get(stage, {}).get('status', 'unknown')}") + if flow.get("key_result"): + lines.append(f"- 关键结果：{flow['key_result']}") + description = finding.get("description") or finding.get("summary") or "" + if description: + lines.extend(["", str(description)]) + for card in displayed_cards: + lines.append( + f"- 证据 {card.get('id', '')}：{card.get('claim', '')}；{card.get('evidence', '')}" + ) + limitations = review.get("limitations") or [] + if limitations: + lines.append(f"- 阶段局限：{'；'.join(map(str, limitations))}") + + lines.extend(["", "## 4. 图表与产物说明", ""]) + selection_mode = (evidence.get("formal_report") or {}).get("selection_mode") + if selection_mode == "representative_fallback": + lines.extend( + [ + "> 候选家族驱动筛选不可用；当前富集图按图形类型与统计证据选取代表，不声称与候选家族相关。", + "", + ] + ) + elif selection_mode == "candidate_guided": + lines.extend([ + "> GO/KEGG/FGSEA 图按正式候选家族榜的比较组与功能支持标签筛选。", + "", + ]) + if figures: + for index, figure in enumerate(figures, 1): + path = Path(figure["path"]) + media_line = ( + f"[{figure['artifact']}]({_relative_link(report_path, path)})" + if path.suffix.lower() == ".pdf" + else f"![{figure['artifact']}]({_relative_link(report_path, path)})" + ) + lines.extend( + [ + f"### 图 {index}. {STAGE_NAMES.get(figure['stage'], figure['stage'])} — {figure['artifact']}", + "", + media_line, + "", + ] + ) + if figure.get("candidate_families"): + lines.append(f"- 候选家族关联：{', '.join(figure['candidate_families'])}") + if figure.get("support_labels"): + lines.append(f"- 正式支持标签：{'; '.join(figure['support_labels'])}") + lines.extend([*_figure_interpretation(figure, workspace, manifests, report_path), ""]) + else: + lines.append("当前成功阶段的 manifest 未登记可嵌入图像；请以分阶段表格和日志为准。") + + lines.extend(["", "## 5. 风险、局限性与解释约束", ""]) + if risks: + for risk in risks: + lines.append(f"- [{risk.get('severity', 'unknown')}] {risk.get('message', '')}") + else: + lines.append("- 自动证据审计未发现已登记风险；这不等同于不存在实验设计或批次效应风险。") + if missing: + lines.append(f"- 当前报告为部分覆盖，尚不能替代缺失阶段：{', '.join(missing)}。") + + lines.extend(["", "## 6. 下一步建议", ""]) + next_steps = evidence.get("next_steps") or [] + if next_steps: + for item in next_steps: + if isinstance(item, dict): + text = item.get("action") or item.get("recommendation") or item.get("text") or str(item) + else: + text = str(item) + lines.append(f"- {text}") + else: + lines.append("- 根据缺失阶段和风险项补齐分析，再重新生成报告以刷新证据快照。") + + lines.extend(["", "## 7. 证据追溯", ""]) + for stage in included: + manifest_path = manifests.get(stage, {}).get("_manifest_path") or "" + lines.append(f"- `{stage}`：`{manifest_path}`") + lines.extend(["", "---", "", "报告正文结束。完整结构化证据见同名 `.evidence.json`。", ""]) + return "\n".join(lines) + + +def _model_claim_catalog(evidence: dict[str, Any]) -> list[dict[str, str]]: + catalog: list[dict[str, str]] = [] + + def add(claim_id: str, text: Any, source_type: str, stage: str = "") -> None: + value = str(text or "").strip() + if value: + catalog.append( + { + "claim_id": claim_id, + "text": value, + "source_type": source_type, + "stage": stage, + } + ) + + add("story_1", evidence.get("biological_story"), "biological_story") + for index, row in enumerate(evidence.get("key_findings") or [], 1): + if not isinstance(row, dict): + continue + add( + f"finding_{index}", + row.get("description") or row.get("summary") or row.get("claim") or row.get("text"), + "key_finding", + str(row.get("stage") or ""), + ) + for index, row in enumerate(evidence.get("major_claims") or [], 1): + if isinstance(row, dict): + text = row.get("claim") or row.get("text") or row.get("title") + stage = str(row.get("stage") or "") + else: + text = row + stage = "" + add(f"claim_{index}", text, "major_claim", stage) + for index, row in enumerate(evidence.get("risk_flags") or [], 1): + if isinstance(row, dict): + text = row.get("message") or row.get("text") + stage = str(row.get("stage") or "") + else: + text = row + stage = "" + add(f"risk_{index}", text, "risk_flag", stage) + return catalog + + +def _render_claim_groups( + plan: dict[str, Any], catalog: list[dict[str, str]] +) -> tuple[str, list[str]]: + groups = plan.get("paragraph_groups") + if not isinstance(groups, list) or not groups or len(groups) > 4: + raise ValueError("paragraph_groups must contain between 1 and 4 groups") + claim_text = {row["claim_id"]: row["text"] for row in catalog} + selected: list[str] = [] + paragraphs: list[str] = [] + for group in groups: + if not isinstance(group, list) or not group: + raise ValueError("each paragraph group must be a non-empty claim-id list") + sentences: list[str] = [] + for raw_claim_id in group: + claim_id = str(raw_claim_id or "").strip() + if claim_id not in claim_text: + raise ValueError(f"unknown evidence claim id: {claim_id}") + if claim_id in selected: + raise ValueError(f"duplicate evidence claim id: {claim_id}") + selected.append(claim_id) + text = claim_text[claim_id] + sentences.append(text if text.endswith(("。", "！", "？", ".", "!", "?")) else f"{text}。") + paragraphs.append("".join(sentences)) + return "\n\n".join(paragraphs), selected + + +def _model_enhancement(evidence: dict[str, Any], enabled: bool) -> tuple[str, dict[str, Any]]: + claim_catalog = _model_claim_catalog(evidence) + metadata: dict[str, Any] = { + "schema": "rnaseq_report_model_v2", + "enabled": enabled, + "status": "disabled" if not enabled else "pending", + "provider": "", + "model": "", + "content": "", + "error": "", + "generated_at": _now(), + "policy": "evidence_claim_ordering_only", + "claim_catalog": claim_catalog, + "organization_plan": {}, + "selected_claim_ids": [], + } + if not enabled: + return "", metadata + if not claim_catalog: + metadata.update({"status": "fallback", "error": "no approved evidence claims available"}) + return "", metadata + try: + from gfa_core.mas.config import config + from gfa_core.mas.llm import LLMManager + from gfa_core.mas.llm.base import LLMMessage + + manager = LLMManager(config.load_llm_config()) + response = manager.call( + [ + LLMMessage( + role="system", + content=( + "你只能组织已批准的RNA-seq证据语句，不能撰写或改写结论。" + "只返回JSON对象：paragraph_groups为1至4个段落，每个段落只能列出输入中的claim_id。" + "不得返回自由文本、未知claim_id、数字、基因、通路、显著性、方向、机制或因果推断。" + ), + ), + LLMMessage( + role="user", + content=json.dumps( + {"task": "order_and_group_evidence_claims", "claim_catalog": claim_catalog}, + ensure_ascii=False, + )[:30000], + ), + ], + allow_fallback=True, + max_tokens=600, + track_agent_name="RNASeqReport", + ) + content = str(response.content or "").strip() + metadata.update( + { + "provider": str(response.provider or ""), + "model": str(response.model or ""), + "content": content, + } + ) + cleaned = re.sub(r"^```(?:json)?\s*|\s*```$", "", content, flags=re.IGNORECASE) + plan = json.loads(cleaned) + if not isinstance(plan, dict): + raise ValueError("model organization plan must be a JSON object") + rendered, selected_claim_ids = _render_claim_groups(plan, claim_catalog) + metadata.update( + { + "status": "embedded_evidence_only", + "organization_plan": plan, + "selected_claim_ids": selected_claim_ids, + } + ) + return rendered, metadata + except Exception as exc: # deterministic report remains authoritative + metadata.update({"status": "fallback", "error": f"{type(exc).__name__}: {exc}"}) + return "", metadata + + +def _artifact(workspace: Path, path: Path, kind: str) -> dict[str, Any]: + return { + "artifact": kind, + "path": str(path.relative_to(workspace)), + "full_path": str(path.absolute()), + "type": "markdown" if path.suffix == ".md" else "json", + "size": path.stat().st_size, + "sha256": _sha256(path), + } + + +def generate_rnaseq_report( + workspace: Path | str, + prefix: str, + *, + rnaseq_scope_id: str = "", + expected_stages: Iterable[str] | None = None, + use_model: bool = True, + action: str = "generate", + job_id: str = "", +) -> RNASeqReportResult: + workspace = Path(workspace).absolute() + final_dir = workspace / "final" + final_dir.mkdir(parents=True, exist_ok=True) + prefix = str(prefix or "").strip() + if not prefix: + raise ValueError("prefix is required") + if action not in {"generate", "open"}: + raise ValueError("action must be 'generate' or 'open'") + + expected = _normalize_expected_stages(expected_stages) + resolved_scope, discovered_scopes, all_sources = _source_selection(final_dir, prefix, rnaseq_scope_id) + successful = { + stage: manifest + for stage, manifest in all_sources.items() + if str(manifest.get("status") or "") in SUCCESS_STATUSES + } + if not successful: + raise NoSuccessfulRNASeqEvidence("暂无可生成的 RNA-seq 结果：未找到当前 prefix/scope 下成功的 RNA-seq 阶段。") + + included = sorted(successful, key=_stage_sort_key) + source_records = _source_records(all_sources) + source_fingerprint, request_fingerprint = _fingerprint(source_records, expected, use_model) + report_path = final_dir / f"{prefix}.rnaseq_report.md" + evidence_path = final_dir / f"{prefix}.rnaseq_report.evidence.json" + model_path = final_dir / f"{prefix}.rnaseq_report.model.json" + manifest_path = final_dir / f"{prefix}.rnaseq_report.manifest.json" + + stage_statuses = {stage: str(all_sources.get(stage, {}).get("status") or "missing") for stage in expected} + coverage_status = "complete" if expected and all(stage_statuses[s] == "success" for s in expected) else "partial" + if action == "open" and all(path.is_file() for path in (report_path, evidence_path, model_path, manifest_path)): + previous = _read_json(manifest_path) + metrics = previous.get("metrics") or {} + if ( + previous.get("status") == "success" + and _identity(previous, "rnaseq_scope_id") == resolved_scope + and metrics.get("request_fingerprint") == request_fingerprint + and metrics.get("report_renderer_version") == REPORT_RENDERER_VERSION + ): + return RNASeqReportResult( + report_path, + evidence_path, + model_path, + manifest_path, + str(metrics.get("coverage_status") or coverage_status), + tuple(metrics.get("included_stages") or included), + resolved_scope, + reused=True, + ) + + evidence = build_rnaseq_evidence_snapshot( + final_dir, + prefix, + rnaseq_scope_id=resolved_scope, + include_stages=included, + ) + if not evidence: + raise NoSuccessfulRNASeqEvidence("暂无可生成的 RNA-seq 结果：成功 manifest 未能构建有效证据。") + generated_at = _now() + figures, omitted_figures, selection_mode = _figures(workspace, successful) + enrichment_figures = [figure for figure in figures if figure.get("category")] + + def inventory_row(figure: dict[str, Any]) -> dict[str, Any]: + return { + "stage": figure["stage"], + "artifact": figure["artifact"], + "filename": Path(figure["path"]).name, + "category": figure.get("category", "other"), + "comparison": figure.get("comparison", ""), + "pathway_id": figure.get("pathway_id", ""), + "candidate_families": list(figure.get("candidate_families") or []), + "support_labels": list(figure.get("support_labels") or []), + "selection_reason": figure.get("selection_reason", ""), + "source_tables": list(figure.get("source_tables") or []), + } + + figure_inventory = [inventory_row(figure) for figure in figures] + omitted_figure_inventory = [inventory_row(figure) for figure in omitted_figures] + evidence["formal_report"] = { + "schema": "rnaseq_formal_report_evidence_v1", + "report_renderer_version": REPORT_RENDERER_VERSION, + "generated_at": generated_at, + "prefix": prefix, + "rnaseq_scope_id": resolved_scope, + "discovered_scope_ids": discovered_scopes, + "expected_stages": expected, + "stage_statuses": stage_statuses, + "coverage_status": coverage_status, + "source_fingerprint": source_fingerprint, + "request_fingerprint": request_fingerprint, + "source_manifests": source_records, + "selection_mode": selection_mode, + "figure_inventory": figure_inventory, + "omitted_figure_inventory": omitted_figure_inventory, + "omitted_figure_count": len(omitted_figure_inventory), + "figure_count": len(figures), + "enrichment_figure_count": len(enrichment_figures), + } + model_summary, model_metadata = _model_enhancement(evidence, use_model) + markdown = _render_deterministic_markdown( + prefix=prefix, + scope_id=resolved_scope, + evidence=evidence, + manifests=successful, + expected_stages=expected, + stage_statuses=stage_statuses, + coverage_status=coverage_status, + report_path=report_path, + workspace=workspace, + generated_at=generated_at, + figures=figures, + model_summary=model_summary, + ) + _write_text_atomic(evidence_path, json.dumps(evidence, ensure_ascii=False, indent=2) + "\n") + _write_text_atomic(model_path, json.dumps(model_metadata, ensure_ascii=False, indent=2) + "\n") + _write_text_atomic(report_path, markdown) + + manifest = { + "manifest_version": "1.0", + "stage": REPORT_STAGE, + "status": "success", + "job_id": job_id or workspace.name, + "prefix": prefix, + "rnaseq_scope_id": resolved_scope, + "generated_at": generated_at, + "parameters": { + "action": action, + "model_enhancement_requested": use_model, + "expected_stages": expected, + "rnaseq_scope_id": resolved_scope, + }, + "metrics": { + "coverage_status": coverage_status, + "included_stage_count": len(included), + "included_stages": included, + "expected_stages": expected, + "stage_statuses": stage_statuses, + "missing_or_failed_stages": [stage for stage in expected if stage_statuses.get(stage) != "success"], + "model_status": model_metadata["status"], + "source_fingerprint": source_fingerprint, + "request_fingerprint": request_fingerprint, + "report_renderer_version": REPORT_RENDERER_VERSION, + "selection_mode": selection_mode, + "figure_count": len(figures), + "enrichment_figure_count": len(enrichment_figures), + "figure_inventory": figure_inventory, + "omitted_figure_inventory": omitted_figure_inventory, + "omitted_figure_count": len(omitted_figure_inventory), + "figure_categories": { + category: sum(1 for figure in enrichment_figures if figure.get("category") == category) + for category in sorted({figure.get("category", "") for figure in enrichment_figures}) + }, + }, + "provenance": {"source_manifests": source_records}, + "artifacts": { + "rnaseq_report": _artifact(workspace, report_path, "rnaseq_report"), + "rnaseq_report_evidence": _artifact(workspace, evidence_path, "rnaseq_report_evidence"), + "rnaseq_report_model": _artifact(workspace, model_path, "rnaseq_report_model"), + }, + } + _write_text_atomic(manifest_path, json.dumps(manifest, ensure_ascii=False, indent=2) + "\n") + return RNASeqReportResult( + report_path, + evidence_path, + model_path, + manifest_path, + coverage_status, + tuple(included), + resolved_scope, + ) diff --git a/tests/unit/test_rnaseq_report.py b/tests/unit/test_rnaseq_report.py new file mode 100644 index 000000000..aa3f2579b --- /dev/null +++ b/tests/unit/test_rnaseq_report.py @@ -0,0 +1,2131 @@ +from __future__ import annotations + +import json +import importlib.util +import re +from pathlib import Path + +import pytest + +import gfa_core.rnaseq_report as rnaseq_report_module +from gfa_core.rnaseq_report import NoSuccessfulRNASeqEvidence, generate_rnaseq_report +from gfa_core.mas.utils import intent_parser as intent_parser_module +from gfa_core.mas.utils.intent_parser import parse_intent_from_query + + +def _fast_intent_recognition(query: str) -> dict | None: + module_path = Path(__file__).resolve().parents[2] / "gfa_core" / "mas" / "core" / "fast_intent.py" + spec = importlib.util.spec_from_file_location("rnaseq_report_fast_intent", module_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.fast_intent_recognition(query) + + +def _write_manifest( + workspace: Path, + prefix: str, + stage: str, + status: str, + scope_id: str, + *, + suffix: str = "", +) -> Path: + final_dir = workspace / "final" + final_dir.mkdir(parents=True, exist_ok=True) + data_path = final_dir / f"{prefix}.{stage}{suffix}.table.tsv" + data_path.write_text("metric\tvalue\nreads\t100\n", encoding="utf-8") + manifest_path = final_dir / f"{prefix}.{stage}{suffix}.manifest.json" + manifest = { + "manifest_version": "1.0", + "stage": stage, + "status": status, + "prefix": prefix, + "rnaseq_scope_id": scope_id, + "metrics": {"sample_count": 2, "total_reads": 100}, + "parameters": {"rnaseq_scope_id": scope_id}, + "artifacts": { + "result_table": { + "artifact": "result_table", + "path": f"final/{data_path.name}", + "full_path": str(data_path), + "type": "table", + } + }, + } + manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8") + return manifest_path + + +def _write_artifact_manifest( + workspace: Path, + prefix: str, + stage: str, + scope_id: str, + artifacts: dict[str, tuple[str, str]], +) -> Path: + final_dir = workspace / "final" + final_dir.mkdir(parents=True, exist_ok=True) + manifest_artifacts = {} + for key, (filename, content) in artifacts.items(): + artifact_path = final_dir / filename + artifact_path.write_text(content, encoding="utf-8") + manifest_artifacts[key] = { + "artifact": key, + "path": f"final/{filename}", + "full_path": str(artifact_path), + "type": artifact_path.suffix.lstrip("."), + } + manifest_path = final_dir / f"{prefix}.{stage}.manifest.json" + manifest_path.write_text( + json.dumps( + { + "manifest_version": "1.0", + "stage": stage, + "status": "success", + "prefix": prefix, + "rnaseq_scope_id": scope_id, + "parameters": {"rnaseq_scope_id": scope_id}, + "metrics": {"comparison_count": 2}, + "artifacts": manifest_artifacts, + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + return manifest_path + + +def _figure_section(markdown: str, filename: str) -> str: + figure_area = markdown.index("## 4. 图表与产物说明") + image_at = markdown.index(filename, figure_area) + section_start = markdown.rfind("### 图 ", 0, image_at) + assert section_start >= 0, f"missing figure heading for {filename}" + next_figure = markdown.find("### 图 ", image_at) + section_end = next_figure if next_figure >= 0 else markdown.find("\n## 5.", image_at) + return markdown[section_start : section_end if section_end >= 0 else None] + + +def _embedded_image_names(markdown: str) -> list[str]: + return [Path(link).name for link in re.findall(r"!\[[^]]*]\(([^)]+)\)", markdown)] + + +def _write_enrichment_manifests(workspace: Path) -> None: + prefix = "Os" + scope = "scope-enrichment" + go_a = f"{prefix}.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich" + go_b = f"{prefix}.rnaseq_go_enrich.IR29__6_vs_IR29__0.GO_enrich" + go_table_header = "ID\tDescription\tOntology\tGeneRatio\tBgRatio\tpvalue\tp.adjust\tCount\tgeneID\n" + _write_artifact_manifest( + workspace, + prefix, + "rnaseq_go_enrich", + scope, + { + "go_a_table": (f"{go_a}.txt", go_table_header + "GO:0009651\tsalt stress response\tBP\t4/20\t20/1000\t0.0001\t0.001\t4\tOs01/Os02/Os03/Os04\n"), + "go_b_table": (f"{go_b}.txt", go_table_header + "GO:0015979\tphotosynthesis\tBP\t3/20\t18/1000\t0.01\t0.04\t3\tOs05/Os06/Os07\n"), + "go_a_bar_png": (f"{go_a}.barplot.png", "png"), + "go_a_bar_pdf": (f"{go_a}.barplot.pdf", "pdf"), + "go_a_bar_svg": (f"{go_a}.barplot.svg", ""), + "go_b_bar_png": (f"{go_b}.barplot.png", "png"), + "go_a_dot": (f"{go_a}.dotplot.png", "png"), + "go_b_dot": (f"{go_b}.dotplot.png", "png"), + "go_a_ontology": (f"{go_a}.ontology_barplot.png", "png"), + "go_b_ontology": (f"{go_b}.ontology_barplot.png", "png"), + }, + ) + + kegg_a = f"{prefix}.rnaseq_kegg_enrich.IR29__6_vs_IR29__0" + kegg_b = f"{prefix}.rnaseq_kegg_enrich.HH11__48_vs_HH11__0" + kegg_header = "ID\tDescription\tGeneRatio\tBgRatio\tpvalue\tp.adjust\tCount\tgeneID\n" + _write_artifact_manifest( + workspace, + prefix, + "rnaseq_kegg_enrich", + scope, + { + "kegg_a_table": ( + f"{kegg_a}.KEGG_enrich.txt", + kegg_header + + "ko00073\tCutin, suberine and wax biosynthesis\t2/20\t10/1000\t0.0001\t0.001\t2\tOsWax1/OsWax2\n" + + "ko00196\tPhotosynthesis - antenna proteins\t2/20\t12/1000\t0.0002\t0.002\t2\tOsLight1/OsLight2\n", + ), + "kegg_b_table": ( + f"{kegg_b}.KEGG_enrich.txt", + kegg_header + + "ko01200\tCarbon metabolism\t3/20\t30/1000\t0.01\t0.04\t3\tOsC1/OsC2/OsC3\n" + + "ko00073\tCutin, suberine and wax biosynthesis\t1/20\t10/1000\t0.02\t0.05\t1\tOsWaxH\n", + ), + "kegg_a_hits": ( + f"{kegg_a}.KEGG_pathway_gene_hits.txt", + "PathwayID\tPathwayName\tKO\tKOName\tGeneCount\tGenes\n" + "ko00073\tCutin, suberine and wax biosynthesis\tK00073\twax enzyme\t2\tOsWax1/OsWax2\n" + "ko00196\tPhotosynthesis - antenna proteins\tK00196\tantenna protein\t2\tOsLight1/OsLight2\n", + ), + "kegg_a_fc": ( + f"{kegg_a}.KEGG_pathway_map.ko_log2fc.tsv", + "KO\tlog2FoldChange\nK00073\t2.5\nK00196\t-1.5\n", + ), + "kegg_a_bar_png": (f"{kegg_a}_KEGG_enrich.png", "png"), + "kegg_a_bar_pdf": (f"{kegg_a}_KEGG_enrich.pdf", "pdf"), + "kegg_b_bar": (f"{kegg_b}_KEGG_enrich.png", "png"), + "kegg_a_dot": (f"{kegg_a}_KEGG_dotplot.png", "png"), + "kegg_b_dot": (f"{kegg_b}_KEGG_dotplot.png", "png"), + "pathway_a_00073": (f"{kegg_a}.KEGG_pathway_map.ko00073.png", "png"), + "pathway_a_00196": (f"{kegg_a}.KEGG_pathway_map.ko00196.png", "png"), + "pathway_b_00073": (f"{kegg_b}.KEGG_pathway_map.ko00073.png", "png"), + }, + ) + + fgsea_a = f"{prefix}.rnaseq_fgsea.IR29__6_vs_IR29__0" + fgsea_b = f"{prefix}.rnaseq_fgsea.HH11__48_vs_HH11__0" + fgsea_header = "pathway\tpathway_name\tpval\tpadj\tES\tNES\tsize\tleadingEdge\n" + _write_artifact_manifest( + workspace, + prefix, + "rnaseq_fgsea", + scope, + { + "fgsea_a_table": ( + f"{fgsea_a}.fgsea_kegg.tsv", + fgsea_header + + "ko00196\tPhotosynthesis - antenna proteins\t0.0001\t0.001\t0.7\t2.1\t18\tOsLight1,OsLight2\n", + ), + "fgsea_b_table": ( + f"{fgsea_b}.fgsea_kegg.tsv", + fgsea_header + + "ko01200\tCarbon metabolism\t0.01\t0.04\t-0.4\t-1.2\t30\tOsC1,OsC2\n", + ), + "fgsea_a_dot": (f"{fgsea_a}.fgsea_kegg_dotplot.png", "png"), + "fgsea_b_dot": (f"{fgsea_b}.fgsea_kegg_dotplot.png", "png"), + "fgsea_a_curve": (f"{fgsea_a}.fgsea_kegg_top_enrichment.png", "png"), + "fgsea_b_curve": (f"{fgsea_b}.fgsea_kegg_top_enrichment.png", "png"), + }, + ) + + +def _write_analysis_ready_family_manifest( + workspace: Path, + *, + prefix: str, + scope_id: str, + rows: list[tuple[str, str, str, str]], +) -> Path: + """Write the real analysis-ready TSV shape used by rnaseq_family_score.""" + header = "rank\tfamily_id\tfamily_name\tbest_comparison\tbest_support_labels\n" + body = "".join( + f"{rank}\t{family_id}\t{family_name}\t{comparison}\t{support_labels}\n" + for rank, (family_id, family_name, comparison, support_labels) in enumerate(rows, 1) + ) + return _write_artifact_manifest( + workspace, + prefix, + "rnaseq_family_score", + scope_id, + { + "analysis_ready_family": ( + f"{prefix}.rnaseq_family_score.analysis_ready_family.tsv", + header + body, + ) + }, + ) + + +def _write_candidate_guided_enrichment_manifests( + workspace: Path, + *, + combined: bool = False, + include_fgsea_support: bool = True, +) -> dict[str, str]: + """Create arbitrary identifiers so selection tests cannot rely on the sample job.""" + prefix = "Zm" + scope = "scope-random-candidate-guided" + candidate_comparison = "LineX__17_vs_LineX__0" + unrelated_comparison = "LineY__29_vs_LineY__0" + supported_pathway_a = "ko54321" + supported_pathway_b = "ko67890" + unrelated_pathway = "ko13579" + go_id = "GO:9876543" + enrich_stage = "rnaseq_enrich" if combined else "" + go_stage = enrich_stage or "rnaseq_go_enrich" + kegg_stage = enrich_stage or "rnaseq_kegg_enrich" + + go_candidate = f"{prefix}.{go_stage}.{candidate_comparison}.GO_enrich" + go_unrelated = f"{prefix}.{go_stage}.{unrelated_comparison}.GO_enrich" + go_header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + go_artifacts = { + "candidate_go_table": ( + f"{go_candidate}.txt", + go_header + + f"{go_id}\trandom candidate function\tBP\t3/20\t0.02\t3\tGeneA/GeneB/GeneC\n", + ), + "unrelated_go_table": ( + f"{go_unrelated}.txt", + go_header + + "GO:1111111\tunrelated stronger function\tBP\t8/20\t0.000001\t8\tOther1/Other2\n", + ), + "candidate_go_bar": (f"{go_candidate}.barplot.png", "png"), + "candidate_go_dot": (f"{go_candidate}.dotplot.png", "png"), + "candidate_go_ontology": (f"{go_candidate}.ontology_barplot.png", "png"), + "unrelated_go_bar": (f"{go_unrelated}.barplot.png", "png"), + "unrelated_go_dot": (f"{go_unrelated}.dotplot.png", "png"), + "unrelated_go_ontology": (f"{go_unrelated}.ontology_barplot.png", "png"), + } + + kegg_candidate = f"{prefix}.{kegg_stage}.{candidate_comparison}" + kegg_unrelated = f"{prefix}.{kegg_stage}.{unrelated_comparison}" + kegg_header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + kegg_artifacts = { + "candidate_kegg_table": ( + f"{kegg_candidate}.KEGG_enrich.txt", + kegg_header + + f"{supported_pathway_a}\trandom pathway alpha\t2/20\t0.02\t2\tGeneA/GeneB\n" + + f"{supported_pathway_b}\trandom pathway beta\t2/20\t0.03\t2\tGeneB/GeneC\n", + ), + "unrelated_kegg_table": ( + f"{kegg_unrelated}.KEGG_enrich.txt", + kegg_header + + f"{supported_pathway_a}\trandom pathway alpha\t7/20\t0.000001\t7\tOther1/Other2\n" + + f"{unrelated_pathway}\tunrelated pathway\t9/20\t0.0000001\t9\tOther3/Other4\n", + ), + "candidate_kegg_bar": (f"{kegg_candidate}_KEGG_enrich.png", "png"), + "candidate_kegg_dot": (f"{kegg_candidate}_KEGG_dotplot.png", "png"), + "unrelated_kegg_bar": (f"{kegg_unrelated}_KEGG_enrich.png", "png"), + "unrelated_kegg_dot": (f"{kegg_unrelated}_KEGG_dotplot.png", "png"), + "candidate_pathway_a": ( + f"{kegg_candidate}.KEGG_pathway_map.{supported_pathway_a}.png", + "png", + ), + "candidate_pathway_b": ( + f"{kegg_candidate}.KEGG_pathway_map.{supported_pathway_b}.png", + "png", + ), + "other_time_supported_pathway": ( + f"{kegg_unrelated}.KEGG_pathway_map.{supported_pathway_a}.png", + "png", + ), + "unrelated_pathway": ( + f"{kegg_unrelated}.KEGG_pathway_map.{unrelated_pathway}.png", + "png", + ), + } + + if combined: + _write_artifact_manifest( + workspace, + prefix, + "rnaseq_enrich", + scope, + {**go_artifacts, **kegg_artifacts}, + ) + else: + _write_artifact_manifest(workspace, prefix, go_stage, scope, go_artifacts) + _write_artifact_manifest(workspace, prefix, kegg_stage, scope, kegg_artifacts) + + fgsea_candidate = f"{prefix}.rnaseq_fgsea.{candidate_comparison}" + fgsea_unrelated = f"{prefix}.rnaseq_fgsea.{unrelated_comparison}" + fgsea_header = "pathway\tpathway_name\tpadj\tNES\tsize\tleadingEdge\n" + _write_artifact_manifest( + workspace, + prefix, + "rnaseq_fgsea", + scope, + { + "candidate_fgsea_table": ( + f"{fgsea_candidate}.fgsea_kegg.tsv", + fgsea_header + + f"{supported_pathway_b}\trandom pathway beta\t0.04\t2.1\t12\tGeneB,GeneC\n", + ), + "unrelated_fgsea_table": ( + f"{fgsea_unrelated}.fgsea_kegg.tsv", + fgsea_header + + f"{unrelated_pathway}\tunrelated pathway\t0.0001\t3.0\t20\tOther1,Other2\n", + ), + "candidate_fgsea_dot": (f"{fgsea_candidate}.fgsea_kegg_dotplot.png", "png"), + "candidate_fgsea_curve": ( + f"{fgsea_candidate}.fgsea_kegg_top_enrichment.png", + "png", + ), + "unrelated_fgsea_dot": (f"{fgsea_unrelated}.fgsea_kegg_dotplot.png", "png"), + "unrelated_fgsea_curve": ( + f"{fgsea_unrelated}.fgsea_kegg_top_enrichment.png", + "png", + ), + }, + ) + + support = ( + f"GO:{go_id}:random candidate function;" + f"KEGG:{supported_pathway_a}:random pathway alpha;" + f"KEGG:{supported_pathway_b}:random pathway beta" + ) + if include_fgsea_support: + support += f";fgsea:{supported_pathway_b}:random pathway beta" + _write_analysis_ready_family_manifest( + workspace, + prefix=prefix, + scope_id=scope, + rows=[ + ("FAM_RANDOM_ALPHA", "random alpha family", candidate_comparison, support), + ( + "FAM_RANDOM_BETA", + "random beta family", + candidate_comparison, + f"KEGG:{supported_pathway_a}:random pathway alpha", + ), + ], + ) + return { + "prefix": prefix, + "scope": scope, + "candidate_comparison": candidate_comparison, + "unrelated_comparison": unrelated_comparison, + "supported_pathway_a": supported_pathway_a, + "supported_pathway_b": supported_pathway_b, + "unrelated_pathway": unrelated_pathway, + "go_stage": go_stage, + "kegg_stage": kegg_stage, + } + +def test_report_is_deterministic_scope_bound_and_reusable(tmp_path: Path) -> None: + workspace = tmp_path / "job_scope" + _write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-alpha") + _write_manifest(workspace, "Os", "rnaseq_de", "success", "scope-beta") + _write_manifest(workspace, "Os", "rnaseq_report", "success", "scope-alpha", suffix=".old") + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-alpha", + expected_stages=["rnaseq_qc"], + use_model=False, + ) + + assert result.coverage_status == "complete" + assert result.included_stages == ("rnaseq_qc",) + assert result.rnaseq_scope_id == "scope-alpha" + markdown = result.report_path.read_text(encoding="utf-8") + assert "## 0. 报告摘要" in markdown + assert "## 7. 证据追溯" in markdown + assert "rnaseq_de" not in markdown + evidence = json.loads(result.evidence_path.read_text(encoding="utf-8")) + assert evidence["formal_report"]["source_fingerprint"] + assert {item["stage"] for item in evidence["formal_report"]["source_manifests"]} == {"rnaseq_qc"} + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + assert manifest["status"] == "success" + assert manifest["metrics"]["coverage_status"] == "complete" + assert manifest["metrics"]["model_status"] == "disabled" + + reopened = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-alpha", + expected_stages=["rnaseq_qc"], + use_model=False, + action="open", + ) + assert reopened.reused is True + + +def test_successful_report_can_have_partial_coverage(tmp_path: Path) -> None: + workspace = tmp_path / "job_partial" + _write_manifest(workspace, "Ft", "rnaseq_qc", "success", "scope-1") + _write_manifest(workspace, "Ft", "rnaseq_de", "failed", "scope-1") + + result = generate_rnaseq_report( + workspace, + "Ft", + rnaseq_scope_id="scope-1", + expected_stages=["rnaseq_qc", "rnaseq_de"], + use_model=False, + ) + + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + assert manifest["status"] == "success" + assert manifest["metrics"]["coverage_status"] == "partial" + assert manifest["metrics"]["stage_statuses"]["rnaseq_de"] == "failed" + assert result.included_stages == ("rnaseq_qc",) + + +def test_report_relocates_stale_absolute_artifact_paths(tmp_path: Path) -> None: + workspace = tmp_path / "relocated_job" + manifest_path = _write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-moved") + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["artifacts"]["result_table"]["full_path"] = "/old/server/workspace/final/result.tsv" + manifest_path.write_text(json.dumps(manifest, ensure_ascii=False), encoding="utf-8") + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-moved", + expected_stages=["rnaseq_qc"], + use_model=False, + ) + + markdown = result.report_path.read_text(encoding="utf-8") + assert "指标 sample_count" in markdown + assert "manifest 登记但文件不存在" not in markdown + evidence = json.loads(result.evidence_path.read_text(encoding="utf-8")) + result_card = next(card for card in evidence["evidence_cards"] if card["claim"] == "产物 result_table") + assert "exists=True" in result_card["evidence"] + + +def test_no_successful_rnaseq_stage_does_not_create_empty_report(tmp_path: Path) -> None: + workspace = tmp_path / "job_empty" + _write_manifest(workspace, "At", "rnaseq_de", "failed", "scope-empty") + + with pytest.raises(NoSuccessfulRNASeqEvidence, match="暂无可生成"): + generate_rnaseq_report( + workspace, + "At", + rnaseq_scope_id="scope-empty", + expected_stages=["rnaseq_de"], + use_model=False, + ) + + assert not (workspace / "final" / "At.rnaseq_report.md").exists() + + +def test_fallback_selects_one_of_each_enrichment_plot_and_deduplicates_pathway_ids( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_enrichment_figures" + _write_enrichment_manifests(workspace) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-enrichment", + expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + embedded = _embedded_image_names(markdown) + inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"][ + "figure_inventory" + ] + + assert sum(name.endswith(".GO_enrich.barplot.png") for name in embedded) == 1 + assert sum(name.endswith(".GO_enrich.dotplot.png") for name in embedded) == 1 + assert sum(name.endswith(".GO_enrich.ontology_barplot.png") for name in embedded) == 1 + assert sum(name.endswith("_KEGG_enrich.png") for name in embedded) == 1 + assert sum(name.endswith("_KEGG_dotplot.png") for name in embedded) == 1 + assert sum(name.endswith(".fgsea_kegg_dotplot.png") for name in embedded) == 1 + assert sum(name.endswith(".fgsea_kegg_top_enrichment.png") for name in embedded) == 1 + assert not any(name.endswith("GO_enrich.barplot.pdf") for name in embedded) + assert not any(name.endswith("GO_enrich.barplot.svg") for name in embedded) + assert not any(name.endswith("KEGG_enrich.pdf") for name in embedded) + + required_pathway_maps = [ + "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00073.png", + "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00196.png", + ] + for filename in required_pathway_maps: + assert embedded.count(filename) == 1 + assert not any( + name == "Os.rnaseq_kegg_enrich.HH11__48_vs_HH11__0.KEGG_pathway_map.ko00073.png" + for name in embedded + ) + + +def test_each_selected_enrichment_figure_has_four_part_table_bound_interpretation(tmp_path: Path) -> None: + workspace = tmp_path / "job_enrichment_explanations" + _write_enrichment_manifests(workspace) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-enrichment", + expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + expected_evidence = { + "Os.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich.barplot.png": "salt stress response", + "Os.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich.dotplot.png": "GO:0009651", + "Os.rnaseq_go_enrich.HH11__24_vs_HH11__0.GO_enrich.ontology_barplot.png": "BP", + "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0_KEGG_enrich.png": "Cutin, suberine and wax biosynthesis", + "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0_KEGG_dotplot.png": "ko00073", + "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00073.png": "OsWax1", + "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0.KEGG_pathway_map.ko00196.png": "OsLight1", + "Os.rnaseq_fgsea.IR29__6_vs_IR29__0.fgsea_kegg_dotplot.png": "NES", + "Os.rnaseq_fgsea.IR29__6_vs_IR29__0.fgsea_kegg_top_enrichment.png": "leadingEdge", + } + for filename, evidence_value in expected_evidence.items(): + section = _figure_section(markdown, filename) + assert "图形含义：" in section + assert "阅读方法：" in section + assert "本次结果：" in section + assert "解释边界：" in section + assert evidence_value in section + + +def test_combined_rnaseq_enrich_stage_keeps_all_plot_forms_and_deduplicates_pathway_ids( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_combined_enrichment" + go_a = "Os.rnaseq_enrich.HH11__24_vs_HH11__0.GO_enrich" + go_b = "Os.rnaseq_enrich.IR29__6_vs_IR29__0.GO_enrich" + kegg_a = "Os.rnaseq_enrich.IR29__6_vs_IR29__0" + kegg_b = "Os.rnaseq_enrich.HH11__48_vs_HH11__0" + go_header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + kegg_header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_enrich", + "scope-combined-enrichment", + { + "go_a_table": ( + f"{go_a}.txt", + go_header + + "GO:0009651\tsalt stress response\tBP\t4/20\t0.001\t4\tOsSalt1/OsSalt2\n" + + "GO:0009507\tchloroplast\tCC\t3/20\t0.002\t3\tOsChl1/OsChl2\n" + + "GO:0016491\toxidoreductase activity\tMF\t2/20\t0.003\t2\tOsOx1/OsOx2\n", + ), + "go_b_table": ( + f"{go_b}.txt", + go_header + "GO:0015979\tphotosynthesis\tBP\t3/20\t0.04\t3\tOsLight1/OsLight2\n", + ), + "go_a_bar_png": (f"{go_a}.barplot.png", "png"), + "go_a_bar_pdf": (f"{go_a}.barplot.pdf", "pdf"), + "go_b_bar": (f"{go_b}.barplot.png", "png"), + "go_a_dot": (f"{go_a}.dotplot.png", "png"), + "go_b_dot": (f"{go_b}.dotplot.png", "png"), + "go_a_ontology": (f"{go_a}.ontology_barplot.png", "png"), + "go_b_ontology": (f"{go_b}.ontology_barplot.png", "png"), + "kegg_a_table": ( + f"{kegg_a}.KEGG_enrich.txt", + kegg_header + + "ko00073\tCutin, suberine and wax biosynthesis\t2/20\t0.001\t2\tOsWax1/OsWax2\n" + + "ko00196\tPhotosynthesis - antenna proteins\t2/20\t0.002\t2\tOsLight1/OsLight2\n", + ), + "kegg_b_table": ( + f"{kegg_b}.KEGG_enrich.txt", + kegg_header + "ko00073\tCutin, suberine and wax biosynthesis\t1/20\t0.04\t1\tOsWaxH\n", + ), + "kegg_a_hits": ( + f"{kegg_a}.KEGG_pathway_gene_hits.txt", + "PathwayID\tPathwayName\tKO\tGenes\n" + "ko00073\tCutin, suberine and wax biosynthesis\tK00073\tOsWax1/OsWax2\n" + "ko00196\tPhotosynthesis - antenna proteins\tK00196\tOsLight1/OsLight2\n", + ), + "kegg_b_hits": ( + f"{kegg_b}.KEGG_pathway_gene_hits.txt", + "PathwayID\tPathwayName\tKO\tGenes\n" + "ko00073\tCutin, suberine and wax biosynthesis\tK10073\tOsWaxH\n", + ), + "kegg_a_bar_png": (f"{kegg_a}_KEGG_enrich.png", "png"), + "kegg_a_bar_pdf": (f"{kegg_a}_KEGG_enrich.pdf", "pdf"), + "kegg_b_bar": (f"{kegg_b}_KEGG_enrich.png", "png"), + "kegg_a_dot": (f"{kegg_a}_KEGG_dotplot.png", "png"), + "kegg_b_dot": (f"{kegg_b}_KEGG_dotplot.png", "png"), + "pathway_a_00073_png": (f"{kegg_a}.KEGG_pathway_map.ko00073.png", "png"), + "pathway_a_00073_pdf": (f"{kegg_a}.KEGG_pathway_map.ko00073.pdf", "pdf"), + "pathway_a_00196": (f"{kegg_a}.KEGG_pathway_map.ko00196.png", "png"), + "pathway_b_00073": (f"{kegg_b}.KEGG_pathway_map.ko00073.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-combined-enrichment", + expected_stages=["rnaseq_enrich"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + embedded = _embedded_image_names(markdown) + inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"][ + "figure_inventory" + ] + + assert sum(name.endswith(".GO_enrich.barplot.png") for name in embedded) == 1 + assert sum(name.endswith(".GO_enrich.dotplot.png") for name in embedded) == 1 + assert sum(name.endswith(".GO_enrich.ontology_barplot.png") for name in embedded) == 1 + assert sum(name.endswith("_KEGG_enrich.png") for name in embedded) == 1 + assert sum(name.endswith("_KEGG_dotplot.png") for name in embedded) == 1 + assert not any(name.endswith((".pdf", ".svg")) for name in embedded) + assert { + category: sum(item["category"] == category for item in inventory) + for category in ("go_bar", "go_dot", "go_ontology", "kegg_bar", "kegg_dot", "kegg_pathway") + } == { + "go_bar": 1, + "go_dot": 1, + "go_ontology": 1, + "kegg_bar": 1, + "kegg_dot": 1, + "kegg_pathway": 2, + } + assert not any(item["filename"].endswith(".pdf") for item in inventory) + + pathway_names = [name for name in embedded if ".KEGG_pathway_map.ko" in name] + assert pathway_names == sorted( + [ + f"{kegg_a}.KEGG_pathway_map.ko00073.png", + f"{kegg_a}.KEGG_pathway_map.ko00196.png", + ] + ) + expected_evidence = { + f"{go_a}.barplot.png": "salt stress response", + f"{go_a}.dotplot.png": "GO:0009651", + f"{go_a}.ontology_barplot.png": "MF=oxidoreductase activity", + f"{kegg_a}_KEGG_enrich.png": "Cutin, suberine and wax biosynthesis", + f"{kegg_a}_KEGG_dotplot.png": "ko00073", + f"{kegg_a}.KEGG_pathway_map.ko00073.png": "OsWax1", + f"{kegg_a}.KEGG_pathway_map.ko00196.png": "OsLight1", + } + for filename, evidence_value in expected_evidence.items(): + section = _figure_section(markdown, filename) + assert all( + label in section + for label in ("图形含义：", "阅读方法：", "本次结果：", "解释边界：") + ) + assert evidence_value in section + + +def test_enrichment_selection_does_not_remove_de_or_other_stage_figures(tmp_path: Path) -> None: + workspace = tmp_path / "job_non_enrichment_figures" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_de", + "scope-all-figures", + { + "de_volcano_a": ("Os.rnaseq_de.HH11__24_vs_HH11__0.volcano.png", "png"), + "de_volcano_b": ("Os.rnaseq_de.IR29__6_vs_IR29__0.volcano.png", "png"), + }, + ) + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_qc", + "scope-all-figures", + { + "qc_before": ("Os.rnaseq_qc.before.png", "png"), + "qc_after": ("Os.rnaseq_qc.after.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-all-figures", + expected_stages=["rnaseq_qc", "rnaseq_de"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + embedded = _embedded_image_names(markdown) + + for filename in ( + "Os.rnaseq_de.HH11__24_vs_HH11__0.volcano.png", + "Os.rnaseq_de.IR29__6_vs_IR29__0.volcano.png", + "Os.rnaseq_qc.before.png", + "Os.rnaseq_qc.after.png", + ): + assert embedded.count(filename) == 1 + + +def test_open_regenerates_v3_report_with_renderer_v4(tmp_path: Path) -> None: + workspace = tmp_path / "job_legacy_report" + _write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-renderer") + generated = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-renderer", + expected_stages=["rnaseq_qc"], + use_model=False, + ) + generated.report_path.write_text("STALE LEGACY REPORT\n", encoding="utf-8") + old_manifest = json.loads(generated.manifest_path.read_text(encoding="utf-8")) + old_manifest["metrics"]["report_renderer_version"] = "3" + generated.manifest_path.write_text(json.dumps(old_manifest, ensure_ascii=False), encoding="utf-8") + + reopened = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-renderer", + expected_stages=["rnaseq_qc"], + use_model=False, + action="open", + ) + + assert reopened.reused is False + assert "STALE LEGACY REPORT" not in reopened.report_path.read_text(encoding="utf-8") + refreshed_manifest = json.loads(reopened.manifest_path.read_text(encoding="utf-8")) + assert refreshed_manifest["metrics"]["report_renderer_version"] == "4" + + +def test_pathway_without_matching_kegg_row_reports_insufficient_evidence(tmp_path: Path) -> None: + workspace = tmp_path / "job_pathway_mismatch" + base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-pathway-mismatch", + { + "kegg_table": ( + f"{base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "ko00073\tCutin, suberine and wax biosynthesis\t2/20\t0.001\t2\tOsWax1/OsWax2\n", + ), + "unmatched_pathway": (f"{base}.KEGG_pathway_map.ko99999.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-pathway-mismatch", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko99999.png") + + assert "证据不足" in section + assert "Cutin, suberine and wax biosynthesis" not in section + + +def test_pathway_without_ko_mapping_does_not_summarize_all_ko_fold_changes(tmp_path: Path) -> None: + workspace = tmp_path / "job_pathway_without_ko_mapping" + base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-no-ko-map", + { + "kegg_table": ( + f"{base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "ko00073\tCutin, suberine and wax biosynthesis\t2/20\t0.001\t2\tOsWax1/OsWax2\n", + ), + "pathway_hits_without_ko": ( + f"{base}.KEGG_pathway_gene_hits.txt", + "PathwayID\tPathwayName\tGenes\n" + "ko00073\tCutin, suberine and wax biosynthesis\tOsWax1/OsWax2\n", + ), + "unscoped_ko_fc": ( + f"{base}.KEGG_pathway_map.ko_log2fc.tsv", + "KO\tlog2FoldChange\nK11111\t2.5\nK22222\t-1.5\n", + ), + "pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-no-ko-map", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png") + + assert "KO对应关系证据不足" in section + assert "KO变化中上调1个、下调1个" not in section + + +def test_kegg_representative_selection_ignores_raw_enrichment_tables(tmp_path: Path) -> None: + workspace = tmp_path / "job_kegg_raw_selection" + header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + a = "Os.rnaseq_kegg_enrich.AA__1_vs_AA__0" + b = "Os.rnaseq_kegg_enrich.BB__1_vs_BB__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-kegg-raw-selection", + { + "a_filtered": (f"{a}.KEGG_enrich.txt", header + "ko00001\tFiltered A\t1/20\t0.05\t1\tOsA\n"), + "a_raw": (f"{a}.KEGG_enrich.raw.txt", header + "ko00001\tRaw A\t1/20\t0.000000001\t1\tOsA\n"), + "a_bar": (f"{a}_KEGG_enrich.png", "png"), + "b_filtered": (f"{b}.KEGG_enrich.txt", header + "ko00002\tFiltered B\t2/20\t0.01\t2\tOsB1/OsB2\n"), + "b_raw": (f"{b}.KEGG_enrich.raw.txt", header + "ko00002\tRaw B\t2/20\t0.9\t2\tOsB1/OsB2\n"), + "b_bar": (f"{b}_KEGG_enrich.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-kegg-raw-selection", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + embedded = _embedded_image_names(result.report_path.read_text(encoding="utf-8")) + + assert f"{b}_KEGG_enrich.png" in embedded + assert f"{a}_KEGG_enrich.png" not in embedded + + +def test_kegg_figure_interpretation_uses_filtered_table_not_raw_table(tmp_path: Path) -> None: + workspace = tmp_path / "job_kegg_raw_interpretation" + base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0" + header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-kegg-raw-interpretation", + { + "raw_first": (f"{base}.KEGG_enrich.raw.txt", header + "ko99998\tRaw-only pathway\t9/20\t0.0001\t9\tRawGene\n"), + "filtered": (f"{base}.KEGG_enrich.txt", header + "ko00073\tFiltered pathway\t2/20\t0.01\t2\tOsWax1/OsWax2\n"), + "bar": (f"{base}_KEGG_enrich.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-kegg-raw-interpretation", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), f"{base}_KEGG_enrich.png") + + assert "Filtered pathway" in section + assert "Raw-only pathway" not in section + + +def test_go_ontology_interpretation_covers_bp_cc_and_mf_rows(tmp_path: Path) -> None: + workspace = tmp_path / "job_go_three_ontologies" + base = "Os.rnaseq_go_enrich.MixedCase__6_vs_MixedCase__0.GO_enrich" + header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_go_enrich", + "scope-go-ontologies", + { + "table": ( + f"{base}.txt", + header + + "GO:0000001\tbiological process theme\tBP\t3/20\t0.001\t3\tA/B/C\n" + + "GO:0000002\tcellular component theme\tCC\t2/20\t0.002\t2\tD/E\n" + + "GO:0000003\tmolecular function theme\tMF\t1/20\t0.003\t1\tF\n", + ), + "ontology": (f"{base}.ontology_barplot.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-go-ontologies", + expected_stages=["rnaseq_go_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), "ontology_barplot.png") + + assert "biological process theme" in section + assert "cellular component theme" in section + assert "molecular function theme" in section + + +def test_svg_fallbacks_are_classified_and_selected_once_per_plot_form(tmp_path: Path) -> None: + workspace = tmp_path / "job_svg_fallbacks" + go_header = "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + kegg_header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_go_enrich", + "scope-svg-fallback", + { + "go_a_table": ("Os.rnaseq_go_enrich.AA.GO_enrich.txt", go_header + "GO:1\tA\tBP\t1/2\t0.01\t1\tA\n"), + "go_b_table": ("Os.rnaseq_go_enrich.BB.GO_enrich.txt", go_header + "GO:2\tB\tBP\t1/2\t0.02\t1\tB\n"), + "go_a_svg": ("Os.rnaseq_go_enrich.AA.GO_enrich.svg", ""), + "go_b_svg": ("Os.rnaseq_go_enrich.BB.GO_enrich.svg", ""), + }, + ) + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-svg-fallback", + { + "kegg_a_table": ("Os.rnaseq_kegg_enrich.AA.KEGG_enrich.txt", kegg_header + "ko1\tA\t1/2\t0.01\t1\tA\n"), + "kegg_b_table": ("Os.rnaseq_kegg_enrich.BB.KEGG_enrich.txt", kegg_header + "ko2\tB\t1/2\t0.02\t1\tB\n"), + "kegg_a_svg": ("Os.rnaseq_kegg_enrich.AA.KEGG_enrich.svg", ""), + "kegg_b_svg": ("Os.rnaseq_kegg_enrich.BB.KEGG_enrich.svg", ""), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-svg-fallback", + expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich"], + use_model=False, + ) + evidence = json.loads(result.evidence_path.read_text(encoding="utf-8")) + categories = [item["category"] for item in evidence["formal_report"]["figure_inventory"]] + + assert categories.count("go_bar") == 1 + assert categories.count("kegg_bar") == 1 + + +def test_comparison_case_is_preserved_in_figure_interpretation(tmp_path: Path) -> None: + workspace = tmp_path / "job_comparison_case" + base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-case", + { + "table": ( + f"{base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\nko00073\tWax\t2/20\t0.01\t2\tA/B\n", + ), + "bar": (f"{base}_KEGG_enrich.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-case", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), f"{base}_KEGG_enrich.png") + + assert "本次结果：比较组 `IR29__6_vs_IR29__0`" in section + + +def test_invalid_adjusted_p_values_do_not_participate_in_representative_selection(tmp_path: Path) -> None: + workspace = tmp_path / "job_invalid_padj" + header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + artifacts: dict[str, tuple[str, str]] = {} + for label, padj in (("NEG", "-0.1"), ("NAN", "NaN"), ("INF", "Inf"), ("VALID", "0.02")): + base = f"Os.rnaseq_kegg_enrich.{label}__1_vs_{label}__0" + artifacts[f"{label}_table"] = (f"{base}.KEGG_enrich.txt", header + f"ko00001\t{label}\t1/20\t{padj}\t1\tGene\n") + artifacts[f"{label}_bar"] = (f"{base}_KEGG_enrich.png", "png") + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-invalid-padj", + artifacts, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-invalid-padj", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + embedded = _embedded_image_names(result.report_path.read_text(encoding="utf-8")) + + assert "Os.rnaseq_kegg_enrich.VALID__1_vs_VALID__0_KEGG_enrich.png" in embedded + + +def test_fgsea_curve_tied_padj_uses_largest_absolute_nes_and_matching_pathway(tmp_path: Path) -> None: + workspace = tmp_path / "job_fgsea_tie" + header = "pathway\tpathway_name\tpval\tpadj\tES\tNES\tsize\tleadingEdge\n" + aa = "Os.rnaseq_fgsea.AA__1_vs_AA__0" + zz = "Os.rnaseq_fgsea.ZZ__1_vs_ZZ__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_fgsea", + "scope-fgsea-tie", + { + "aa_table": (f"{aa}.fgsea_kegg.tsv", header + "ko00001\tAA pathway\t0.001\t0.01\t0.4\t2.0\t10\tA,B\n"), + "aa_curve": (f"{aa}.fgsea_kegg_top_enrichment.png", "png"), + "zz_table": ( + f"{zz}.fgsea_kegg.tsv", + header + + "ko00002\tZZ weaker tied pathway\t0.001\t0.01\t0.2\t1.0\t10\tC,D\n" + + "ko00003\tZZ strongest tied pathway\t0.001\t0.01\t-0.7\t-3.0\t12\tE,F\n", + ), + "zz_curve": (f"{zz}.fgsea_kegg_top_enrichment.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-fgsea-tie", + expected_stages=["rnaseq_fgsea"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + embedded = _embedded_image_names(markdown) + + assert f"{zz}.fgsea_kegg_top_enrichment.png" in embedded + section = _figure_section(markdown, f"{zz}.fgsea_kegg_top_enrichment.png") + assert "ko00003" in section + assert "NES=-3.0" in section + + +def test_pdf_only_pathway_is_linked_or_recorded_in_figure_inventory(tmp_path: Path) -> None: + workspace = tmp_path / "job_pdf_pathway" + base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0" + pdf_name = f"{base}.KEGG_pathway_map.ko00073.pdf" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-pdf-pathway", + { + "table": ( + f"{base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\nko00073\tWax\t2/20\t0.01\t2\tA/B\n", + ), + "pdf_pathway": (pdf_name, "pdf"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-pdf-pathway", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + evidence = json.loads(result.evidence_path.read_text(encoding="utf-8")) + inventory_names = {item["filename"] for item in evidence["formal_report"]["figure_inventory"]} + linked = re.search(rf"\[[^]]*\]\([^)]*{re.escape(pdf_name)}[^)]*\)", markdown) is not None + + assert linked or pdf_name in inventory_names + + +def test_renderer_and_figure_inventory_are_consistent_in_evidence_and_manifest(tmp_path: Path) -> None: + workspace = tmp_path / "job_report_metadata" + _write_enrichment_manifests(workspace) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-enrichment", + expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"], + use_model=False, + ) + evidence = json.loads(result.evidence_path.read_text(encoding="utf-8")) + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + formal = evidence["formal_report"] + metrics = manifest["metrics"] + + assert formal["report_renderer_version"] == metrics["report_renderer_version"] + assert formal["figure_inventory"] + assert metrics.get("figure_inventory") == formal["figure_inventory"] + + +def test_go_ontology_interpretation_accepts_real_long_ontology_names(tmp_path: Path) -> None: + workspace = tmp_path / "job_go_real_ontology_names" + base = "Os.rnaseq_go_enrich.A__1_vs_A__0.GO_enrich" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_go_enrich", + "scope-go-real-ontology", + { + "table": ( + f"{base}.txt", + "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "GO:0000001\tlong-name BP theme\tBiological process\t3/20\t0.001\t3\tA/B/C\n" + "GO:0000002\tlong-name CC theme\tCellular component\t2/20\t0.002\t2\tD/E\n" + "GO:0000003\tlong-name MF theme\tMolecular function\t1/20\t0.003\t1\tF\n", + ), + "ontology": (f"{base}.ontology_barplot.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-go-real-ontology", + expected_stages=["rnaseq_go_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), "ontology_barplot.png") + + assert "BP=long-name BP theme" in section + assert "CC=long-name CC theme" in section + assert "MF=long-name MF theme" in section + + +def test_pathway_interpretation_merges_genes_from_all_matching_ko_rows(tmp_path: Path) -> None: + workspace = tmp_path / "job_pathway_multi_ko_hits" + base = "Os.rnaseq_kegg_enrich.A__1_vs_A__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-pathway-multi-ko", + { + "enrichment": ( + f"{base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "ko00073\tWax pathway\t3/20\t0.001\t3\tGeneA/GeneB/GeneC\n", + ), + "hits": ( + f"{base}.KEGG_pathway_gene_hits.txt", + "PathwayID\tPathwayName\tKO\tGenes\n" + "ko00073\tWax pathway\tK00001\tGeneA/GeneB\n" + "ko00073\tWax pathway\tK00002\tGeneC\n", + ), + "pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-pathway-multi-ko", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png") + + assert "GeneA" in section + assert "GeneB" in section + assert "GeneC" in section + + +def test_figure_uses_exact_comparison_table_when_names_contain_each_other(tmp_path: Path) -> None: + workspace = tmp_path / "job_exact_comparison" + base_a = "Os.rnaseq_go_enrich.A.GO_enrich" + base_aa = "Os.rnaseq_go_enrich.AA.GO_enrich" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_go_enrich", + "scope-exact-comparison", + { + "aa_table": ( + f"{base_aa}.txt", + "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "GO:0000002\tAA-only evidence\tBP\t2/20\t0.0001\t2\tAA1/AA2\n", + ), + "a_table": ( + f"{base_a}.txt", + "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "GO:0000001\tA exact evidence\tBP\t1/20\t0.01\t1\tA1\n", + ), + "a_bar": (f"{base_a}.barplot.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-exact-comparison", + expected_stages=["rnaseq_go_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), f"{base_a}.barplot.png") + + assert "A exact evidence" in section + assert "AA-only evidence" not in section + + +@pytest.mark.parametrize("invalid_padj", ["NaN", "Inf", "-0.1"]) +def test_mandatory_pathway_does_not_render_invalid_padj_as_significance( + tmp_path: Path, invalid_padj: str +) -> None: + workspace = tmp_path / f"job_invalid_pathway_padj_{invalid_padj.replace('-', 'neg')}" + base = "Os.rnaseq_kegg_enrich.A__1_vs_A__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-invalid-pathway-padj", + { + "table": ( + f"{base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + f"ko00073\tWax pathway\t2/20\t{invalid_padj}\t2\tA/B\n", + ), + "pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-invalid-pathway-padj", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png") + + assert "p.adjust=未登记" in section + assert f"p.adjust={invalid_padj}" not in section + + +def test_pdf_only_non_pathway_enrichment_forms_are_linked_and_in_inventory(tmp_path: Path) -> None: + workspace = tmp_path / "job_pdf_only_enrichment_forms" + go_base = "Os.rnaseq_go_enrich.A__1_vs_A__0.GO_enrich" + kegg_base = "Os.rnaseq_kegg_enrich.A__1_vs_A__0" + fgsea_base = "Os.rnaseq_fgsea.A__1_vs_A__0" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_go_enrich", + "scope-pdf-forms", + { + "table": ( + f"{go_base}.txt", + "ID\tDescription\tOntology\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "GO:1\tGO evidence\tBP\t1/20\t0.01\t1\tA\n", + ), + "bar": (f"{go_base}.barplot.pdf", "pdf"), + "dot": (f"{go_base}.dotplot.pdf", "pdf"), + "ontology": (f"{go_base}.ontology_barplot.pdf", "pdf"), + }, + ) + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-pdf-forms", + { + "table": ( + f"{kegg_base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + "ko00001\tKEGG evidence\t1/20\t0.01\t1\tA\n", + ), + "bar": (f"{kegg_base}_KEGG_enrich.pdf", "pdf"), + "dot": (f"{kegg_base}_KEGG_dotplot.pdf", "pdf"), + }, + ) + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_fgsea", + "scope-pdf-forms", + { + "table": ( + f"{fgsea_base}.fgsea_kegg.tsv", + "pathway\tpathway_name\tpadj\tNES\tleadingEdge\n" + "ko00001\tFGSEA evidence\t0.01\t2.0\tA,B\n", + ), + "dot": (f"{fgsea_base}.fgsea_kegg_dotplot.pdf", "pdf"), + "curve": (f"{fgsea_base}.fgsea_kegg_top_enrichment.pdf", "pdf"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-pdf-forms", + expected_stages=["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + evidence = json.loads(result.evidence_path.read_text(encoding="utf-8")) + expected = { + f"{go_base}.barplot.pdf", + f"{go_base}.dotplot.pdf", + f"{go_base}.ontology_barplot.pdf", + f"{kegg_base}_KEGG_enrich.pdf", + f"{kegg_base}_KEGG_dotplot.pdf", + f"{fgsea_base}.fgsea_kegg_dotplot.pdf", + f"{fgsea_base}.fgsea_kegg_top_enrichment.pdf", + } + inventory_names = {item["filename"] for item in evidence["formal_report"]["figure_inventory"]} + + assert expected <= inventory_names + for filename in expected: + assert re.search(rf"\[[^]]*\]\([^)]*{re.escape(filename)}[^)]*\)", markdown) + + +def test_representative_selection_requires_legal_nonempty_evidence_before_format( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_representative_evidence_priority" + header = "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + artifacts: dict[str, tuple[str, str]] = {} + cases = ( + ("EMPTY", "", "png"), + ("INVALID", header + "ko1\tinvalid\t1/20\t-0.1\t1\tA\n", "png"), + ("VALID_SVG", header + "ko2\tvalid svg\t1/20\t0.04\t1\tB\n", "svg"), + ("VALID_PNG", header + "ko3\tvalid png\t1/20\t0.03\t1\tC\n", "png"), + ) + for label, table_content, extension in cases: + base = f"Os.rnaseq_kegg_enrich.{label}" + artifacts[f"{label}_table"] = (f"{base}.KEGG_enrich.txt", table_content) + artifacts[f"{label}_figure"] = (f"{base}_KEGG_enrich.{extension}", "figure") + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-representative-priority", + artifacts, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-representative-priority", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"]["figure_inventory"] + selected = [item["filename"] for item in inventory if item["category"] == "kegg_bar"] + + assert selected == ["Os.rnaseq_kegg_enrich.VALID_PNG_KEGG_enrich.png"] + + +def test_model_enhancement_rejects_numbers_and_activation_claims_absent_from_evidence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from gfa_core.mas import llm as llm_module + from gfa_core.mas.config import config + + class FakeManager: + def __init__(self, _config: object) -> None: + pass + + def call(self, *_args: object, **_kwargs: object) -> object: + return type( + "Response", + (), + { + "content": ( + "模型声称42个基因显著变化，因此该通路被激活并直接导致抗逆性增强；" + "同时还声称处理后产量提高了37%，这些数字与因果结论均不在输入证据中。" + ), + "provider": "fake", + "model": "fake-model", + }, + )() + + monkeypatch.setattr(config, "load_llm_config", lambda: {}) + monkeypatch.setattr(llm_module, "LLMManager", FakeManager) + evidence = { + "metadata": {"included_stages": ["rnaseq_kegg_enrich"]}, + "biological_story": "检测到KEGG富集结果，具体证据见结果表。", + "key_findings": [], + "major_claims": [], + "risk_flags": [], + } + + summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True) + + assert summary == "" + assert metadata["status"] == "fallback" + assert metadata["error"] + + +def _install_fake_model_response(monkeypatch: pytest.MonkeyPatch, content: str) -> None: + from gfa_core.mas import llm as llm_module + from gfa_core.mas.config import config + + class FakeManager: + def __init__(self, _config: object) -> None: + pass + + def call(self, *_args: object, **_kwargs: object) -> object: + return type( + "Response", + (), + {"content": content, "provider": "fake", "model": "fake-model"}, + )() + + monkeypatch.setattr(config, "load_llm_config", lambda: {}) + monkeypatch.setattr(llm_module, "LLMManager", FakeManager) + + +def test_free_form_model_output_is_rejected_and_never_embedded_in_markdown( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = tmp_path / "job_model_not_embedded" + _write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-model-not-embedded") + model_text = ( + "现有证据仅支持对已完成阶段进行描述性总结，各项结果仍需结合实验设计和原始结果表审慎解释，" + "不应扩展为未被数据支持的生物学结论。" + ) + _install_fake_model_response(monkeypatch, model_text) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-model-not-embedded", + expected_stages=["rnaseq_qc"], + use_model=True, + ) + + markdown = result.report_path.read_text(encoding="utf-8") + model_record = json.loads(result.model_path.read_text(encoding="utf-8")) + assert model_text not in markdown + assert "### 模型辅助语言润色（不作为新增证据）" in markdown + assert "API 仅参与已有证据语句的排序和组织" in markdown + assert model_record["content"] == model_text + assert model_record["status"] == "fallback" + assert model_record["policy"] == "evidence_claim_ordering_only" + assert model_record["error"] + + +def test_noncausal_free_form_model_text_is_still_rejected( + monkeypatch: pytest.MonkeyPatch, +) -> None: + model_text = ( + "这些富集结果只能说明统计关联，不能证明因果关系，也无法认定相关通路被激活，" + "仍需结合独立实验进行验证。" + ) + _install_fake_model_response(monkeypatch, model_text) + evidence = { + "metadata": {"included_stages": ["rnaseq_kegg_enrich"]}, + "biological_story": "检测到KEGG富集结果，具体证据见结果表。", + "key_findings": [], + "major_claims": [], + "risk_flags": [], + } + + summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True) + + assert summary == "" + assert metadata["content"] == model_text + assert metadata["status"] == "fallback" + assert metadata["policy"] == "evidence_claim_ordering_only" + assert metadata["error"] + + +def test_model_cannot_embed_mechanistic_claims_even_when_all_numbers_exist_in_evidence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + model_text = ( + "已有结果中的12个基因显著上调并促进耐盐，揭示了该通路控制抗逆性的调控机制，" + "因此可作为后续育种的确定依据。" + ) + _install_fake_model_response(monkeypatch, model_text) + evidence = { + "metadata": {"included_stages": ["rnaseq_kegg_enrich"]}, + "biological_story": "结果表登记12个命中基因。", + "key_findings": [], + "major_claims": [], + "risk_flags": [], + } + + summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True) + + assert summary == "" + assert metadata["content"] == model_text + assert metadata["status"] == "fallback" + assert metadata["policy"] == "evidence_claim_ordering_only" + + +def test_model_can_only_order_and_group_approved_evidence_claims( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evidence = { + "metadata": {"included_stages": ["rnaseq_qc"]}, + "biological_story": "质控阶段已经完成，具体质量指标见结果表。", + "key_findings": [ + {"stage": "rnaseq_qc", "description": "成功阶段提供了可追溯的质控结果。"} + ], + "major_claims": [], + "risk_flags": [], + } + organization = { + "paragraph_groups": [["finding_1", "story_1"]], + } + _install_fake_model_response(monkeypatch, json.dumps(organization, ensure_ascii=False)) + + summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True) + + assert summary == "成功阶段提供了可追溯的质控结果。质控阶段已经完成，具体质量指标见结果表。" + assert metadata["status"] == "embedded_evidence_only" + assert metadata["policy"] == "evidence_claim_ordering_only" + assert metadata["organization_plan"] == organization + assert metadata["selected_claim_ids"] == ["finding_1", "story_1"] + assert {row["claim_id"] for row in metadata["claim_catalog"]} == {"story_1", "finding_1"} + + +def test_unknown_model_claim_id_forces_deterministic_fallback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evidence = { + "metadata": {"included_stages": ["rnaseq_qc"]}, + "biological_story": "质控阶段已经完成。", + "key_findings": [], + "major_claims": [], + "risk_flags": [], + } + _install_fake_model_response( + monkeypatch, + json.dumps({"paragraph_groups": [["story_1", "invented_claim"]]}, ensure_ascii=False), + ) + + summary, metadata = rnaseq_report_module._model_enhancement(evidence, enabled=True) + + assert summary == "" + assert metadata["status"] == "fallback" + assert "invented_claim" in metadata["error"] + + +def test_report_always_states_the_api_evidence_boundary(tmp_path: Path) -> None: + workspace = tmp_path / "job_model_boundary" + _write_manifest(workspace, "Os", "rnaseq_qc", "success", "scope-model-boundary") + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-model-boundary", + expected_stages=["rnaseq_qc"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + + assert "### 模型辅助语言润色（不作为新增证据）" in markdown + assert "API 仅参与已有证据语句的排序和组织" in markdown + assert "不参与结果分析或推断" in markdown + assert "未采用模型组织结果" in markdown + + +def test_pathway_interpretation_previews_only_twelve_genes_and_kos_and_links_full_tables( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_pathway_long_lists" + base = "Os.rnaseq_kegg_enrich.IR29__6_vs_IR29__0" + genes = [f"Gene{index:03d}" for index in range(1, 456)] + kos = [f"K{index:05d}" for index in range(1, 456)] + hit_rows = "".join( + f"ko00073\tWax pathway\t{ko}\t{gene}\n" for ko, gene in zip(kos, genes) + ) + ko_rows = "".join(f"{ko}\t1.0\n" for ko in kos) + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_kegg_enrich", + "scope-pathway-long-lists", + { + "filtered": ( + f"{base}.KEGG_enrich.txt", + "ID\tDescription\tGeneRatio\tp.adjust\tCount\tgeneID\n" + f"ko00073\tWax pathway\t455/1000\t0.001\t455\t{'/'.join(genes)}\n", + ), + "hits": ( + f"{base}.KEGG_pathway_gene_hits.txt", + "PathwayID\tPathwayName\tKO\tGenes\n" + hit_rows, + ), + "ko_log2fc": ( + f"{base}.KEGG_pathway_map.ko_log2fc.tsv", + "KO\tlog2FoldChange\n" + ko_rows, + ), + "pathway": (f"{base}.KEGG_pathway_map.ko00073.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-pathway-long-lists", + expected_stages=["rnaseq_kegg_enrich"], + use_model=False, + ) + section = _figure_section(result.report_path.read_text(encoding="utf-8"), "KEGG_pathway_map.ko00073.png") + + assert "命中基因总数=455" in section + assert "KO总数=455" in section + assert all(gene in section for gene in genes[:12]) + assert all(ko in section for ko in kos[:12]) + assert genes[12] not in section + assert kos[12] not in section + assert f"[{base}.KEGG_pathway_gene_hits.txt]" in section + assert f"[{base}.KEGG_pathway_map.ko_log2fc.tsv]" in section + + +def test_fgsea_leading_edge_previews_twelve_genes_and_links_full_result_table( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_fgsea_long_leading_edge" + base = "Os.rnaseq_fgsea.IR29__6_vs_IR29__0" + leading_edge = [f"LeadGene{index:03d}" for index in range(1, 31)] + table_name = f"{base}.fgsea_kegg.tsv" + _write_artifact_manifest( + workspace, + "Os", + "rnaseq_fgsea", + "scope-fgsea-long-leading-edge", + { + "fgsea_table": ( + table_name, + "pathway\tpathway_name\tpadj\tNES\tsize\tleadingEdge\n" + f"ko00196\tPhotosynthesis - antenna proteins\t0.001\t2.1\t30\t{','.join(leading_edge)}\n", + ), + "fgsea_curve": (f"{base}.fgsea_kegg_top_enrichment.png", "png"), + }, + ) + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-fgsea-long-leading-edge", + expected_stages=["rnaseq_fgsea"], + use_model=False, + ) + markdown = result.report_path.read_text(encoding="utf-8") + section = _figure_section(markdown, f"{base}.fgsea_kegg_top_enrichment.png") + + assert "leadingEdge总数=30" in section + assert "前12项=" in section + assert all(gene in section for gene in leading_edge[:12]) + assert all(gene not in section for gene in leading_edge[12:]) + assert f"[{table_name}](" in section + assert ( + "> 报告生成原则：结果判断只来自脚本、Manifest 和结果表；API 仅参与已有证据语句的排序和组织，" + "不参与结果分析或推断。" in markdown + ) + + +def test_candidate_guided_pathways_keep_all_supported_contents_and_omit_unrelated( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_candidate_guided_pathways" + ids = _write_candidate_guided_enrichment_manifests(workspace) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=[ + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_fgsea", + "rnaseq_family_score", + ], + use_model=False, + ) + embedded = _embedded_image_names(result.report_path.read_text(encoding="utf-8")) + evidence = json.loads(result.evidence_path.read_text(encoding="utf-8")) + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + formal = evidence["formal_report"] + pathway_rows = [ + row for row in formal["figure_inventory"] if row["category"] == "kegg_pathway" + ] + + assert formal["selection_mode"] == "candidate_guided" + assert {row["pathway_id"] for row in pathway_rows} == { + ids["supported_pathway_a"], + ids["supported_pathway_b"], + } + assert all(row["comparison"] == ids["candidate_comparison"] for row in pathway_rows) + assert not any(ids["unrelated_pathway"] in name for name in embedded) + assert not any( + ids["unrelated_comparison"] in name and ids["supported_pathway_a"] in name + for name in embedded + ) + + shared = next( + row for row in pathway_rows if row["pathway_id"] == ids["supported_pathway_a"] + ) + assert shared["candidate_families"] == ["FAM_RANDOM_ALPHA", "FAM_RANDOM_BETA"] + assert any( + label.startswith(f"KEGG:{ids['supported_pathway_a']}:") + for label in shared["support_labels"] + ) + assert shared["selection_reason"] == "candidate_family_support" + assert sum(ids["supported_pathway_a"] in name for name in embedded) == 1 + + omitted = formal["omitted_figure_inventory"] + assert any(row["pathway_id"] == ids["unrelated_pathway"] for row in omitted) + assert formal["omitted_figure_count"] == len(omitted) + assert manifest["metrics"]["selection_mode"] == formal["selection_mode"] + assert manifest["metrics"]["omitted_figure_count"] == formal["omitted_figure_count"] + + +def test_candidate_guided_representatives_use_candidate_comparison_and_supported_fgsea( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_candidate_guided_forms" + ids = _write_candidate_guided_enrichment_manifests(workspace) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=[ + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_fgsea", + "rnaseq_family_score", + ], + use_model=False, + ) + inventory = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"][ + "figure_inventory" + ] + selected_categories = { + category: [row for row in inventory if row["category"] == category] + for category in ( + "go_bar", + "go_dot", + "go_ontology", + "kegg_bar", + "kegg_dot", + "fgsea_dot", + "fgsea_curve", + ) + } + + assert all(len(rows) == 1 for rows in selected_categories.values()) + assert all( + rows[0]["comparison"] == ids["candidate_comparison"] + for rows in selected_categories.values() + ) + assert all(rows[0]["candidate_families"] for rows in selected_categories.values()) + assert all( + rows[0]["selection_reason"] == "candidate_family_support" + for rows in selected_categories.values() + ) + fgsea_curve = selected_categories["fgsea_curve"][0] + assert fgsea_curve["candidate_families"] == ["FAM_RANDOM_ALPHA"] + assert fgsea_curve["support_labels"] == [ + f"fgsea:{ids['supported_pathway_b']}:random pathway beta" + ] + + +def test_fgsea_curve_requires_exact_candidate_support_for_its_actual_top_pathway( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_fgsea_curve_exact_identity" + ids = _write_candidate_guided_enrichment_manifests(workspace) + fgsea_table = ( + workspace + / "final" + / f"{ids['prefix']}.rnaseq_fgsea.{ids['candidate_comparison']}.fgsea_kegg.tsv" + ) + fgsea_table.write_text( + "pathway\tpathway_name\tpadj\tNES\tsize\tleadingEdge\n" + "ko99999\tnon-candidate top pathway\t0.000001\t3.5\t20\tOtherA,OtherB\n" + f"{ids['supported_pathway_b']}\trandom pathway beta\t0.04\t2.1\t12\tGeneB,GeneC\n", + encoding="utf-8", + ) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=[ + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_fgsea", + "rnaseq_family_score", + ], + use_model=False, + ) + formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"] + selected = formal["figure_inventory"] + omitted = formal["omitted_figure_inventory"] + fgsea_dots = [row for row in selected if row["category"] == "fgsea_dot"] + fgsea_curves = [row for row in selected if row["category"] == "fgsea_curve"] + + assert len(fgsea_dots) == 1 + assert fgsea_dots[0]["comparison"] == ids["candidate_comparison"] + assert fgsea_dots[0]["candidate_families"] == ["FAM_RANDOM_ALPHA"] + assert fgsea_curves == [] + assert any(row["category"] == "fgsea_curve" for row in omitted) + + +def test_fgsea_is_excluded_from_candidate_mainline_without_fgsea_support(tmp_path: Path) -> None: + workspace = tmp_path / "job_candidate_without_fgsea" + ids = _write_candidate_guided_enrichment_manifests( + workspace, + include_fgsea_support=False, + ) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=[ + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_fgsea", + "rnaseq_family_score", + ], + use_model=False, + ) + formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"] + categories = {row["category"] for row in formal["figure_inventory"]} + omitted_categories = {row["category"] for row in formal["omitted_figure_inventory"]} + + assert "fgsea_dot" not in categories + assert "fgsea_curve" not in categories + assert {"fgsea_dot", "fgsea_curve"}.issubset(omitted_categories) + + +@pytest.mark.parametrize("analysis_ready_state", ["missing", "empty"]) +def test_missing_or_empty_analysis_ready_table_uses_non_candidate_fallback( + tmp_path: Path, + analysis_ready_state: str, +) -> None: + workspace = tmp_path / f"job_fallback_{analysis_ready_state}" + _write_enrichment_manifests(workspace) + expected_stages = ["rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_fgsea"] + if analysis_ready_state == "empty": + _write_analysis_ready_family_manifest( + workspace, + prefix="Os", + scope_id="scope-enrichment", + rows=[], + ) + expected_stages.append("rnaseq_family_score") + + result = generate_rnaseq_report( + workspace, + "Os", + rnaseq_scope_id="scope-enrichment", + expected_stages=expected_stages, + use_model=False, + ) + formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"] + inventory = formal["figure_inventory"] + categories = [row["category"] for row in inventory] + pathway_rows = [row for row in inventory if row["category"] == "kegg_pathway"] + + assert formal["selection_mode"] == "representative_fallback" + assert { + category: categories.count(category) + for category in ("go_bar", "go_dot", "go_ontology", "kegg_bar", "kegg_dot") + } == { + "go_bar": 1, + "go_dot": 1, + "go_ontology": 1, + "kegg_bar": 1, + "kegg_dot": 1, + } + assert {(row["comparison"], row["pathway_id"]) for row in pathway_rows} == { + ("IR29__6_vs_IR29__0", "ko00073"), + ("IR29__6_vs_IR29__0", "ko00196"), + } + assert all(row["candidate_families"] == [] for row in inventory) + assert all(row["support_labels"] == [] for row in inventory) + assert all( + row["selection_reason"] == "representative_fallback" + for row in inventory + if row["category"] + ) + assert "候选家族驱动筛选不可用" in result.report_path.read_text(encoding="utf-8") + + +def test_candidate_guided_selection_supports_combined_rnaseq_enrich_stage(tmp_path: Path) -> None: + workspace = tmp_path / "job_candidate_combined" + ids = _write_candidate_guided_enrichment_manifests(workspace, combined=True) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=["rnaseq_enrich", "rnaseq_fgsea", "rnaseq_family_score"], + use_model=False, + ) + formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"] + enrichment_rows = [row for row in formal["figure_inventory"] if row["category"]] + + assert formal["selection_mode"] == "candidate_guided" + assert {row["category"] for row in enrichment_rows} == { + "go_bar", + "go_dot", + "go_ontology", + "kegg_bar", + "kegg_dot", + "kegg_pathway", + "fgsea_dot", + "fgsea_curve", + } + assert all(row["comparison"] == ids["candidate_comparison"] for row in enrichment_rows) + assert { + row["pathway_id"] + for row in enrichment_rows + if row["category"] == "kegg_pathway" + } == {ids["supported_pathway_a"], ids["supported_pathway_b"]} + + +@pytest.mark.parametrize( + "support_labels", + ["", "KEGG:ko99998:pathway absent from all selected comparison outputs"], + ids=["empty-support-labels", "unmatched-support-labels"], +) +def test_analysis_ready_rows_without_matchable_support_use_representative_fallback( + tmp_path: Path, + support_labels: str, +) -> None: + workspace = tmp_path / "job_analysis_ready_without_matchable_support" + ids = _write_candidate_guided_enrichment_manifests(workspace) + _write_analysis_ready_family_manifest( + workspace, + prefix=ids["prefix"], + scope_id=ids["scope"], + rows=[ + ( + "FAM_WITHOUT_MATCHABLE_EVIDENCE", + "family without usable evidence", + ids["candidate_comparison"], + support_labels, + ) + ], + ) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=[ + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_fgsea", + "rnaseq_family_score", + ], + use_model=False, + ) + formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"] + + assert formal["selection_mode"] == "representative_fallback" + assert all(row["candidate_families"] == [] for row in formal["figure_inventory"]) + markdown = result.report_path.read_text(encoding="utf-8") + assert "候选家族驱动筛选不可用" in markdown + assert "candidate_guided" not in markdown + + +def test_unmatched_fgsea_only_support_uses_fallback_and_does_not_mark_dot_candidate( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_unmatched_fgsea_only" + ids = _write_candidate_guided_enrichment_manifests(workspace) + _write_analysis_ready_family_manifest( + workspace, + prefix=ids["prefix"], + scope_id=ids["scope"], + rows=[ + ( + "FAM_FGSEA_PATHWAY_ABSENT", + "family with absent fgsea pathway", + ids["candidate_comparison"], + "fgsea:ko99998:pathway absent from comparison fgsea table", + ) + ], + ) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=[ + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_fgsea", + "rnaseq_family_score", + ], + use_model=False, + ) + formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"] + fgsea_dots = [ + row for row in formal["figure_inventory"] if row["category"] == "fgsea_dot" + ] + + assert formal["selection_mode"] == "representative_fallback" + assert len(fgsea_dots) == 1 + assert fgsea_dots[0]["candidate_families"] == [] + assert fgsea_dots[0]["support_labels"] == [] + assert fgsea_dots[0]["selection_reason"] == "representative_fallback" + + +def test_candidate_mode_omits_fgsea_dot_when_only_fgsea_label_is_unmatched( + tmp_path: Path, +) -> None: + workspace = tmp_path / "job_go_candidate_with_unmatched_fgsea" + ids = _write_candidate_guided_enrichment_manifests(workspace) + _write_analysis_ready_family_manifest( + workspace, + prefix=ids["prefix"], + scope_id=ids["scope"], + rows=[ + ( + "FAM_GO_ONLY_USABLE", + "family with usable go only", + ids["candidate_comparison"], + "GO:GO:9876543:random candidate function;" + "fgsea:ko99998:pathway absent from comparison fgsea table", + ) + ], + ) + + result = generate_rnaseq_report( + workspace, + ids["prefix"], + rnaseq_scope_id=ids["scope"], + expected_stages=[ + "rnaseq_go_enrich", + "rnaseq_kegg_enrich", + "rnaseq_fgsea", + "rnaseq_family_score", + ], + use_model=False, + ) + formal = json.loads(result.evidence_path.read_text(encoding="utf-8"))["formal_report"] + selected = formal["figure_inventory"] + omitted = formal["omitted_figure_inventory"] + + assert formal["selection_mode"] == "candidate_guided" + assert any( + row["category"] == "go_bar" + and row["candidate_families"] == ["FAM_GO_ONLY_USABLE"] + for row in selected + ) + assert not any(row["category"] == "fgsea_dot" for row in selected) + assert any( + row["category"] == "fgsea_dot" + and row["comparison"] == ids["candidate_comparison"] + for row in omitted + ) + + +@pytest.mark.parametrize( + ("query", "action"), + [ + ("生成rnaseq结果报告", "generate"), + ("生成 RNA-seq 结果报告", "generate"), + ("打开rnaseq结果报告", "open"), + ("生成转录组结果报告", "generate"), + ], +) +def test_explicit_rnaseq_report_intent_is_provider_independent(query: str, action: str) -> None: + parsed = parse_intent_from_query(query) + fast = _fast_intent_recognition(query) + + assert parsed["stages"] == ["rnaseq_report"] + assert parsed["params"]["rnaseqReportAction"] == action + assert fast is not None + assert fast["stages"] == ["rnaseq_report"] + assert fast["params"]["rnaseqReportAction"] == action + + +def test_generic_report_without_rnaseq_context_is_not_stolen() -> None: + parsed = parse_intent_from_query("生成结果报告") + assert parsed["stages"] != ["rnaseq_report"] + + +def test_explicit_report_chat_intent_does_not_require_router_model() -> None: + assert hasattr(intent_parser_module, "deterministic_agent_chat_intent") + parser = intent_parser_module.deterministic_agent_chat_intent + + opened = parser("打开rnaseq结果报告") + generated = parser("生成 RNA-seq 结果报告") + + assert opened["stages"] == ["rnaseq_report"] + assert opened["params"]["rnaseqReportAction"] == "open" + assert generated["params"]["rnaseqReportAction"] == "generate" + assert parser("解释一下差异表达结果") is None + + +def test_generic_report_chat_intent_requires_rnaseq_job_context() -> None: + parser = intent_parser_module.deterministic_agent_chat_intent + + assert parser("生成结果报告") is None + generated = parser("生成结果报告", rnaseq_context=True) + opened = parser("打开结果报告", rnaseq_context=True) + + assert generated["stages"] == ["rnaseq_report"] + assert generated["params"]["rnaseqReportAction"] == "generate" + assert opened["params"]["rnaseqReportAction"] == "open"