#!/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())