# 基于文件夹的多物种工作流设计 ## 核心设计 通过物理文件夹结构明确区分目标物种和参考物种： ``` input/{job_id}/ ├── target/ # 目标物种（必需，唯一） │ ├── Ft.pep.fasta # 蛋白序列 │ ├── Ft.gff3 # 注释文件 │ └── Ft.genome.fasta # 基因组（可选） │ ├── references/ # 参考物种（可选，可多个） │ ├── Ath/ # 拟南芥 │ │ ├── Ath.pep.fasta │ │ └── Ath.gff3 │ ├── Os/ # 水稻 │ │ ├── Os.pep.fasta │ │ └── Os.gff3 │ └── Sl/ # 番茄 │ ├── Sl.pep.fasta │ └── Sl.gff3 │ └── optional/ # 其他可选文件 ├── SBP.hmm # HMM文件 └── custom_config.yaml # 自定义配置 ``` ## 文件夹结构说明 | 文件夹 | 用途 | 数量限制 | 必需文件 | |:---|:---|:---:|:---| | `target/` | 目标物种数据 | 1个物种 | `*.pep.fasta` 或 `*.pep.fa` | | `references/` | 参考物种数据 | 0-N个物种 | 每个子文件夹需要 `*.pep.fasta` | | `optional/` | 可选文件 | 任意 | 无 | ## 工作流程 ### 阶段1: 数据准备 (genome_prepare) **输入**: `input/{job_id}/target/` 和 `input/{job_id}/references/` **处理**: ```python def process_target_folder(target_dir: Path, output_dir: Path, prefix: str): """处理目标物种文件夹""" # 查找蛋白文件 pep_files = list(target_dir.glob("*.pep.fasta")) + list(target_dir.glob("*.pep.fa")) if not pep_files: raise FileNotFoundError("目标物种文件夹缺少蛋白文件 (*.pep.fasta)") # 处理目标物种 process_species(target_dir, output_dir, prefix, is_target=True) def process_reference_folders(ref_dir: Path, output_dir: Path): """处理参考物种文件夹""" references = [] for species_dir in ref_dir.iterdir(): if not species_dir.is_dir(): continue species_code = species_dir.name # Ath, Os, Sl, etc. # 检查必需文件 pep_files = list(species_dir.glob("*.pep.fasta")) + list(species_dir.glob("*.pep.fa")) if not pep_files: logger.warning(f"参考物种 {species_code} 缺少蛋白文件，跳过") continue # 处理参考物种 process_species(species_dir, output_dir, species_code, is_target=False) references.append(species_code) return references ``` **输出**: ``` input/{job_id}/ ├── target/ # 原始目标物种数据 ├── references/ # 原始参考物种数据 ├── processed/ # 处理后的标准化数据 │ ├── target.pep.fasta # 目标物种蛋白（重命名） │ ├── target.gff3 │ ├── Ath_ref.pep.fasta # 参考物种1（添加_ref后缀） │ ├── Ath_ref.gff3 │ ├── Os_ref.pep.fasta # 参考物种2 │ ├── Os_ref.gff3 │ └── ... └── genome_prepare.manifest.json ``` ### 阶段2: 基因家族鉴定 (identify) **输入**: - 目标蛋白: `processed/target.pep.fasta` - 参考蛋白: `processed/*_ref.pep.fasta`（多个） **处理**: ```python def run_identify_with_references( target_pep: Path, reference_peps: List[Path], hmm_file: Optional[Path], output_dir: Path ): """使用多参考物种进行基因家族鉴定""" # 1. HMM搜索目标物种 hmm_results = run_hmmsearch(target_pep, hmm_file) # 2. 合并参考物种构建BLAST数据库 if reference_peps: combined_db = combine_references(reference_peps) blast_db = build_blast_db(combined_db) # 3. BLAST搜索 blast_results = run_blastp(target_pep, blast_db) # 4. 分析参考物种来源 ref_analysis = analyze_reference_sources(blast_results) logger.info(f"参考物种匹配情况: {ref_analysis}") # 5. 整合结果 final_members = integrate_results(hmm_results, blast_results) return final_members def combine_references(reference_peps: List[Path]) -> Path: """合并多个参考物种的蛋白序列""" combined = Path("combined_ref.pep.fa") with open(combined, 'w') as out: for ref_pep in reference_peps: species_code = ref_pep.stem.replace('_ref', '') # Ath_ref -> Ath with open(ref_pep) as f: for line in f: if line.startswith('>'): # 添加物种标识 out.write(f">{species_code}|{line[1:]}") else: out.write(line) return combined ``` **输出**: ``` final/ ├── {prefix}.final_members.pep.fa # 目标物种基因家族成员 ├── {prefix}.final_members.id ├── {prefix}.identify.manifest.json ├── {prefix}.identify_report.md └── reference_analysis.json # 参考物种分析 ``` ### 阶段3: 进化分析 (phylogeny) **输入**: - 目标物种基因家族成员 - 参考物种同源基因 **处理**: 构建包含多物种的系统发育树 ## 前端UI设计 ### 上传界面 ``` ┌─────────────────────────────────────────┐ │ 📁 目标物种 (Target Species) │ │ [选择文件夹] 或 [拖拽文件夹到此处] │ │ 必需: 蛋白序列文件 (*.pep.fasta) │ │ 可选: 注释文件 (*.gff3), 基因组文件 │ ├─────────────────────────────────────────┤ │ 📁 参考物种 (Reference Species) │ │ [添加参考物种] │ │ ├─ 📁 Ath (拟南芥) [删除] │ │ │ [选择文件夹] │ │ ├─ 📁 Os (水稻) [删除] │ │ │ [选择文件夹] │ │ └─ [+ 添加更多参考物种] │ │ 可选: 用于辅助鉴定的已知物种 │ ├─────────────────────────────────────────┤ │ 📁 可选文件 (Optional) │ │ [选择文件] HMM文件, 配置文件等 │ ├─────────────────────────────────────────┤ │ [开始上传并处理] │ └─────────────────────────────────────────┘ ``` ### 参数配置 ```yaml # 自动从文件夹结构生成 analysis: target_species: "Ft" # 从target/文件夹名获取 target_prefix: "Ft.SPL" # 用户输入 reference_species: # 从references/子文件夹获取 - "Ath" - "Os" - "Sl" gene_family: "SPL" # 从prefix解析 ``` ## 实现代码 ### 文件夹扫描 ```python class FolderScanner: """扫描输入文件夹结构""" def scan(self, input_dir: Path) -> Dict[str, Any]: """扫描文件夹，返回结构信息""" result = { "target": None, "references": [], "optional": [] } # 扫描目标物种 target_dir = input_dir / "target" if target_dir.exists(): result["target"] = self.scan_species_folder(target_dir, is_target=True) # 扫描参考物种 ref_dir = input_dir / "references" if ref_dir.exists(): for species_dir in ref_dir.iterdir(): if species_dir.is_dir(): species_info = self.scan_species_folder(species_dir, is_target=False) result["references"].append(species_info) # 扫描可选文件 opt_dir = input_dir / "optional" if opt_dir.exists(): result["optional"] = list(opt_dir.iterdir()) return result def scan_species_folder(self, species_dir: Path, is_target: bool) -> Dict[str, Any]: """扫描物种文件夹""" species_code = species_dir.name files = { "code": species_code, "is_target": is_target, "pep": None, "gff": None, "genome": None, "cds": None } for file in species_dir.iterdir(): if not file.is_file(): continue name_lower = file.name.lower() if 'pep' in name_lower or 'protein' in name_lower: files["pep"] = file elif 'cds' in name_lower: files["cds"] = file elif name_lower.endswith('.gff3') or name_lower.endswith('.gff'): files["gff"] = file elif 'genome' in name_lower or 'chromosome' in name_lower: files["genome"] = file return files ``` ### 处理流程 ```python def process_folder_structure(input_dir: Path, output_dir: Path, prefix: str): """处理文件夹结构""" scanner = FolderScanner() structure = scanner.scan(input_dir) # 检查目标物种 if not structure["target"]: raise ValueError("缺少目标物种文件夹: input/{job_id}/target/") if not structure["target"]["pep"]: raise ValueError("目标物种文件夹缺少蛋白文件 (*.pep.fasta)")