diff --git a/gfa_core/rnaseq_report.py b/gfa_core/rnaseq_report.py
index e7f42adf0..273de49ec 100644
--- a/gfa_core/rnaseq_report.py
+++ b/gfa_core/rnaseq_report.py
@@ -7,17 +7,21 @@ job's ``final/`` directory.
"""
from __future__ import annotations
+import base64
import hashlib
import ast
import csv
+import html
import json
import math
+import mimetypes
import os
import re
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable
+from urllib.parse import unquote
from gfa_core.rnaseq_evidence import (
RNASEQ_STAGE_NAMES,
@@ -29,7 +33,7 @@ from gfa_core.rnaseq_evidence import (
REPORT_STAGE = "rnaseq_report"
-REPORT_RENDERER_VERSION = "4"
+REPORT_RENDERER_VERSION = "17"
SUCCESS_STATUSES = {"success", "partial_success"}
DEFAULT_EXPECTED_STAGES = [
"rnaseq_qc",
@@ -642,26 +646,84 @@ def _ordered_field_items(row: dict[str, str], *keys: str) -> list[str]:
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}"
+ if not ordered:
+ return ""
+ preview = "/".join(ordered[:5])
+ return f"{label}总数={len(ordered)}，前5项={preview}"
def _source_table_links(report_path: Path, tables: list[tuple[Path, list[dict[str, str]]]], markers: tuple[str, ...]) -> str:
+ matched = [path for path, _ in tables if any(marker in path.name.lower() for marker in markers)]
links = [
- f"[{path.name}]({_relative_link(report_path, path)})"
- for path, _ in tables
- if any(marker in path.name.lower() for marker in markers)
+ f"[下载完整数据{' ' + str(index) if len(matched) > 1 else ''}]({_relative_link(report_path, report_path.parent / path.name)})"
+ for index, path in enumerate(matched, start=1)
]
- return "；完整源表：" + "、".join(links) if links else ""
+ 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", "")
+ comparison = _figure_comparison(figure)
if not category:
- return [f"该图来自 `{figure['stage']}` 阶段。具体趋势应与该阶段的数值表和 manifest 指标联合判断。"]
- comparison = figure.get("comparison", "")
+ figure_type = _figure_display_type(figure)
+ meanings = {
+ "pca": "PCA图概括样本整体表达谱的主要差异及样本间距离。",
+ "sample_clustering": "样本聚类图按整体表达相似性展示样本之间的层次关系。",
+ "sample_correlation": "样本相关性热图展示任意两个样本表达谱的一致程度。",
+ "sample_distance": "样本距离热图展示样本之间整体表达差异的大小。",
+ "volcano": "火山图同时展示差异基因的变化幅度和统计显著性。",
+ "ma_plot": "MA图展示平均表达量与表达变化幅度之间的关系。",
+ "heatmap": "热图展示代表基因在不同样本中的相对表达模式。",
+ "upset": "UpSet图展示多个比较组差异基因集合的交集关系。",
+ "venn": "Venn图展示不同比较组差异基因集合的共有和特有部分。",
+ "module_trait": "模块性状相关图展示共表达模块与样本性状之间的相关性。",
+ "network": "网络图展示候选基因或共表达模块成员之间的连接关系。",
+ "tree": "聚类树展示基因或样本按相似性形成的层次结构。",
+ "boxplot": "箱线图展示不同样本或分组中数值的分布范围。",
+ "density": "密度图展示各样本表达量的整体分布形态。",
+ "other": "结果图展示该分析阶段的主要数据分布或分组关系。",
+ }
+ readings = {
+ "pca": "查看同组重复是否聚集、不同组是否分离，并留意明显偏离的样本。",
+ "sample_clustering": "查看同组样本是否优先聚为一支，并关注聚类距离较远的重复。",
+ "sample_correlation": "结合颜色图例读取相关系数，颜色越接近高值表示样本越相似。",
+ "sample_distance": "结合颜色图例比较样本距离，重点查看组内重复是否小于组间距离。",
+ "volcano": "横轴表示log2FoldChange，纵轴表示显著性，结合颜色区分上调、下调和不显著基因。",
+ "ma_plot": "横轴表示平均表达水平，纵轴表示log2FoldChange，重点查看显著基因的变化方向。",
+ "heatmap": "按颜色图例读取相对高低表达，并比较样本聚类与分组是否一致。",
+ "upset": "先看各集合大小，再读取连接点对应的交集柱高。",
+ "venn": "读取重叠区域和非重叠区域的基因数量。",
+ "module_trait": "结合颜色和相关系数读取正负相关，并同时关注显著性标记。",
+ "network": "结合节点、连线和图例查看关键节点及连接较紧密的区域。",
+ "tree": "分支越近表示相似性越高，重点查看分组或模块划分位置。",
+ "boxplot": "比较中位数、四分位范围和离群点，并查看组间分布差异。",
+ "density": "比较曲线峰值和整体形状，检查样本分布是否大体一致。",
+ "other": "结合图题、坐标轴、图例和分组标签读取主要结果。",
+ }
+ metrics = manifests.get(str(figure.get("stage") or ""), {}).get("metrics") or {}
+ if figure_type in {"volcano", "ma_plot"}:
+ comparison_metrics = (metrics.get("comparisons") or {}).get(comparison, {})
+ up_count = _value(comparison_metrics, "up")
+ down_count = _value(comparison_metrics, "down")
+ total_count = _value(comparison_metrics, "total")
+ result = (
+ f"比较组 `{comparison or '未识别'}` 的差异基因分布见图；"
+ f"上调基因数={up_count}，下调基因数={down_count}，"
+ f"显著差异基因总数={total_count}。"
+ )
+ elif figure_type in {"pca", "sample_clustering", "sample_correlation", "sample_distance"}:
+ result = "本次样本的组内重复关系和组间差异见图，具体以图中的样本标签、距离或颜色数值为准。"
+ elif comparison:
+ result = f"比较组 `{comparison}` 的主要结果见图，具体数值以坐标轴、图例和标注为准。"
+ else:
+ result = "本次分析的主要结果见图，具体数值以坐标轴、图例和标注为准。"
+ return [
+ f"图形含义：{meanings.get(figure_type, meanings['other'])}",
+ f"阅读方法：{readings.get(figure_type, readings['other'])}",
+ f"本次结果：{result}",
+ ]
tables = _supporting_tables(workspace, manifests[figure["stage"]], comparison, category)
pathway_id = figure.get("pathway_id", "")
row = (
@@ -704,11 +766,15 @@ def _figure_interpretation(
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"))))
+ details = [
+ f"GeneRatio={_value(row, 'GeneRatio')}",
+ f"Count={_value(row, 'Count')}",
+ f"p.adjust={_probability_value(row, 'p.adjust', 'FDR')}",
+ ]
+ details.extend(summary for summary in (_item_summary('命中基因', genes), _item_summary('KO', kos)) if summary)
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"{'，'.join(details)}"
f"{_source_table_links(report_path, tables, ('.kegg_enrich.txt',))}。"
)
elif category == "kegg_pathway":
@@ -746,20 +812,26 @@ def _figure_interpretation(
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表达方向"
+ details = [f"p.adjust={_probability_value(row, 'p.adjust', 'padj', 'FDR')}"]
+ details.extend(
+ summary for summary in (_item_summary('命中基因', pathway_genes), _item_summary('KO', pathway_kos)) if summary
+ )
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"{'，'.join(details)}{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 "未登记"
+ details = [
+ f"NES={_value(row, 'NES')}",
+ f"padj={_probability_value(row, 'padj', 'p.adjust')}",
+ ]
+ if leading_edge:
+ details.append(f"leadingEdge总数={len(leading_edge)}，前5项={'/'.join(leading_edge[:5])}")
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"{'，'.join(details)}"
f"{_source_table_links(report_path, tables, ('.fgsea_kegg.tsv',))}。"
)
@@ -792,7 +864,6 @@ def _figure_interpretation(
f"图形含义：{meanings[category]}",
f"阅读方法：{readings[category]}",
f"本次结果：{result}",
- f"解释边界：{boundary}",
]
@@ -803,6 +874,223 @@ def _relative_link(report_path: Path, target: Path) -> str:
return str(target)
+def _table_cell(value: Any) -> str:
+ return str(value if value is not None else "").replace("|", "/").replace("\r", " ").replace("\n", " ").strip()
+
+
+def _display_metrics(manifest: dict[str, Any], max_metrics: int = 12) -> list[tuple[str, str]]:
+ labels = {
+ "total_samples": "样本数",
+ "sample_count": "样本数",
+ "total_raw_reads": "原始 reads 总数",
+ "total_clean_reads": "过滤后 reads 总数",
+ "tested_genes": "参与检验的基因数",
+ "significant_degs": "显著差异基因数",
+ "up_genes": "上调基因数",
+ "down_genes": "下调基因数",
+ }
+ hidden_terms = ("fingerprint", "inventory", "renderer", "source_manifest", "request_", "omitted_")
+ rows: list[tuple[str, str]] = []
+ for key, value in (manifest.get("metrics") or {}).items():
+ lowered = str(key).lower()
+ if any(term in lowered for term in hidden_terms):
+ continue
+ if value is None or isinstance(value, (dict, list, tuple, set)):
+ continue
+ if isinstance(value, bool):
+ display = "是" if value else "否"
+ elif isinstance(value, int):
+ display = f"{value:,}"
+ elif isinstance(value, float):
+ if not math.isfinite(value):
+ continue
+ display = f"{value:,.4g}"
+ else:
+ display = str(value).strip()
+ if not display or len(display) > 160 or "samples={" in display.lower():
+ continue
+ label = labels.get(lowered, str(key).replace("_", " "))
+ rows.append((_table_cell(label), _table_cell(display)))
+ if len(rows) >= max_metrics:
+ break
+ return rows
+
+
+def _preferred_web_figure(path: Path) -> Path:
+ for suffix in (".png", ".webp", ".jpg", ".jpeg", ".svg"):
+ candidate = path.with_suffix(suffix)
+ if candidate.is_file():
+ return candidate
+ return path
+
+
+def _sample_metric_table(manifest: dict[str, Any], stage: str, max_columns: int = 8) -> dict[str, Any] | None:
+ samples = (manifest.get("metrics") or {}).get("samples")
+ if not isinstance(samples, dict) or not samples:
+ return None
+ available: set[str] = set()
+ for values in samples.values():
+ if isinstance(values, dict):
+ available.update(
+ str(key) for key, value in values.items()
+ if value is None or isinstance(value, (str, int, float, bool))
+ )
+ priorities = {
+ "rnaseq_qc": ("raw_reads", "clean_reads", "q20", "q30", "gc", "adapter", "low_quality", "duplication"),
+ "rnaseq_align": ("total_reads", "alignment_rate", "overall_alignment", "unique", "multiple", "unmapped"),
+ "rnaseq_quant": ("assigned_reads", "total_counted_reads", "assigned_percent", "detected_genes", "mapped_reads"),
+ }.get(stage, ())
+ columns: list[str] = []
+ for token in priorities:
+ for key in sorted(available):
+ if key not in columns and token in key.lower():
+ columns.append(key)
+ columns.extend(key for key in sorted(available) if key not in columns)
+ columns = columns[:max_columns]
+ if not columns:
+ return None
+ rows = []
+ for sample_id, values in sorted(samples.items()):
+ values = values if isinstance(values, dict) else {}
+ rows.append([_table_cell(sample_id), *[_table_cell(values.get(column, "")) for column in columns]])
+ return {
+ "columns": ["sample_id", *[column.replace("_", " ") for column in columns]],
+ "rows": rows,
+ }
+
+
+def _figure_display_type(figure: dict[str, Any]) -> str:
+ category = str(figure.get("category") or "").strip().lower()
+ if category:
+ return category
+ name = f"{figure.get('artifact', '')} {Path(str(figure.get('path') or '')).stem}".lower()
+ types = (
+ ("sample_correlation", ("correlation",)),
+ ("sample_distance", ("sample_distance", "distance_heatmap")),
+ ("sample_clustering", ("sample_clustering", "cluster")),
+ ("pca", ("pca",)),
+ ("volcano", ("volcano",)),
+ ("ma_plot", ("ma_plot", "maplot")),
+ ("heatmap", ("heatmap",)),
+ ("upset", ("upset",)),
+ ("venn", ("venn",)),
+ ("boxplot", ("boxplot",)),
+ ("density", ("density",)),
+ ("module_trait", ("module_trait",)),
+ ("network", ("network",)),
+ ("sample_tree", ("sample_tree",)),
+ ("gene_dendrogram", ("gene_dendrogram", "dendrogram_modules")),
+ ("tree", ("dendrogram", "tree")),
+ )
+ for figure_type, tokens in types:
+ if any(token in name for token in tokens):
+ return figure_type
+ return "other"
+
+
+def _figure_comparison(figure: dict[str, Any]) -> str:
+ explicit = str(figure.get("comparison") or "").strip()
+ if explicit:
+ return explicit
+ name = Path(str(figure.get("path") or "")).stem
+ lowered = name.lower()
+ marker = "rnaseq_de."
+ start = lowered.find(marker)
+ value = name[start + len(marker):] if start >= 0 else name
+ for suffix in ("_volcano_plot", ".volcano", "_ma_plot", ".ma_plot", ".top_de", "_heatmap"):
+ index = value.lower().find(suffix)
+ if index > 0:
+ return value[:index]
+ return ""
+
+
+def _select_report_figures(figures: list[dict[str, Any]]) -> list[dict[str, Any]]:
+ table_only_stages = {"rnaseq_qc", "rnaseq_align", "rnaseq_quant"}
+ de_groups: dict[str, set[str]] = {}
+ for figure in figures:
+ if figure.get("stage") != "rnaseq_de":
+ continue
+ comparison = _figure_comparison(figure)
+ if comparison:
+ de_groups.setdefault(comparison, set()).add(_figure_display_type(figure))
+ selected_comparison = ""
+ if de_groups:
+ selected_comparison = sorted(de_groups, key=lambda value: (-len(de_groups[value]), value))[0]
+
+ selected: dict[tuple[str, str], dict[str, Any]] = {}
+ order: list[tuple[str, str]] = []
+ format_rank = {".png": 0, ".webp": 1, ".jpg": 2, ".jpeg": 2, ".svg": 3, ".pdf": 4}
+ for figure in figures:
+ stage = str(figure.get("stage") or "")
+ if stage in table_only_stages:
+ continue
+ if stage == "rnaseq_de" and selected_comparison and _figure_comparison(figure) != selected_comparison:
+ continue
+ key = (stage, _figure_display_type(figure))
+ current = selected.get(key)
+ if current is None:
+ selected[key] = figure
+ order.append(key)
+ continue
+ current_rank = format_rank.get(Path(str(current.get("path") or "")).suffix.lower(), 9)
+ new_rank = format_rank.get(Path(str(figure.get("path") or "")).suffix.lower(), 9)
+ if new_rank < current_rank:
+ selected[key] = figure
+ return [selected[key] for key in order]
+
+
+def _stage_table_previews(
+ workspace: Path,
+ manifest: dict[str, Any],
+ report_path: Path,
+ *,
+ max_tables: int = 2,
+ max_rows: int = 10,
+ max_columns: int = 8,
+) -> list[dict[str, Any]]:
+ candidates: list[tuple[int, str, Path]] = []
+ priority_terms = ("summary", "candidate", "top", "metric", "stat", "result", "deg", "enrich")
+ for name, item in (manifest.get("artifacts") or {}).items():
+ if not isinstance(item, dict):
+ continue
+ path = _artifact_path(workspace, manifest, item)
+ if path is None or path.suffix.lower() not in {".csv", ".tsv"}:
+ continue
+ lowered = f"{name} {path.name}".lower()
+ score = sum(1 for term in priority_terms if term in lowered)
+ candidates.append((-score, str(name), path))
+ previews: list[dict[str, Any]] = []
+ for _score, name, path in sorted(candidates, key=lambda item: (item[0], item[2].name)):
+ delimiter = "," if path.suffix.lower() == ".csv" else "\t"
+ try:
+ with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
+ reader = csv.DictReader(handle, delimiter=delimiter)
+ columns = [str(column) for column in (reader.fieldnames or [])[:max_columns] if str(column).strip()]
+ if not columns:
+ continue
+ rows = []
+ for row in reader:
+ rows.append([_table_cell(row.get(column, "")) for column in columns])
+ if len(rows) >= max_rows:
+ break
+ except (OSError, csv.Error):
+ continue
+ if not rows:
+ continue
+ previews.append(
+ {
+ "name": name,
+ "filename": path.name,
+ "columns": columns,
+ "rows": rows,
+ "link": _relative_link(report_path, path),
+ }
+ )
+ if len(previews) >= max_tables:
+ break
+ return previews
+
+
def _render_deterministic_markdown(
*,
prefix: str,
@@ -826,147 +1114,367 @@ def _render_deterministic_markdown(
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
- )
+ figures = _select_report_figures(figures)
+ figure_index = 0
+ rendered_figures: set[str] = set()
- 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 "现有证据不足以形成跨阶段生物学叙事，请查看分阶段结果。"),
+ section_specs = [
+ ("## 3. 测序质量检查", ["rnaseq_qc"], "本节检查测序数据是否稳定、是否存在需要留意的样本。"),
+ ("## 4. 参考基因组比对", ["rnaseq_align"], "比对率表示测序片段能在参考基因组中找到位置的比例。"),
+ ("## 5. 基因表达定量", ["rnaseq_quant", "expression"], "表达定量用于统计每个基因在各样本中的表达水平。"),
+ ("## 6. 样本关系与重复可靠性", ["rnaseq_sample_qc"], "这一部分判断同组重复是否相似，以及是否存在明显离群样本。"),
+ ("## 7. 差异表达", ["rnaseq_de"], "上调和下调均相对于对应比较中的对照组，表示表达量升高或降低。"),
+ (
+ "## 8. GO、KEGG 与全基因排序分析",
+ ["annotation", "annotation_go", "annotation_kegg", "rnaseq_go_enrich", "rnaseq_kegg_enrich", "rnaseq_enrich", "rnaseq_fgsea"],
+ "本节汇总 GO、KEGG 和全基因排序分析中的主要功能条目及对应图表。",
+ ),
+ ("## 9. 共表达网络", ["rnaseq_wgcna"], "共表达网络把变化趋势相近的基因归为模块，用于缩小候选范围。"),
+ (
+ "## 10. 候选基因与候选家族",
+ ["rnaseq_candidate_family", "rnaseq_family_score"],
+ "本节展示候选基因、候选家族及其排序依据。",
+ ),
]
- 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(
- [
- "> 候选家族驱动筛选不可用；当前富集图按图形类型与统计证据选取代表，不声称与候选家族相关。",
- "",
- ]
+ def _stage_name(stage: str) -> str:
+ return STAGE_NAMES.get(stage, stage)
+
+ def _status_text(stage: str) -> str:
+ status = str((flow_rows.get(stage) or {}).get("status") or manifests.get(stage, {}).get("status") or "unknown")
+ return {"success": "已完成", "partial_success": "部分完成", "failed": "失败", "missing": "未完成"}.get(status, status)
+
+ def _render_figure(figure: dict[str, Any]) -> None:
+ nonlocal figure_index
+ figure_index += 1
+ original_path = Path(figure["path"])
+ path = report_path.parent / _preferred_web_figure(original_path).name
+ rendered_figures.add(str(original_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)})"
)
- 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"### 图 {figure_index}. {figure['artifact']}", "", media_line, ""])
+ lines.extend([*_figure_interpretation(figure, workspace, manifests, report_path), ""])
+
+ def _render_stage_group(stage_ids: list[str], empty_text: str) -> None:
+ selected = [stage for stage in included if stage in stage_ids]
+ if not selected:
+ lines.extend(["", empty_text])
+ return
+ table_stages = {"rnaseq_qc", "rnaseq_align", "rnaseq_quant"}
+ for stage in selected:
+ show_tables = stage in table_stages
+ metrics = _display_metrics(manifests.get(stage, {})) if show_tables else []
+ sample_table = _sample_metric_table(manifests.get(stage, {}), stage) if show_tables else None
+ lines.extend(["", f"### {_stage_name(stage)}", "", f"状态：{_status_text(stage)}。"])
+ if sample_table:
+ lines.extend(
+ [
+ "",
+ "逐样本结果：",
+ "",
+ "| " + " | ".join(_table_cell(column) for column in sample_table["columns"]) + " |",
+ "|" + "|".join("---" for _ in sample_table["columns"]) + "|",
+ ]
+ )
+ for row in sample_table["rows"]:
+ lines.append("| " + " | ".join(row) + " |")
+ if metrics:
+ lines.extend(["", "本次数据：", "", "| 指标 | 结果 |", "|---|---|"])
+ for label, value in metrics:
+ lines.append(f"| {label} | {value} |")
+ previews = (
+ _stage_table_previews(
+ workspace,
+ manifests.get(stage, {}),
+ report_path,
+ max_tables=3,
+ max_rows=30,
+ max_columns=10,
+ )
+ if stage in {"rnaseq_candidate_family", "rnaseq_family_score"}
+ else []
)
- 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 未登记可嵌入图像；请以分阶段表格和日志为准。")
+ for preview in previews:
+ lines.extend(
+ [
+ "",
+ f"#### 数据表：{preview['filename']}",
+ "",
+ "| " + " | ".join(_table_cell(column) for column in preview["columns"]) + " |",
+ "|" + "|".join("---" for _ in preview["columns"]) + "|",
+ ]
+ )
+ for row in preview["rows"]:
+ lines.append("| " + " | ".join(row) + " |")
+ lines.extend(["", f"[下载完整数据表]({preview['link']})"])
+ for figure in figures:
+ if figure.get("stage") in stage_ids:
+ _render_figure(figure)
- 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)}。")
+ coverage_text = (
+ f"当前计划中的分析阶段已全部完成，共纳入 {len(included)} 个可用阶段。"
+ if coverage_status == "complete"
+ else f"当前报告基于已经完成的 {len(included)} 个阶段生成，仍有部分分析尚未完成。"
+ )
+ included_text = "、".join(_stage_name(stage) for stage in included) or "无"
+ missing_text = "、".join(_stage_name(stage) for stage in missing) or "无"
+ lines = [
+ f"# {prefix} RNA-seq 结果解读报告",
+ "",
+ f"生成时间：{generated_at} ",
+ f"任务编号：`{workspace.name}`",
+ "",
+ "**任务结果路径：** ",
+ f"`{workspace / 'final'}`",
+ "",
+ "## 1. 项目概览",
+ "",
+ "| 项目 | 结果 |",
+ "|---|---|",
+ f"| 报告前缀 | {_table_cell(prefix)} |",
+ f"| 已完成分析 | {_table_cell(included_text)} |",
+ f"| 待补充分析 | {_table_cell(missing_text)} |",
+ f"| 可用阶段数 | {len(included)} |",
+ "",
+ coverage_text,
+ "",
+ "本报告以统计表和结果图为主，并在每个阶段提供简短的读图说明和初步结果概括。",
+ "",
+ "## 2. 样本与分析设计",
+ "",
+ f"计划分析：{'、'.join(_stage_name(stage) for stage in expected_stages) if expected_stages else '当前已有 RNA-seq 结果'}。",
+ "",
+ f"实际完成：{included_text}。",
+ "",
+ "上调和下调方向按照各差异比较结果表中的实验组与对照组关系展示。",
+ ]
- 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("- 根据缺失阶段和风险项补齐分析，再重新生成报告以刷新证据快照。")
+ empty_messages = {
+ "rnaseq_qc": "当前没有可用的测序质控结果，无法判断原始数据质量。",
+ "rnaseq_align": "当前没有可用的比对结果，无法判断测序片段与参考基因组的匹配情况。",
+ "rnaseq_quant": "当前没有可用的表达定量结果。",
+ "rnaseq_sample_qc": "当前没有可用的样本关系结果，无法判断重复一致性或离群样本。",
+ "rnaseq_de": "当前没有可用的差异表达结果。",
+ "rnaseq_go_enrich": "当前没有完整的 GO、KEGG 或全基因排序分析结果，不能判断具体功能主题。",
+ "rnaseq_wgcna": "当前没有可用的共表达网络结果，本节不作模块判断。",
+ "rnaseq_candidate_family": "当前没有可用的候选基因或候选家族结果。",
+ }
+ for heading, stage_ids, introduction in section_specs:
+ lines.extend(["", heading, "", introduction])
+ _render_stage_group(stage_ids, empty_messages.get(stage_ids[0], "当前没有可用结果。"))
- lines.extend(["", "## 7. 证据追溯", ""])
+ remaining_figures = [figure for figure in figures if str(Path(figure["path"])) not in rendered_figures]
+ lines.extend(["", "## 11. 结果汇总", "", "| 分析阶段 | 状态 | 主要结果 |", "|---|---|---|"])
for stage in included:
- manifest_path = manifests.get(stage, {}).get("_manifest_path") or ""
- lines.append(f"- `{stage}`：`{manifest_path}`")
- lines.extend(["", "---", "", "报告正文结束。完整结构化证据见同名 `.evidence.json`。", ""])
+ artifact_count = len((manifests.get(stage, {}).get("artifacts") or {}))
+ figure_count = sum(1 for figure in figures if figure.get("stage") == stage)
+ summary = f"已登记 {artifact_count} 个结果文件，报告展示 {figure_count} 张图"
+ lines.append(f"| {_table_cell(_stage_name(stage))} | {_table_cell(_status_text(stage))} | {_table_cell(summary)} |")
+ if remaining_figures:
+ lines.extend(["", "### 其他结果图", ""])
+ for figure in remaining_figures:
+ _render_figure(figure)
+ lines.extend(["", "---", "", "完整证据和来源记录保存在同名 `.evidence.json` 与 `.manifest.json` 文件中。", ""])
return "\n".join(lines)
+def _resolve_html_asset(report_path: Path, raw_target: str) -> Path:
+ clean_target = unquote(raw_target.split("#", 1)[0].split("?", 1)[0]).replace("\\", "/")
+ if clean_target.startswith("/workspace/"):
+ parts = [part for part in clean_target.strip("/").split("/") if part]
+ job_root = report_path.parent.parent
+ if len(parts) >= 3 and parts[1] == job_root.name:
+ return job_root.joinpath(*parts[2:]).resolve()
+ target = Path(clean_target)
+ if target.is_absolute():
+ return target.resolve()
+ return (report_path.parent / target).resolve()
+
+
+def _markdown_inline(text: str, report_path: Path) -> str:
+ tokens: list[str] = []
+
+ def stash(value: str) -> str:
+ token = f"@@HTMLTOKEN{len(tokens)}@@"
+ tokens.append(value)
+ return token
+
+ def image_replacement(match: re.Match[str]) -> str:
+ alt = match.group(1).strip() or "RNA-seq result figure"
+ raw_target = match.group(2).strip().replace("\\", "/")
+ source = raw_target
+ target = report_path.parent / Path(raw_target).name
+ mime = mimetypes.guess_type(target.name)[0] or ""
+ if target.is_file() and mime.startswith("image/"):
+ encoded = base64.b64encode(target.read_bytes()).decode("ascii")
+ source = f"data:{mime};base64,{encoded}"
+ return stash(
+ f''
+ f'{html.escape(alt)}'
+ )
+
+ def link_replacement(match: re.Match[str]) -> str:
+ label = html.escape(match.group(1).strip())
+ raw_target = match.group(2).strip().replace("\\", "/")
+ target = html.escape(raw_target, quote=True)
+ if Path(raw_target).suffix.lower() == ".pdf":
+ return stash(
+ f''
+ f'{label}'
+ )
+ return stash(f'{label}')
+
+ value = re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", image_replacement, text)
+ value = re.sub(r"(?\1", value)
+ value = re.sub(r"\*\*([^*]+)\*\*", r"\1", value)
+ for index, token_value in enumerate(tokens):
+ value = value.replace(f"@@HTMLTOKEN{index}@@", token_value)
+ return value
+
+
+def _markdown_to_html(markdown_text: str, report_path: Path, title: str) -> str:
+ output: list[str] = []
+ list_tag = ""
+ table_rows: list[list[str]] = []
+ in_code = False
+ code_lines: list[str] = []
+
+ def close_list() -> None:
+ nonlocal list_tag
+ if list_tag:
+ output.append(f"")
+ list_tag = ""
+
+ def flush_table() -> None:
+ nonlocal table_rows
+ if not table_rows:
+ return
+ rows = table_rows
+ table_rows = []
+ if len(rows) > 1 and all(re.fullmatch(r":?-{3,}:?", cell.replace(" ", "")) for cell in rows[1]):
+ header = rows[0]
+ body_rows = rows[2:]
+ else:
+ header = []
+ body_rows = rows
+ output.append('
')
+ if header:
+ output.append("
" + "".join(f"
{_markdown_inline(cell, report_path)}
" for cell in header) + "
")
+ output.append("")
+ for row in body_rows:
+ output.append("
" + "".join(f"
{_markdown_inline(cell, report_path)}
" for cell in row) + "
")
+ output.extend(["", "
"])
+
+ for raw_line in markdown_text.splitlines():
+ line = raw_line.rstrip()
+ if line.startswith("```"):
+ close_list()
+ flush_table()
+ if in_code:
+ output.append(f"
{html.escape(chr(10).join(code_lines))}
")
+ code_lines = []
+ in_code = False
+ else:
+ in_code = True
+ continue
+ if in_code:
+ code_lines.append(line)
+ continue
+ if line.startswith("|") and line.endswith("|"):
+ close_list()
+ table_rows.append([cell.strip() for cell in line.strip("|").split("|")])
+ continue
+ flush_table()
+ if not line.strip():
+ close_list()
+ continue
+ heading = re.match(r"^(#{1,6})\s+(.+)$", line)
+ if heading:
+ close_list()
+ level = len(heading.group(1))
+ output.append(f"{_markdown_inline(heading.group(2), report_path)}")
+ continue
+ if line == "---":
+ close_list()
+ output.append("")
+ continue
+ if line.startswith("> "):
+ close_list()
+ output.append(f"
{_markdown_inline(line[2:], report_path)}
")
+ continue
+ bullet = re.match(r"^-\s+(.+)$", line)
+ ordered = re.match(r"^\d+\.\s+(.+)$", line)
+ if bullet or ordered:
+ wanted = "ul" if bullet else "ol"
+ if list_tag != wanted:
+ close_list()
+ list_tag = wanted
+ output.append(f"<{list_tag}>")
+ item = bullet.group(1) if bullet else ordered.group(1)
+ output.append(f"