admin/PlantAI-dev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

merge: preserve latest report refinements

Browse Source
This commit is contained in:
wtx
2026-08-09 16:06:20 +08:00
parent b8dcb63fa1 0e5cece447
commit 9b9c88b18e
6 changed files with 854 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/web/vue3/main.py
+4
View File
@@ -15353,6 +15353,10 @@ async def get_results(
if group_stage not in visible_stage_names:
result_groups.pop(group_name, None)
gene_family_report_group = result_groups.get("gene_family_report")
if isinstance(gene_family_report_group, dict):
stages["gene_family_report"] = gene_family_report_group
stages["_result_groups"] = result_groups
stages["_scope"] = {
"rnaseq_scope_id": active_rnaseq_scope_id,
gfa_core/gene_family_report.py
+51 -1
View File
@@ -14,6 +14,12 @@ from gfa_core.gene_family_report_data import (
ReportSection,
build_report_data,
)
from gfa_core.gene_family_report_html import (
REPORT_RENDERER_VERSION,
collect_stage_figures,
markdown_to_html,
render_stage_figures,
)
from gfa_core.mas.agents.interpretation_evidence import load_stage_manifests
@@ -40,6 +46,7 @@ class GeneFamilyReportResult:
generation_mode: str
included_stages: tuple[str, ...]
reused: bool = False
html_path: Path | None = None
def _normalized_stages(stages: Iterable[str] | None) -> list[str]:
@@ -185,12 +192,32 @@ def _arranged_content(section: ReportSection, arranged: Mapping[str, Any]) -> st
return "\n\n".join(paragraphs)
def _render_completion_table(report_data: ReportData) -> str:
supported_stages = {section.stage for section in report_data.sections if section.stage}
included = set(report_data.included)
status_labels = {"success": "成功", "partial_success": "部分完成", "failed": "失败", "missing": "未完成"}
lines = [
f"本次计划 {len(report_data.planned)} 个阶段，纳入解读 {len(report_data.included)} 个，未纳入 {len(report_data.planned) - len(report_data.included)} 个。",
"",
"| 分析阶段 | 运行状态 | 模板支持 | 纳入报告 |",
"|---|---|---|---|",
]
for stage in report_data.planned:
status = str(report_data.statuses.get(stage, "missing"))
lines.append(
f"| `{stage}` | {status_labels.get(status, status)} | "
f"{'是' if stage in supported_stages else '否'} | {'是' if stage in included else '否'} |"
)
return "\n".join(lines)
def _render_markdown(
report_data: ReportData,
*,
job_id: str,
generation_mode: str,
arrangement: Mapping[str, Any],
figures_by_stage: Mapping[str, list[dict[str, str]]],
) -> str:
section_data = {section.section_id: section for section in report_data.sections}
arranged = {section["section_id"]: section for section in arrangement["sections"]}
@@ -198,7 +225,15 @@ def _render_markdown(
for section_id, block in _template_blocks():
if section_id not in section_data:
continue
rendered.append(block.replace("{{SECTION_CONTENT}}", _arranged_content(section_data[section_id], arranged[section_id])))
content = (
_render_completion_table(report_data)
if section_id == "common.completeness"
else _arranged_content(section_data[section_id], arranged[section_id])
)
stage = section_id.removeprefix("stage.") if section_id.startswith("stage.") else ""
if stage and figures_by_stage.get(stage):
content += render_stage_figures(stage, figures_by_stage[stage])
rendered.append(block.replace("{{SECTION_CONTENT}}", content))
markdown = "\n\n".join(rendered)
fallback_notice = (
"> 生成模式：候补脚本。模型 API 不可用或输出越过语言编排边界；报告事实仍来自确定性结果数据。"
@@ -237,9 +272,14 @@ def generate_gene_family_report(
raise ValueError("action must be 'generate' or 'open'")
final_dir = Path(workspace).expanduser().resolve() / "final"
report_path = final_dir / f"{prefix}.gene_family_report.md"
html_path = final_dir / f"{prefix}.gene_family_report.html"
manifest_path = final_dir / f"{prefix}.gene_family_report.manifest.json"
if action == "open" and report_path.is_file() and manifest_path.is_file():
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
if not html_path.is_file():
html_path.write_text(markdown_to_html(report_path.read_text(encoding="utf-8"), report_path, f"{prefix} 基因家族分析结果解读报告"), encoding="utf-8")
manifest.setdefault("artifacts", {})["gene_family_report_html"] = {"path": f"final/{html_path.name}", "full_path": str(html_path), "type": "html"}
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
metrics = manifest.get("metrics") or {}
return GeneFamilyReportResult(
report_path,
@@ -248,6 +288,7 @@ def generate_gene_family_report(
str(metrics.get("generation_mode") or "fallback_script"),
tuple(metrics.get("included_stages") or []),
reused=True,
html_path=html_path,
)
planned = _normalized_stages(expected_stages)
@@ -283,14 +324,17 @@ def generate_gene_family_report(
"reason": "api_unavailable_or_boundary_violation",
}
figures_by_stage = collect_stage_figures(final_dir, report_data.included, manifests)
markdown = _render_markdown(
report_data,
job_id=job_id,
generation_mode=generation_mode,
arrangement=arrangement,
figures_by_stage=figures_by_stage,
)
final_dir.mkdir(parents=True, exist_ok=True)
report_path.write_text(markdown, encoding="utf-8")
html_path.write_text(markdown_to_html(markdown, report_path, f"{prefix} 基因家族分析结果解读报告"), encoding="utf-8")
manifest = {
"manifest_version": "1.0",
"stage": REPORT_STAGE,
@@ -308,6 +352,8 @@ def generate_gene_family_report(
"included_stages": list(report_data.included),
"unsupported_stages": list(report_data.unsupported_stages),
"failed_or_missing_stages": [stage for stage in planned if stage not in report_data.included],
"report_renderer_version": REPORT_RENDERER_VERSION,
"embedded_figure_count": sum(len(items) for items in figures_by_stage.values()),
},
"model": model_metadata,
"artifacts": {
@@ -316,6 +362,9 @@ def generate_gene_family_report(
"full_path": str(report_path),
"type": "markdown",
},
"gene_family_report_html": {
"path": f"final/{html_path.name}", "full_path": str(html_path), "type": "html",
},
"gene_family_report_manifest": {
"path": f"final/{manifest_path.name}",
"full_path": str(manifest_path),
@@ -330,4 +379,5 @@ def generate_gene_family_report(
report_data.report_status,
generation_mode,
report_data.included,
html_path=html_path,
)
gfa_core/gene_family_report_data.py
+1 -7
View File
@@ -719,13 +719,7 @@ def build_report_data(
)
for index, sentence in enumerate(integrated_observations, start=1)
),
interpretations=(
_sentence(
"common.integrated.boundary",
"综合结论只并列上述已提取事实；当前源结果不足以无偏地指定候选基因，也不据此建立功能因果。",
"interpretation",
),
),
interpretations=(),
),
ReportSection(
"common.limitations",
gfa_core/gene_family_report_html.py
+122
View File
@@ -0,0 +1,122 @@
# HTML rendering and representative-figure selection for gene-family reports.
from __future__ import annotations
import base64
import html
import mimetypes
from pathlib import Path
import re
from typing import Any, Iterable, Mapping
REPORT_RENDERER_VERSION = "gene_family_report_html_v4"
IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp", ".svg"}
STAGE_FIGURE_TEXT = {
"phylogeny": ("系统发育图展示目标成员与参考成员之间的聚类关系。", "结合分支拓扑、节点支持率和成员标签阅读，不把邻近关系直接等同于功能一致。", "目标与参考序列的聚类关系见图，具体分支以图中节点和标签为准。"),
"reference_classification": ("参考分类图展示目标成员与已知参考亚家族的相对位置。", "结合参考成员标签、目标成员所在分支及支持率判断分类线索。", "目标成员的参考辅助分类位置见图，分类结论以本节确定性结果为准。"),
"motif": ("Motif图展示各成员短保守片段的组成、顺序和相对位置。", "按成员逐行比较Motif颜色、编号、数量和排列，不把Motif相似直接写成功能相同。", "本次成员的Motif组成与排列见图，数量统计以本节结果数据为准。"),
"domain": ("Domain图展示蛋白主要功能区域的类型和位置。", "结合结构域名称、覆盖区间和成员间差异阅读。", "各成员的结构域组成见图，覆盖数量和比例以本节结果数据为准。"),
"structure": ("基因结构图展示外显子、内含子和基因长度的相对关系。", "结合图例比较外显子数量、结构排列和异常长基因。", "目标成员的基因结构差异见图，具体统计以本节结果数据为准。"),
"combo": ("综合组合图并列展示系统发育、Motif、Domain和基因结构信息。", "沿同一成员横向比较多个证据面板，关注一致模式与明显差异。", "多类结构证据的对应关系见图，不据此单独确认功能改变。"),
"promoter": ("启动子图展示顺式作用元件的类型、数量或位置分布。", "结合元件图例、位置坐标和成员标签比较，不把预测位点写成已验证调控。", "目标成员启动子元件的分布见图，数量和类别统计以本节结果数据为准。"),
"synteny": ("共线性图展示染色体区段和基因对之间的连线关系。", "结合染色体标签、连线和家族成员标记阅读，区分全基因组背景与家族证据。", "本次共线性背景见图，家族专属关系以本节可追溯结果为准。"),
"karyotype": ("染色体分布图展示家族成员在染色体上的物理位置。", "结合染色体编号、位置刻度和基因标签查看分布范围。", "目标家族成员的染色体位置见图，成员数和染色体数以本节结果数据为准。"),
"interspecies_synteny": ("种间共线性图展示目标物种与参考物种之间的保守区块和基因连线。", "结合物种、染色体、标记连线和成员标签阅读，不直接移植参考物种功能。", "跨物种共线性关系见图，目标与参考家族配对数以本节结果数据为准。"),
"pep": ("蛋白特征图展示理化性质、跨膜区域或亚细胞定位预测结果。", "根据具体图例比较成员分布，区分不同预测子任务。", "本次蛋白预测结果见图，具体数量与分类以本节结果数据为准。"),
"tfbs": ("Motif统计图展示不同基因的位点计数、全部Motif计数和高频Motif。", "结合基因标签、颜色图例、计数刻度和Motif名称比较分布。", "本次gene、all和top motif统计结果见图，具体计数以图例和源结果为准。"),
}
def _artifact_image(final_dir: Path, artifact: Any) -> Path | None:
if not isinstance(artifact, Mapping): return None
raw = str(artifact.get("full_path") or artifact.get("path") or "").strip()
target = final_dir / Path(raw).name if raw else None
return target if target and target.is_file() and target.suffix.lower() in IMAGE_SUFFIXES else None
def _kind(role: str, filename: str) -> str:
value = f"{role} {filename}".lower()
groups = (("gene_motif_count", ("gene_motif_heatmap",)), ("all_motif_count", ("all_motif_count_heatmap",)), ("top_motif", ("top_motifs_barplot",)), ("tree", ("tree",)), ("msa", ("msa_plot", "conserved_region")), ("synteny", ("synteny", "circos")), ("motif", ("motif",)), ("domain", ("domain",)), ("structure", ("structure",)), ("combo", ("combo",)), ("promoter_combined", ("promoter_v6_combined",)), ("promoter_position", ("promoter_position_distribution",)), ("promoter", ("promoter",)), ("karyotype", ("karyotype",)), ("tmhmm", ("tmhmm",)), ("subcellular", ("subcellular",)), ("properties", ("properties",)))
return next((kind for kind, tokens in groups if any(token in value for token in tokens)), role.lower())
def _rank(role: str, path: Path) -> tuple[int, int, str]:
value = f"{role} {path.name}".lower()
role_lower = role.lower()
preferred = -5 if role_lower == "synteny_marked_png" else -4 if "circular" in value else -2 if "marked" in value else -1 if "summary" in value else 0
return (preferred, 0, path.name)
def collect_stage_figures(final_dir: Path, stages: Iterable[str], manifests: Mapping[str, Mapping[str, Any]]) -> dict[str, list[dict[str, str]]]:
result = {}
for stage in stages:
selected = {}; seen = set()
for role, artifact in ((manifests.get(stage) or {}).get("artifacts") or {}).items():
target = _artifact_image(final_dir, artifact)
if target is None or target.name in seen: continue
seen.add(target.name); kind = _kind(str(role), target.name)
candidate = (_rank(str(role), target), {"role": str(role), "filename": target.name})
if kind not in selected or candidate[0] < selected[kind][0]: selected[kind] = candidate
if stage == "tfbs":
for role, pattern in (
("gene_motif_heatmap", "*gene_motif_heatmap.png"),
("all_motif_count_heatmap", "*all_motif_count_heatmap.png"),
("top_motifs_barplot", "*top_motifs_barplot.png"),
):
matches = sorted(final_dir.glob(pattern))
if not matches:
continue
target = matches[0]
kind = _kind(role, target.name)
selected[kind] = (_rank(role, target), {"role": role, "filename": target.name})
figures = [entry[1] for _, entry in sorted(selected.items(), key=lambda item: (item[1][0], item[0]))][:4]
if figures: result[stage] = figures
return result
def render_stage_figures(stage: str, figures: list[dict[str, str]]) -> str:
meaning, reading, result = STAGE_FIGURE_TEXT.get(stage, ("结果图展示该阶段的主要计算产物。", "结合图题、坐标轴、图例和成员标签阅读。", "本阶段代表结果见图，具体事实以本节确定性数据为准。"))
lines = ["", "### 结果图"]
for index, figure in enumerate(figures, 1):
title = figure["role"]
lines.extend(["", f"#### 图 {index}. {title}", "", f"![{title}]({figure['filename']})", "", f"图形含义：{meaning}", "", f"阅读方法：{reading}", "", f"本次结果：{result}"])
return "\n".join(lines)
def _inline(value: str, report_path: Path) -> str:
tokens = []
def stash(content):
token = f"@@GFAHTML{len(tokens)}@@"; tokens.append(content); return token
def image(match):
alt = match.group(1).strip() or "基因家族结果图"; target = report_path.parent / Path(match.group(2).strip().replace("\\", "/")).name
source = target.name; mime = mimetypes.guess_type(target.name)[0] or ""
if target.is_file() and mime.startswith("image/"): source = f"data:{mime};base64,{base64.b64encode(target.read_bytes()).decode('ascii')}"
return stash(f'<figure><img src="{html.escape(source, quote=True)}" alt="{html.escape(alt, quote=True)}" loading="lazy"><figcaption>{html.escape(alt)}</figcaption></figure>')
def link(match):
label = html.escape(match.group(1).strip()); target = Path(match.group(2).strip().replace("\\", "/")).name
return stash(f'<a href="{html.escape(target, quote=True)}" target="_blank" rel="noopener">{label}</a>')
value = re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", image, value); value = re.sub(r"(?<!!)\[([^\]]+)\]\(([^)]+)\)", link, value)
value = re.sub(r"`([^`]+)`", lambda m: stash(f"<code>{html.escape(m.group(1))}</code>"), value); value = html.escape(value); value = re.sub(r"\*\*([^*]+)\*\*", r"<strong>\1</strong>", value)
for index, content in enumerate(tokens): value = value.replace(f"@@GFAHTML{index}@@", content)
return value
def markdown_to_html(markdown: str, report_path: Path, title: str) -> str:
output = []; list_tag = ""; table_rows = []
def close_list():
nonlocal list_tag
if list_tag: output.append(f"</{list_tag}>"); list_tag = ""
def flush_table():
nonlocal table_rows
if not table_rows: return
rows, table_rows = table_rows, []
if len(rows) >= 2 and all(re.fullmatch(r":?-{3,}:?", cell.replace(" ", "")) for cell in rows[1]):
output.append('<div class="table-wrap"><table><thead><tr>' + ''.join(f"<th>{_inline(c, report_path)}</th>" for c in rows[0]) + '</tr></thead><tbody>'); output.extend('<tr>' + ''.join(f"<td>{_inline(c, report_path)}</td>" for c in row) + '</tr>' for row in rows[2:]); output.append('</tbody></table></div>')
else: output.extend(f"<p>{_inline(' | '.join(row), report_path)}</p>" for row in rows)
for raw in markdown.splitlines():
line = raw.strip()
if line.startswith("|") and line.endswith("|"): close_list(); table_rows.append([c.strip() for c in line.strip("|").split("|")]); continue
flush_table()
if not line: close_list(); continue
heading = re.match(r"^(#{1,6})\s+(.+)$", line)
if heading: close_list(); level=len(heading.group(1)); output.append(f"<h{level}>{_inline(heading.group(2), report_path)}</h{level}>"); continue
if line.startswith("> "): close_list(); output.append(f"<blockquote>{_inline(line[2:], report_path)}</blockquote>"); continue
item = re.match(r"^([-*]|\d+\.)\s+(.+)$", line)
if item:
tag = "ol" if item.group(1)[0].isdigit() else "ul"
if list_tag != tag: close_list(); list_tag=tag; output.append(f"<{tag}>")
output.append(f"<li>{_inline(item.group(2), report_path)}</li>"); continue
close_list(); rendered=_inline(line, report_path); output.append(rendered if line.startswith("![") else f"<p>{rendered}</p>")
close_list(); flush_table(); body="\n".join(output); safe_title=html.escape(title)
return f'''<!doctype html><html lang="zh-CN"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>{safe_title}</title><style>:root{{--ink:#17352c;--paper:#fffdf7;--line:#dce5dc;--accent:#157a57;--soft:#edf6ef;--warm:#f2e5c7}}*{{box-sizing:border-box}}html{{background:#e9efe9}}body{{margin:0;color:var(--ink);font-family:"Noto Serif SC","Source Han Serif SC","Microsoft YaHei",serif;line-height:1.85;background:radial-gradient(circle at top left,#f8edcf 0,transparent 28rem),linear-gradient(150deg,#edf4ec,#f6f2e8)}}main{{width:min(1120px,calc(100% - 32px));margin:32px auto 64px;padding:clamp(28px,5vw,72px);background:var(--paper);border:1px solid rgba(23,53,44,.12);border-radius:22px;box-shadow:0 24px 70px rgba(35,62,48,.14);overflow:hidden}}h1{{font-size:clamp(2rem,5vw,3.5rem);line-height:1.2;color:#0d4f39}}h2{{margin:3.2rem 0 1.1rem;padding-top:.7rem;border-top:3px solid var(--warm);font-size:clamp(1.45rem,3vw,2rem)}}h3{{color:var(--accent)}}p,h1,h2,h3,h4,a,figcaption,code{{overflow-wrap:anywhere;word-break:break-word}}blockquote{{margin:1.4rem 0;padding:1rem 1.25rem;border-left:5px solid var(--accent);background:var(--soft)}}code{{padding:.14em .4em;border-radius:5px;background:#edf0eb}}figure{{margin:1.6rem 0 2.2rem;padding:clamp(8px,2vw,18px);border:1px solid var(--line);border-radius:16px;background:#fff;box-shadow:0 10px 30px rgba(28,62,48,.09)}}img{{display:block;max-width:100%;height:auto;margin:auto;border-radius:10px}}figcaption{{margin-top:.7rem;text-align:center;color:#60726b}}a{{color:#087553}}.table-wrap{{overflow-x:auto;margin:1.4rem 0;border:1px solid var(--line);border-radius:12px}}table{{width:100%;border-collapse:collapse;background:#fff}}th,td{{padding:.72rem .85rem;border-bottom:1px solid var(--line);text-align:left;vertical-align:top}}th{{background:var(--soft)}}@media(max-width:640px){{main{{width:100%;margin:0;padding:24px 18px 48px;border:0;border-radius:0}}}}@media print{{html,body{{background:#fff}}main{{width:100%;margin:0;padding:0;border:0;box-shadow:none}}figure{{break-inside:avoid}}}}</style></head><body><main>{body}</main></body></html>'''
gfa_core/rnaseq_report.py
+663 -150
View File
@@ -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'<figure><img src="{html.escape(source, quote=True)}" '
f'alt="{html.escape(alt, quote=True)}" loading="lazy">'
f'<figcaption>{html.escape(alt)}</figcaption></figure>'
)
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'<figure><iframe class="pdf-frame" src="{target}" title="{label}"></iframe>'
f'<figcaption><a href="{target}" target="_blank" rel="noopener">{label}</a></figcaption></figure>'
)
return stash(f'<a href="{target}" target="_blank" rel="noopener">{label}</a>')
value = re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", image_replacement, text)
value = re.sub(r"(?<!!)\[([^\]]+)\]\(([^)]+)\)", link_replacement, value)
value = html.escape(value)
value = re.sub(r"`([^`]+)`", r"<code>\1</code>", value)
value = re.sub(r"\*\*([^*]+)\*\*", r"<strong>\1</strong>", 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}>")
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('<div class="table-wrap"><table>')
if header:
output.append("<thead><tr>" + "".join(f"<th>{_markdown_inline(cell, report_path)}</th>" for cell in header) + "</tr></thead>")
output.append("<tbody>")
for row in body_rows:
output.append("<tr>" + "".join(f"<td>{_markdown_inline(cell, report_path)}</td>" for cell in row) + "</tr>")
output.extend(["</tbody>", "</table></div>"])
for raw_line in markdown_text.splitlines():
line = raw_line.rstrip()
if line.startswith("```"):
close_list()
flush_table()
if in_code:
output.append(f"<pre><code>{html.escape(chr(10).join(code_lines))}</code></pre>")
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"<h{level}>{_markdown_inline(heading.group(2), report_path)}</h{level}>")
continue
if line == "---":
close_list()
output.append("<hr>")
continue
if line.startswith("> "):
close_list()
output.append(f"<blockquote>{_markdown_inline(line[2:], report_path)}</blockquote>")
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"<li>{_markdown_inline(item, report_path)}</li>")
continue
close_list()
if re.fullmatch(r"!?\[[^\]]+\]\([^)]+\)", line):
output.append(_markdown_inline(line, report_path))
continue
output.append(f"<p>{_markdown_inline(line, report_path)}</p>")
close_list()
flush_table()
if in_code:
output.append(f"<pre><code>{html.escape(chr(10).join(code_lines))}</code></pre>")
safe_title = html.escape(title)
body = "\n".join(output)
return f"""<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{safe_title}</title>
<style>
:root {{ --ink:#17352c; --muted:#60726b; --paper:#fffdf7; --line:#dce5dc; --accent:#157a57; --soft:#edf6ef; --warm:#f2e5c7; }}
* {{ box-sizing:border-box; }}
html {{ background:#e9efe9; scroll-behavior:smooth; }}
body {{ margin:0; color:var(--ink); font-family:"Noto Serif SC","Source Han Serif SC","Microsoft YaHei",serif; line-height:1.85; background:radial-gradient(circle at top left,#f8edcf 0,transparent 28rem),linear-gradient(150deg,#edf4ec,#f6f2e8); }}
main {{ width:min(1120px,calc(100% - 32px)); margin:32px auto 64px; padding:clamp(28px,5vw,72px); background:var(--paper); border:1px solid rgba(23,53,44,.12); border-radius:22px; box-shadow:0 24px 70px rgba(35,62,48,.14); }}
h1 {{ margin:0 0 2rem; font-size:clamp(2rem,5vw,3.5rem); line-height:1.2; letter-spacing:-.04em; color:#0d4f39; }}
h2 {{ margin:3.2rem 0 1.1rem; padding-top:.7rem; border-top:3px solid var(--warm); font-size:clamp(1.45rem,3vw,2rem); line-height:1.35; }}
h3 {{ margin:2rem 0 .8rem; color:var(--accent); font-size:1.25rem; }}
p {{ margin:.75rem 0; }}
p,h3,a,figcaption,code {{ overflow-wrap:anywhere; word-break:break-word; }}
ul,ol {{ padding-left:1.45rem; }}
li {{ margin:.42rem 0; }}
blockquote {{ margin:1.4rem 0; padding:1rem 1.25rem; border-left:5px solid var(--accent); border-radius:0 12px 12px 0; background:var(--soft); color:#35584c; }}
code {{ padding:.14em .4em; border-radius:5px; background:#edf0eb; font-family:"Cascadia Mono","Fira Code",monospace; font-size:.9em; }}
pre {{ overflow:auto; padding:1rem; border-radius:12px; background:#142a23; color:#f3f6f2; }}
pre code {{ padding:0; background:none; color:inherit; }}
figure {{ margin:1.6rem 0 2.2rem; padding:clamp(8px,2vw,18px); border:1px solid var(--line); border-radius:16px; background:#fff; box-shadow:0 10px 30px rgba(28,62,48,.09); }}
img {{ display:block; max-width:100%; height:auto; margin:auto; border-radius:10px; }}
.pdf-frame {{ display:block; width:100%; height:min(72vh,760px); border:0; border-radius:10px; background:#f3f5f2; }}
figcaption {{ margin-top:.7rem; text-align:center; color:var(--muted); font-size:.92rem; }}
a {{ color:#087553; text-decoration-thickness:1px; text-underline-offset:3px; }}
.table-wrap {{ max-height:36rem; overflow:auto; margin:1.4rem 0; border:1px solid var(--line); border-radius:12px; scrollbar-gutter:stable; }}
table {{ width:100%; min-width:max-content; border-collapse:separate; border-spacing:0; background:#fff; }}
th,td {{ padding:.72rem .85rem; border-bottom:1px solid var(--line); text-align:left; vertical-align:top; white-space:nowrap; }}
th {{ position:sticky; top:0; z-index:2; background:var(--soft); color:#174d3b; }}
th:first-child,td:first-child {{ position:sticky; left:0; z-index:1; background:#fff; }}
th:first-child {{ z-index:3; background:var(--soft); }}
hr {{ margin:3rem 0; border:0; border-top:1px solid var(--line); }}
@media (max-width:640px) {{ main {{ width:100%; margin:0; padding:24px 18px 48px; border:0; border-radius:0; }} h2 {{ margin-top:2.4rem; }} }}
@media print {{ html,body {{ background:#fff; }} main {{ width:100%; margin:0; padding:0; border:0; box-shadow:none; }} figure {{ break-inside:avoid; }} }}
</style>
</head>
<body><main>{body}</main></body>
</html>
"""
def _model_claim_catalog(evidence: dict[str, Any]) -> list[dict[str, str]]:
catalog: list[dict[str, str]] = []
@@ -1113,11 +1621,12 @@ def _model_enhancement(evidence: dict[str, Any], enabled: bool) -> tuple[str, di
def _artifact(workspace: Path, path: Path, kind: str) -> dict[str, Any]:
artifact_type = {".md": "markdown", ".html": "html", ".json": "json"}.get(path.suffix.lower(), "file")
return {
"artifact": kind,
"path": str(path.relative_to(workspace)),
"full_path": str(path.absolute()),
"type": "markdown" if path.suffix == ".md" else "json",
"type": artifact_type,
"size": path.stat().st_size,
"sha256": _sha256(path),
}
@@ -1156,13 +1665,14 @@ def generate_rnaseq_report(
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"
html_path = final_dir / f"{prefix}.rnaseq_report.html"
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)):
if action == "open" and all(path.is_file() for path in (report_path, html_path, evidence_path, model_path, manifest_path)):
previous = _read_json(manifest_path)
metrics = previous.get("metrics") or {}
if (
@@ -1248,6 +1758,8 @@ def generate_rnaseq_report(
_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)
html_report = _markdown_to_html(markdown, report_path, f"{prefix} RNA-seq 结果解读报告")
_write_text_atomic(html_path, html_report)
manifest = {
"manifest_version": "1.0",
@@ -1288,6 +1800,7 @@ def generate_rnaseq_report(
"provenance": {"source_manifests": source_records},
"artifacts": {
"rnaseq_report": _artifact(workspace, report_path, "rnaseq_report"),
"rnaseq_report_html": _artifact(workspace, html_path, "rnaseq_report_html"),
"rnaseq_report_evidence": _artifact(workspace, evidence_path, "rnaseq_report_evidence"),
"rnaseq_report_model": _artifact(workspace, model_path, "rnaseq_report_model"),
},
gfa_core/templates/gene_family_report_v1.md
+13 -32
View File
@@ -3,10 +3,8 @@
- 任务：{{JOB_ID}}
- 分析前缀：{{PREFIX}}
- 生成模式：{{GENERATION_MODE}}
- 生成时间：{{GENERATED_AT}}
{{FALLBACK_NOTICE}}
## 一、报告首页
@@ -51,91 +49,91 @@
报告按计划逐阶段选择成功 Manifest；当各阶段使用的成员集合不同，正文明确区分目标成员、目标加参考成员和全基因组背景。
<!-- GFA_SECTION stage.identify -->
## 四、基因家族鉴定（Identify） (`identify`)
## 四、基因家族鉴定（Identify）
### 4.3 最终成员数量与命名
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.phylogeny -->
## 五、亲缘关系分析（Phylogeny） (`phylogeny`)
## 五、亲缘关系分析（Phylogeny）
### 5.5 亲缘关系阶段小结
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.reference_classification -->
## 六、参考物种辅助的小组分类（Reference Classification） (`reference_classification`)
## 六、参考物种辅助的小组分类（Reference Classification）
### 6.2 分类可信度表述
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.motif -->
## 七、短保守片段分析（Motif） (`motif`)
## 七、短保守片段分析（Motif）
### 7.2 Motif 数量与出现频率
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.domain -->
## 八、主要功能区域分析（Domain） (`domain`)
## 八、主要功能区域分析（Domain）
### 8.1 Domain 整体统计
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.structure -->
## 九、基因内部结构分析（Structure） (`structure`)
## 九、基因内部结构分析（Structure）
### 9.1 结构覆盖与异常结构
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.combo -->
## 十、综合组合图（Combo） (`combo`)
## 十、综合组合图（Combo）
### 10.1 树 + Motif + Domain + Structure 组合图
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.promoter -->
## 十一、启动子元件分析（Promoter） (`promoter`)
## 十一、启动子元件分析（Promoter）
### 11.1 元件总量、类别与高级统计
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.tfbs -->
## 十二、转录因子结合位点预测（TFBS） (`tfbs`)
## 十二、转录因子结合位点预测（TFBS）
### 12.1 高频结合片段与统计边界
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.synteny -->
## 十三、种内共线性分析（Synteny） (`synteny`)
## 十三、种内共线性分析（Synteny）
### 13.1 全基因组共线性背景与家族证据缺口
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.karyotype -->
## 十四、染色体分布（Karyotype） (`karyotype`)
## 十四、染色体分布（Karyotype）
### 14.1 染色体位置概述
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.interspecies_synteny -->
## 十五、种间共线性与跨物种关系（Interspecies Synteny） (`interspecies_synteny`)
## 十五、种间共线性与跨物种关系（Interspecies Synteny）
### 15.6 Interspecies Synteny 阶段小结
{{SECTION_CONTENT}}
<!-- GFA_SECTION stage.pep -->
## 十六、蛋白基本特征分析（PEP） (`pep`)
## 十六、蛋白基本特征分析（PEP）
### 16.1 蛋白理化性质分布图
@@ -151,20 +149,3 @@
### 17.5 综合结论
{{SECTION_CONTENT}}
<!-- GFA_SECTION common.limitations -->
## 十八、主要限制
{{SECTION_CONTENT}}
<!-- GFA_SECTION common.recommendations -->
## 十九、后续验证建议
{{SECTION_CONTENT}}
<!-- GFA_SECTION common.checklist -->
## 二十、报告交付前检查清单
### 20.1 可追溯性与语言检查
{{SECTION_CONTENT}}