集成层模块

services/integration/ 包含流水线、报表、工作流等集成功能。

---

模块列表

模块	文件	功能	状态
工作流引擎	workflow_engine.py	任务编排、执行	✅
批处理	batch_processing.py	批量任务管理	✅
自动报告	auto_report_service.py	报告自动生成	✅
报告生成器	report_generator.py	PDF/HTML 报告	✅
医疗报告服务	medical_report_service.py	医疗报告模板	✅
导出服务	export_service.py	CSV/Excel/JSON 导出	✅
仪表盘	reporting.py	可视化仪表盘	✅
热图分析	heatmap.py	热图生成	✅
代谢分析	metabolism.py	代谢通路分析	✅
代谢炎症分析	metabolic_inflammation.py	代谢炎症关联	✅
肠脑轴	gut_brain.py	肠脑轴分析	✅
统计分析	stats.py	统计分析集成	✅
多组学整合	multiomics_integrator.py	DIABLO/MOFA 整合	✅
PDF引擎	pdf_engine.py	PDF 渲染	✅
静态渲染	static_renderer.py	静态页面渲染	✅
GSA E2E	gsa_e2e.py	全基因组分析	✅
异步报告	async_report_processor.py	异步报告处理	✅

---

1. 工作流引擎 (workflow_engine.py)

功能

编排和执行复杂的多步骤分析任务。

流程

任务定义 → 依赖解析 → 执行调度 → 状态监控 → 结果汇总

核心类

from services.integration.workflow_engine import (
    WorkflowEngine,
    WorkflowTemplates,
    TaskType,
    TaskStatus
)

# 创建工作流
engine = WorkflowEngine()

# 使用模板
workflow = WorkflowTemplates.create_16s_analysis_workflow(
    sample_ids=["S001", "S002", "S003"]
)

# 执行
result = await engine.execute(workflow)

任务类型

class TaskType(Enum):
    DATA_PROCESSING = "data_processing"
    ANALYSIS = "analysis"
    REPORT_GENERATION = "report_generation"
    EXPORT = "export"

---

2. 批处理 (batch_processing.py)

功能

批量处理大量样本分析任务。

流程

任务列表 → 队列调度 → 并行处理 → 进度跟踪 → 结果汇总

核心类

from services.integration.batch_processing import BatchJob, BatchStatus, BatchTask

# 创建批处理任务
job = BatchJob(
    name="批量微生物组分析",
    items=[
        {"sample_id": f"S{i:04d}", "fastq": f"data/{i}.fastq"}
        for i in range(100)
    ],
    job_type="microbiome_analysis"
)

# 提交任务
job_id = create_async_task(job)

# 查询状态
status = get_task_status(job_id)

---

3. 报告生成 (report_generator.py)

功能

生成医学报告（PDF/HTML 格式）。

流程

数据输入 → 模板选择 → 内容填充 → 样式应用 → 格式输出

使用方式

from services.integration.report_generator import (
    get_styles,
    section,
    build_medical_abn_table
)

# 获取样式
styles = get_styles()

# 创建标题
title = section("微生物组分析报告", styles, level=1)

# 创建异常表格
abn_table = build_medical_abn_table(abnormalities, styles)

报告结构

报告
├── 标题
├── 患者信息
├── 检测摘要
├── 详细结果
│   ├── 物种组成
│   ├── 多样性指数
│   └── 疾病风险
├── 异常指标表
├── 建议
└── 页脚

---

4. 导出服务 (export_service.py)

功能

导出分析结果为多种格式。

支持格式

• CSV
• Excel (.xlsx)
• JSON
• PDF

核心函数

from services.integration.export_service import (
    export_to_csv,
    export_to_excel,
    export_to_json,
    export_patient_report,
    export_batch_data
)

# 导出 CSV
csv_response = export_to_csv(
    data=[{"sample": "S001", "diversity": 3.5}],
    filename="microbiome_results.csv"
)

# 导出 Excel
xlsx_response = export_to_excel(
    data=[...],
    filename="analysis_report.xlsx",
    sheet_name="Results"
)

# 批量导出
batch_response = export_batch_data(
    job_id="BATCH001",
    format="xlsx"
)

---

5. 仪表盘 (reporting.py)

功能

生成可视化仪表盘。

核心函数

from services.integration.reporting import (
    generate_line_chart,
    generate_bar_chart,
    generate_pie_chart,
    generate_heatmap,
    generate_radar_chart,
    generate_patient_dashboard,
    generate_population_dashboard
)

# 生成患者仪表盘
dashboard = generate_patient_dashboard(
    patient_data={
        "microbiome": microbiome_data,
        "metabolomics": metabolomics_data,
        "genomics": genomics_data
    }
)

# 生成人群仪表盘
pop_dashboard = generate_population_dashboard(
    population_data=[...]
)

图表类型

图表	用途
折线图	时间序列数据
柱状图	分类对比
饼图	组成比例
热图	相关性矩阵
雷达图	多维度评估
仪表图	单指标状态

---

6. 热图分析 (heatmap.py)

功能

生成相关性热图和聚类分析。

核心函数

from services.integration.heatmap import (
    calculate_correlation_matrix,
    perform_hierarchical_clustering,
    generate_heatmap_echarts,
    analyze_multi_omics_correlation
)

# 计算相关性矩阵
corr_matrix = calculate_correlation_matrix(
    data={"gene1": [...], "gene2": [...]},
    method="pearson"
)

# 生成热图
heatmap_config = generate_heatmap_echarts(
    title="多组学相关性分析",
    x_labels=["微生物", "代谢物", "基因"],
    y_labels=["微生物", "代谢物", "基因"],
    data=corr_matrix
)

---

7. 多组学整合 (multiomics_integrator.py)

功能

使用高级统计方法整合多组学数据。

支持方法

• DIABLO: 监督式多组学整合
• MOFA+: 非监督式因子分析
• CCA: 典型相关分析
• NMF: 非负矩阵分解

核心类

from services.integration.multiomics_integrator import (
    DIABLOIntegrator,
    MOFAIntegrator,
    CrossOmicsRuleEngine,
    MultiOmicsIntegrationEngine
)

# DIABLO 整合
diablo = DIABLOIntegrator()
result = diablo.integrate(
    omics_data=[microbiome_data, metabolomics_data, genomics_data],
    labels=disease_labels
)

# MOFA 整合
mofa = MOFAIntegrator()
factors = mofa.run(omics_data=[...])