314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""Lightweight, provider-free RNA-seq manifest evidence snapshot.
|
||
|
||
The live interpretation view and the formal report share this module so scope
|
||
selection and manifest provenance have one implementation. It intentionally
|
||
does not import the MAS agent package.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any, Iterable
|
||
|
||
|
||
REPORT_STAGE = "rnaseq_report"
|
||
SUCCESS_STATUSES = {"success", "partial_success"}
|
||
GLOBAL_LEGACY_STAGES = {
|
||
"rnaseq_qc",
|
||
"rnaseq_align",
|
||
"rnaseq_quant",
|
||
"annotation",
|
||
"annotation_go",
|
||
"annotation_kegg",
|
||
}
|
||
RNASEQ_STAGE_NAMES = {
|
||
"rnaseq_qc": "RNA-seq质控",
|
||
"rnaseq_align": "RNA-seq比对",
|
||
"rnaseq_quant": "RNA-seq定量",
|
||
"rnaseq_sample_qc": "RNA-seq样本质量评估",
|
||
"rnaseq_de": "RNA-seq差异表达",
|
||
"annotation": "背景功能注释",
|
||
"annotation_go": "GO背景自动注释",
|
||
"annotation_kegg": "KEGG/KO背景自动注释",
|
||
"rnaseq_go_enrich": "RNA-seq GO富集分析",
|
||
"rnaseq_kegg_enrich": "RNA-seq KEGG富集分析",
|
||
"rnaseq_enrich": "RNA-seq富集分析（兼容入口）",
|
||
"rnaseq_fgsea": "RNA-seq fgsea排序富集",
|
||
"rnaseq_wgcna": "RNA-seq WGCNA共表达分析",
|
||
"rnaseq_candidate_family": "RNA-seq候选家族筛选",
|
||
"rnaseq_family_score": "RNA-seq候选家族评分",
|
||
}
|
||
RNASEQ_STAGE_ORDER = list(RNASEQ_STAGE_NAMES)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RNASeqManifestSelection:
|
||
rnaseq_scope_id: str
|
||
discovered_scope_ids: tuple[str, ...]
|
||
manifests: dict[str, dict[str, Any]]
|
||
|
||
|
||
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 manifest_identity_value(manifest: dict[str, Any], key: str) -> str:
|
||
context = manifest.get("stage_context") or manifest.get("stageContext") or {}
|
||
params = manifest.get("parameters") or {}
|
||
return str(manifest.get(key) or params.get(key) or context.get(key) or "").strip()
|
||
|
||
|
||
def is_rnaseq_evidence_stage(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 RNASEQ_STAGE_ORDER.index(stage), stage
|
||
except ValueError:
|
||
return len(RNASEQ_STAGE_ORDER), stage
|
||
|
||
|
||
def _manifest_score(path: Path, manifest: dict[str, Any]) -> tuple[int, int, int, float]:
|
||
status = str(manifest.get("status") or "")
|
||
status_rank = 2 if status == "success" else 1 if status == "partial_success" else 0
|
||
metrics = manifest.get("metrics") if isinstance(manifest.get("metrics"), dict) else {}
|
||
artifacts = manifest.get("artifacts") if isinstance(manifest.get("artifacts"), dict) else {}
|
||
try:
|
||
mtime = path.stat().st_mtime
|
||
except OSError:
|
||
mtime = 0.0
|
||
return status_rank, len(metrics), len(artifacts), mtime
|
||
|
||
|
||
def select_rnaseq_manifests(
|
||
final_dir: Path,
|
||
prefix: str,
|
||
*,
|
||
rnaseq_scope_id: str = "",
|
||
) -> RNASeqManifestSelection:
|
||
"""Select one non-mixing manifest per RNA-seq stage for a prefix/scope."""
|
||
final_dir = Path(final_dir)
|
||
pattern = f"{prefix}.*.manifest.json" if prefix else "*.manifest.json"
|
||
candidates: list[tuple[Path, dict[str, Any]]] = []
|
||
latest_by_scope: dict[str, float] = {}
|
||
for path in sorted(final_dir.glob(pattern)) if final_dir.exists() else []:
|
||
manifest = read_json(path)
|
||
stage = str(manifest.get("stage") or "")
|
||
if not is_rnaseq_evidence_stage(stage):
|
||
continue
|
||
manifest_prefix = str(manifest.get("prefix") or "").strip()
|
||
if prefix and manifest_prefix and manifest_prefix != prefix:
|
||
continue
|
||
candidates.append((path, manifest))
|
||
scope = manifest_identity_value(manifest, "rnaseq_scope_id")
|
||
if scope:
|
||
try:
|
||
latest_by_scope[scope] = max(latest_by_scope.get(scope, 0.0), path.stat().st_mtime)
|
||
except OSError:
|
||
latest_by_scope.setdefault(scope, 0.0)
|
||
|
||
discovered = sorted(latest_by_scope, key=lambda item: (-latest_by_scope[item], item))
|
||
resolved_scope = str(rnaseq_scope_id or "").strip() or (discovered[0] if discovered else "")
|
||
selected_paths: dict[str, Path] = {}
|
||
selected: dict[str, dict[str, Any]] = {}
|
||
for path, manifest in candidates:
|
||
stage = str(manifest.get("stage") or "")
|
||
scope = manifest_identity_value(manifest, "rnaseq_scope_id")
|
||
if resolved_scope and scope and scope != resolved_scope:
|
||
continue
|
||
if resolved_scope and not scope and stage not in GLOBAL_LEGACY_STAGES:
|
||
continue
|
||
existing = selected.get(stage)
|
||
if existing is not None and _manifest_score(selected_paths[stage], existing) >= _manifest_score(path, manifest):
|
||
continue
|
||
manifest = dict(manifest)
|
||
manifest["_manifest_path"] = str(path)
|
||
selected[stage] = manifest
|
||
selected_paths[stage] = path
|
||
return RNASeqManifestSelection(resolved_scope, tuple(discovered), selected)
|
||
|
||
|
||
def _compact_value(value: Any, limit: int = 240) -> str:
|
||
if isinstance(value, (dict, list)):
|
||
text = json.dumps(value, ensure_ascii=False, sort_keys=True)
|
||
else:
|
||
text = str(value)
|
||
return text if len(text) <= limit else text[: limit - 1] + "…"
|
||
|
||
|
||
def resolve_artifact_path(
|
||
final_dir: Path,
|
||
item: dict[str, Any],
|
||
*,
|
||
require_exists: bool = False,
|
||
) -> Path | None:
|
||
"""Resolve artifacts after a workspace has been moved or mounted elsewhere."""
|
||
final_dir = Path(final_dir)
|
||
workspace = final_dir.parent
|
||
candidates: list[Path] = []
|
||
for raw in (item.get("full_path"), item.get("path")):
|
||
if not raw:
|
||
continue
|
||
path = Path(str(raw))
|
||
candidates.append(path if path.is_absolute() else workspace / path)
|
||
if path.is_absolute():
|
||
candidates.append(final_dir / path.name)
|
||
|
||
unique: list[Path] = []
|
||
seen: set[str] = set()
|
||
for path in candidates:
|
||
identity = str(path)
|
||
if identity not in seen:
|
||
seen.add(identity)
|
||
unique.append(path)
|
||
for path in unique:
|
||
if path.exists():
|
||
return path
|
||
if require_exists:
|
||
return None
|
||
return unique[0] if unique else None
|
||
|
||
|
||
def build_rnaseq_evidence_snapshot(
|
||
final_dir: Path,
|
||
prefix: str,
|
||
*,
|
||
rnaseq_scope_id: str = "",
|
||
include_stages: Iterable[str] | None = None,
|
||
selection: RNASeqManifestSelection | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Build a compact evidence object without invoking an LLM or agent cache."""
|
||
selection = selection or select_rnaseq_manifests(
|
||
final_dir, prefix, rnaseq_scope_id=rnaseq_scope_id
|
||
)
|
||
allowed = {str(stage) for stage in include_stages or [] if str(stage).strip()}
|
||
manifests = {
|
||
stage: manifest
|
||
for stage, manifest in selection.manifests.items()
|
||
if not allowed or stage in allowed
|
||
}
|
||
stages = sorted(manifests, key=stage_sort_key)
|
||
stage_flow: list[dict[str, Any]] = []
|
||
findings: list[dict[str, Any]] = []
|
||
reviews: list[dict[str, Any]] = []
|
||
cards: list[dict[str, Any]] = []
|
||
risks: list[dict[str, Any]] = []
|
||
card_index = 0
|
||
for stage in stages:
|
||
manifest = manifests[stage]
|
||
status = str(manifest.get("status") or "unknown")
|
||
metrics = manifest.get("metrics") if isinstance(manifest.get("metrics"), dict) else {}
|
||
parameters = manifest.get("parameters") if isinstance(manifest.get("parameters"), dict) else {}
|
||
artifacts = manifest.get("artifacts") if isinstance(manifest.get("artifacts"), dict) else {}
|
||
metric_items = list(sorted(metrics.items()))[:12]
|
||
metric_summary = ", ".join(f"{key}={_compact_value(value, 80)}" for key, value in metric_items[:5])
|
||
key_result = metric_summary or f"登记产物 {len(artifacts)} 个"
|
||
stage_flow.append(
|
||
{
|
||
"id": stage,
|
||
"name": RNASEQ_STAGE_NAMES.get(stage, stage),
|
||
"status": status,
|
||
"key_result": key_result,
|
||
}
|
||
)
|
||
findings.append(
|
||
{
|
||
"stage": stage,
|
||
"title": RNASEQ_STAGE_NAMES.get(stage, stage),
|
||
"description": f"manifest 状态为 {status}；{key_result}。",
|
||
}
|
||
)
|
||
limitations: list[str] = []
|
||
if status == "partial_success":
|
||
limitations.append("该阶段为 partial_success，结果可用但不完整。")
|
||
elif status != "success":
|
||
limitations.append("该阶段未成功，不纳入正式结论。")
|
||
missing_artifacts: list[str] = []
|
||
for name, item in sorted(artifacts.items()):
|
||
if not isinstance(item, dict):
|
||
continue
|
||
path = resolve_artifact_path(final_dir, item)
|
||
exists = bool(path and path.exists())
|
||
card_index += 1
|
||
cards.append(
|
||
{
|
||
"id": f"RNA-E{card_index:04d}",
|
||
"stage": stage,
|
||
"claim": f"产物 {name}",
|
||
"evidence": f"path={path or ''}, exists={exists}",
|
||
"source_type": "raw_file" if exists else "manifest",
|
||
}
|
||
)
|
||
if not exists:
|
||
missing_artifacts.append(str(name))
|
||
for key, value in metric_items:
|
||
card_index += 1
|
||
cards.append(
|
||
{
|
||
"id": f"RNA-E{card_index:04d}",
|
||
"stage": stage,
|
||
"claim": f"指标 {key}",
|
||
"evidence": _compact_value(value),
|
||
"source_type": "manifest",
|
||
}
|
||
)
|
||
safe_parameter_items = [
|
||
(key, value)
|
||
for key, value in sorted(parameters.items())
|
||
if not any(secret in str(key).lower() for secret in ("api_key", "token", "secret", "password"))
|
||
][:10]
|
||
for key, value in safe_parameter_items:
|
||
card_index += 1
|
||
cards.append(
|
||
{
|
||
"id": f"RNA-E{card_index:04d}",
|
||
"stage": stage,
|
||
"claim": f"参数 {key}",
|
||
"evidence": _compact_value(value),
|
||
"source_type": "manifest",
|
||
}
|
||
)
|
||
if missing_artifacts:
|
||
shown = missing_artifacts[:12]
|
||
remainder = len(missing_artifacts) - len(shown)
|
||
suffix = f"（另有 {remainder} 个，详见 evidence JSON）" if remainder else ""
|
||
message = f"manifest 登记但文件不存在的产物：{', '.join(shown)}{suffix}"
|
||
limitations.append(message)
|
||
risks.append({"stage": stage, "severity": "high", "message": message})
|
||
if status not in SUCCESS_STATUSES:
|
||
risks.append({"stage": stage, "severity": "high", "message": f"阶段状态为 {status}，不得作为成功证据。"})
|
||
reviews.append({"stage": stage, "stage_name": RNASEQ_STAGE_NAMES.get(stage, stage), "limitations": limitations})
|
||
|
||
successful = [stage for stage in stages if str(manifests[stage].get("status") or "") in SUCCESS_STATUSES]
|
||
return {
|
||
"schema": "rnaseq_manifest_evidence_v1",
|
||
"metadata": {
|
||
"prefix": prefix,
|
||
"rnaseq_scope_id": selection.rnaseq_scope_id,
|
||
"discovered_scope_ids": list(selection.discovered_scope_ids),
|
||
"included_stages": stages,
|
||
"successful_stages": successful,
|
||
"interpretation_boundary": "仅陈述当前 prefix 与 RNA-seq scope 下 manifest/产物支持的事实；富集、相关性和候选排序不得写成调控因果。",
|
||
},
|
||
"stage_flow": {"visualization": " -> ".join(stages), "stages": stage_flow},
|
||
"key_findings": findings,
|
||
"stage_reviews": reviews,
|
||
"evidence_cards": cards,
|
||
"risk_flags": risks,
|
||
"major_claims": [],
|
||
"biological_story": (
|
||
f"当前报告汇总了 {len(successful)} 个成功或部分成功的 RNA-seq 阶段。"
|
||
"跨阶段生物学解释必须以各阶段数值、图表和风险标记为边界。"
|
||
),
|
||
"next_steps": ["补齐缺失或失败阶段后重新生成报告，以刷新证据快照。"],
|
||
}
|