89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Summary Skill - 综合报告生成技能"""
|
||
|
|
|
||
|
|
import argparse, hashlib, json, logging, os, sys
|
||
|
|
from datetime import datetime
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Dict, Optional
|
||
|
|
import yaml
|
||
|
|
|
||
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
class SummarySkill:
|
||
|
|
def __init__(self, config_path=None):
|
||
|
|
self.config = self._load_config(config_path)
|
||
|
|
self.commands, self.checksums = [], {}
|
||
|
|
|
||
|
|
def _load_config(self, config_path):
|
||
|
|
if config_path and os.path.exists(config_path):
|
||
|
|
with open(config_path) as f: return yaml.safe_load(f)
|
||
|
|
return {"skill": {"name": "summary", "version": "1.0.0"}, "parameters": {}, "dependencies": {"conda": {"channels": ["bioconda"], "packages": ["pandas"]}}}
|
||
|
|
|
||
|
|
def _compute_checksum(self, filepath):
|
||
|
|
sha256 = hashlib.sha256()
|
||
|
|
with open(filepath, "rb") as f:
|
||
|
|
for chunk in iter(lambda: f.read(8192), b""): sha256.update(chunk)
|
||
|
|
return sha256.hexdigest()
|
||
|
|
|
||
|
|
def _record_file(self, filepath, file_type="output"):
|
||
|
|
if filepath.exists():
|
||
|
|
self.checksums[str(filepath)] = {"sha256": self._compute_checksum(filepath), "size": filepath.stat().st_size, "type": file_type}
|
||
|
|
|
||
|
|
def generate_report(self, output_dir):
|
||
|
|
report_file = output_dir / "report.md"
|
||
|
|
content = f"""# 基因家族分析综合报告
|
||
|
|
|
||
|
|
## 分析概要
|
||
|
|
|
||
|
|
本报告汇总了基因家族分析的完整结果，包括：
|
||
|
|
- 基因家族成员鉴定 (Identify)
|
||
|
|
- 系统发育分析 (Phylogeny)
|
||
|
|
- 保守基序分析 (Motif)
|
||
|
|
- 蛋白结构域分析 (Domain)
|
||
|
|
- 基因结构分析 (Structure)
|
||
|
|
- 启动子分析 (Promoter)
|
||
|
|
- 共线性分析 (Synteny)
|
||
|
|
- 核型分析 (Karyotype)
|
||
|
|
- 物种间比较 (Interspecies)
|
||
|
|
- 表达分析 (Expression)
|
||
|
|
|
||
|
|
## 基本信息
|
||
|
|
- **分析时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}
|
||
|
|
- **技能版本**: {self.config["skill"]["version"]}
|
||
|
|
|
||
|
|
## 详细结果
|
||
|
|
|
||
|
|
请参考各分析阶段的独立报告。
|
||
|
|
|
||
|
|
## 可复现性
|
||
|
|
|
||
|
|
所有分析命令已记录在 `commands.sh` 中，环境配置见 `environment.yml`。
|
||
|
|
"""
|
||
|
|
report_file.write_text(content)
|
||
|
|
self._record_file(report_file)
|
||
|
|
return report_file
|
||
|
|
|
||
|
|
def write_reproducibility_bundle(self, output_dir):
|
||
|
|
(output_dir / "commands.sh").write_text("#!/bin/bash\n# 综合报告\n")
|
||
|
|
(output_dir / "environment.yml").write_text("name: summary\nchannels: [bioconda]\ndependencies: [pandas]")
|
||
|
|
with open(output_dir / "checksums.sha256", "w") as f:
|
||
|
|
for filepath, info in self.checksums.items(): f.write(f"{info['sha256']} {filepath}\n")
|
||
|
|
|
||
|
|
def run_demo(self, output_dir):
|
||
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
|
report = self.generate_report(output_dir)
|
||
|
|
self.write_reproducibility_bundle(output_dir)
|
||
|
|
return {"status": "success", "output_dir": str(output_dir), "report": str(report)}
|
||
|
|
|
||
|
|
def main():
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--output", "-o", type=Path, required=True)
|
||
|
|
parser.add_argument("--demo", action="store_true")
|
||
|
|
args = parser.parse_args()
|
||
|
|
skill = SummarySkill()
|
||
|
|
result = skill.run_demo(args.output)
|
||
|
|
print(json.dumps(result, indent=2))
|
||
|
|
return 0
|
||
|
|
|
||
|
|
if __name__ == "__main__": sys.exit(main())
|