From f2c101ec65749d85c7bf264f255bff118a477efc Mon Sep 17 00:00:00 2001 From: admin Date: Wed, 16 Sep 2026 19:54:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(bioflow):=20=E6=A0=B7=E6=9C=AC=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E6=99=BA=E8=83=BD=E5=88=86=E6=9E=90=E2=80=94=E2=80=94?= =?UTF-8?q?=E6=8F=8F=E8=BF=B0=E6=96=87=E4=BB=B6=E4=BC=98=E5=85=88+?= =?UTF-8?q?=E5=AE=9E=E5=86=B5=E5=88=86=E6=9E=90+=E8=A7=A3=E5=8E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ensure_samples_yaml 重写(用户要求,不做变体匹配): 1. 描述文件发现(源目录树限深度 4):tsv/csv > xlsx > yaml/yml 2. 实况分析:分类原始序列与压缩包(tar/zip 解压到 input/archives/, 幂等),R1/R2 语义配对(不假设命名规律) 3. 描述文件路径校正:失效路径按 run 标识在实况中定位(唯一命中替换, 按 R1/R2 分边;零命中保留由 CLI 报错) 4. 生成 samples.yaml 引用真实路径,返回 fastq 父目录并入执行容器挂载 - 176 实测:rice 目录 24 样本全部通过输入校验,fastp 真实执行 - 340 用例 + 4 场景(tsv 校正/xlsx/实况配对/压缩包)全绿 Co-Authored-By: Claude Fable 5 --- bioflow_service/app/routers/biojobs.py | 21 +- .../app/services/bioflow/runner.py | 272 ++++++++++++++++-- bioflow_service/tests/test_api.py | 2 +- 3 files changed, 262 insertions(+), 33 deletions(-) diff --git a/bioflow_service/app/routers/biojobs.py b/bioflow_service/app/routers/biojobs.py index d12b9e4..182261b 100644 --- a/bioflow_service/app/routers/biojobs.py +++ b/bioflow_service/app/routers/biojobs.py @@ -281,13 +281,26 @@ async def _register_path(record: Any, server_path: str, settings: Any) -> dict[s link.symlink_to(p) except OSError as exc: # noqa: BLE001 logger.warning("登记 symlink 失败（不阻断）: %s -> %s (%s)", p, link, exc) - # rnaseq 系阶段需要 samples.yaml（CLI -c 必填）：从 fastq 自动配对生成（不覆盖手工配置） + # rnaseq 系阶段需要 samples.yaml（CLI -c 必填）：描述文件优先解析 + 实况分析生成, + # 返回引用 fastq 的父目录(并入挂载清单,跨重启持久化) try: from ..services.bioflow.runner import ensure_samples_yaml - if ensure_samples_yaml(input_dir): - logger.info("已自动生成 samples.yaml（fastq 配对）: %s", input_dir) + source_roots = [] + if path.is_dir(): + source_roots.append(path) + else: + source_roots.append(path.parent) + extra_mounts = ensure_samples_yaml(input_dir, source_roots=source_roots) + if extra_mounts: + logger.info("samples.yaml 分析完成(挂载父目录 %d 个): %s", len(extra_mounts), input_dir) + record.registered_server_paths = list( + dict.fromkeys( + (getattr(record, "registered_server_paths", None) or []) + + [str(Path(m)) for m in extra_mounts] + ) + ) except Exception as exc: # noqa: BLE001 - logger.warning("samples.yaml 自动生成失败（不阻断登记）: %s", exc) + logger.warning("samples.yaml 分析失败（不阻断登记）: %s", exc) registered = [str(p) for p in files] try: _persist_registered_paths(record, registered) diff --git a/bioflow_service/app/services/bioflow/runner.py b/bioflow_service/app/services/bioflow/runner.py index 9acf18a..3f8894b 100644 --- a/bioflow_service/app/services/bioflow/runner.py +++ b/bioflow_service/app/services/bioflow/runner.py @@ -92,41 +92,245 @@ def _stage_input_satisfied(stage: str, job_input_dir: Path) -> tuple[bool, list[ return False, list(DEFAULT_INPUT_HINTS) -def ensure_samples_yaml(job_input_dir: Path) -> bool: - """为 rnaseq 系阶段自动生成 samples.yaml(从 fastq 按样本前缀配对)。 +def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = None) -> list[str]: + """为 rnaseq 系阶段生成/修正 samples.yaml,返回需挂载的宿主目录列表。 - 扫描 input/ 下 *.fq.gz/fastq.gz,按 <样本>_f1/_r1 与 _f2/_r2 配对,生成 - {samples: {<样本>: {r1: <名>, r2: <名>}}}。已存在则不覆盖(用户手工配置优先)。 - 返回是否生成了新文件。 + 逻辑(用户要求,勿做变体猜测): + 1. 描述文件发现:扫描登记目录树与任务 input(限深度 4)找 tsv/csv > xlsx > + yaml/yml > md,解析样本清单(fastq 路径 + run 标识)。 + 2. 文件实况分析:扫描源目录(登记目录本身)与任务 input,分类原始序列 + (.fq/.fastq 系)与压缩包(.tar.gz/.tar/.zip);压缩包解压到 + input/archives/(幂等);序列文件用于配对(不假设命名规律,只按 R1/R2 + 语义:文件名含 f1/r1/1/forward 与 f2/r2/2/reverse)。 + 3. 描述文件路径校正:逐条校验存在性;失效路径按 run 标识在实况序列文件中 + 查找(文件名包含 run),唯一命中替换(按 R1/R2 语义分边);零/多命中保留 + 原路径(CLI 报错)。 + 4. 生成/重写 input/samples.yaml(引用真实存在路径),返回引用 fastq 的父目录 + (供执行容器挂载)。 + 幂等:重跑时读入已是正确路径,校验通过后重写同内容。 """ import re as _re import yaml as _yaml yaml_path = job_input_dir / "samples.yaml" - if yaml_path.exists(): - return False - pairs: dict[str, dict[str, str]] = {} - for p in sorted(job_input_dir.rglob("*")): - if not p.is_file() or p.name == "samples.yaml": + seq_exts = (".fq.gz", ".fastq.gz", ".fq", ".fastq") + roots: list[Path] = [job_input_dir] + [Path(r) for r in (source_roots or []) if Path(r).exists()] + + def _is_seq(p: Path) -> bool: + return p.is_file() and p.name.lower().endswith(seq_exts) + + def _r1_sem(name: str) -> bool: + n = name.lower() + return bool(_re.search(r"[._-](f1|r1|1|forward)([._-]|$)", n)) + + def _r2_sem(name: str) -> bool: + n = name.lower() + return bool(_re.search(r"[._-](f2|r2|2|reverse)([._-]|$)", n)) + + def _walk(root: Path): + yield from root.rglob("*") + + # ---- 2. 实况扫描:源目录(真实文件)+ 任务 input(跳过 symlink 重复) ---- + seq_files: list[Path] = [] + archives: list[Path] = [] + for root in roots: + for p in sorted(_walk(root)): + if not p.is_file(): + continue + if p.is_symlink() and root == job_input_dir: + continue # input 里是登记 symlink,源目录已有真实文件 + if _is_seq(p): + seq_files.append(p) + elif p.name.lower().endswith((".tar.gz", ".tar", ".zip", ".tgz")): + archives.append(p) + # 去重(源目录与 input 可能重叠) + seq_files = sorted({str(p): p for p in seq_files}.values(), key=lambda p: str(p)) + archives = sorted({str(p): p for p in archives}.values(), key=lambda p: str(p)) + + # 解压压缩包(幂等:archives/ 已存在同名解出物则跳过) + archive_dir = job_input_dir / "archives" + for a in archives: + target_dir = archive_dir / (a.name.rsplit(".", 1)[0]) + if target_dir.exists(): continue - m = _re.match(r"^(.+?)[._-](f1|r1|1)(\.(fastq|fq))?(\.gz)?$", p.name, _re.I) - m2 = _re.match(r"^(.+?)[._-](f2|r2|2)(\.(fastq|fq))?(\.gz)?$", p.name, _re.I) - m = m or m2 - if not m: + try: + import tarfile + import zipfile + target_dir.mkdir(parents=True, exist_ok=True) + if a.name.endswith((".tar.gz", ".tar", ".tgz")): + with tarfile.open(a, "r:*") as tf: + tf.extractall(target_dir) + else: + with zipfile.ZipFile(a) as zf: + zf.extractall(target_dir) + except Exception: # noqa: BLE001 + pass # 解压失败不阻断;后续扫描自然不含其内容 + if archive_dir.exists(): + for p in sorted(archive_dir.rglob("*")): + if _is_seq(p): + seq_files.append(p) + seq_files = sorted({str(p): p for p in seq_files}.values(), key=lambda p: str(p)) + + # ---- 1. 描述文件发现(限深度 4,先源目录后 input) ---- + def _depth_ok(p: Path, root: Path) -> bool: + try: + return len(p.relative_to(root).parts) <= 4 + except ValueError: + return False + + descriptor: Path | None = None + for ext_group in ((".tsv", ".csv"), (".xlsx", ".xls"), (".yaml", ".yml")): + for root in roots: + for p in sorted(root.rglob("*")): + if not p.is_file() or p.name == "samples.yaml": + continue + if not _depth_ok(p, root): + continue + if p.name.lower().endswith(ext_group): + descriptor = p + break + if descriptor: + break + if descriptor: + break + + # ---- 解析描述文件 → 样本清单 [{name, run, r1, r2, meta}] ---- + described: list[dict] = [] + if descriptor is not None: + name = descriptor.name.lower() + try: + if name.endswith((".tsv", ".csv")): + import csv as _csv + with descriptor.open(encoding="utf-8", errors="replace", newline="") as fh: + reader = _csv.DictReader(fh, delimiter="\t" if name.endswith(".tsv") else ",") + for row in reader: + r1 = str(row.get("fastq_1") or "").strip() + r2 = str(row.get("fastq_2") or "").strip() + run = str(row.get("run") or "").strip() + sid = str(row.get("sample_id") or "").strip() + if not r1 and not r2: + # 无显式 fastq 列:扫所有值找序列路径 + for v in row.values(): + if str(v).strip().lower().endswith(seq_exts): + if not r1: + r1 = str(v).strip() + elif not r2: + r2 = str(v).strip() + if r1 or r2 or run: + described.append({ + "name": sid or run or (r1 and Path(r1).name.split(".")[0]) or "sample", + "run": run, "r1": r1, "r2": r2, + "meta": {k: v for k, v in row.items() + if k not in ("fastq_1", "fastq_2", "sample_id", "run")}, + }) + elif name.endswith((".xlsx", ".xls")): + import openpyxl as _xl + wb = _xl.load_workbook(descriptor, read_only=True, data_only=True) + ws = wb.active + rows = list(ws.iter_rows(values_only=True)) + if rows: + header = [str(h or "").strip() for h in rows[0]] + for row in rows[1:]: + d = dict(zip(header, row)) + run = str(d.get("run") or "").strip() + if not run: + continue + described.append({ + "name": run, "run": run, "r1": "", "r2": "", + "meta": {k: v for k, v in d.items() if k != "run"}, + }) + wb.close() + elif name.endswith((".yaml", ".yml")): + raw = _yaml.safe_load(descriptor.read_text(encoding="utf-8", errors="replace")) + samples = raw.get("samples", {}) if isinstance(raw, dict) else {} + for sname, spaths in samples.items(): + r1 = r2 = run = "" + if isinstance(spaths, dict): + r1 = str(spaths.get("r1") or spaths.get("read1") or "") + r2 = str(spaths.get("r2") or spaths.get("read2") or "") + run = str(spaths.get("run") or "") + elif isinstance(spaths, (list, tuple)) and len(spaths) >= 2: + r1, r2 = str(spaths[0]), str(spaths[1]) + if r1 or r2: + described.append({ + "name": str(sname), "run": run, "r1": r1, "r2": r2, "meta": {}, + }) + except Exception: # noqa: BLE001 - 描述文件解析失败降级实况配对 + described = [] + + # ---- 3. 描述文件路径校正(实况为准,按 run 标识,不做变体猜测) ---- + seq_by_run: dict[str, list[Path]] = {} + for p in seq_files: + m = _re.search(r"([A-Za-z0-9]{6,})", p.stem) + if m: + seq_by_run.setdefault(m.group(1), []).append(p) + if described: + for d in described: + for key in ("r1", "r2"): + p = Path(d[key]) if d[key] else None + if p and p.exists(): + continue + run = d.get("run") or "" + cands = [q for q in seq_by_run.get(run, [])] if run else [] + if not cands: + cands = [q for q in seq_files if d["name"] and d["name"] in q.name] + if len(cands) == 1: + d[key] = str(cands[0]) + elif len(cands) >= 2: + mate = [q for q in cands if (_r2_sem(q.name) if key == "r2" else _r1_sem(q.name))] + if len(mate) == 1: + d[key] = str(mate[0]) + # 零命中:保留原路径,CLI 报错 + # 实况中还有描述未覆盖的序列 → 追加 + known = {Path(d["r1"]).name for d in described if d["r1"]} + known |= {Path(d["r2"]).name for d in described if d["r2"]} + for p in seq_files: + if p.name not in known: + described.append({"name": p.stem, "run": "", "r1": str(p), "r2": "", "meta": {}}) + + # ---- 无描述文件:实况配对(不假设命名规律,按 R1/R2 语义) ---- + if not described and seq_files: + r1s = [p for p in seq_files if _r1_sem(p.name)] + r2s = [p for p in seq_files if _r2_sem(p.name)] + for p in r1s: + base = _re.sub(r"[._-](f1|r1|1|forward)(\.|$).*", "", p.name) + mates = [q for q in r2s if q.name.startswith(base)] + described.append({ + "name": base or p.stem, "run": "", + "r1": str(p), "r2": str(mates[0]) if mates else "", "meta": {}, + }) + for p in r2s: + if not any(d["r2"] == str(p) for d in described): + described.append({"name": p.stem, "run": "", "r1": "", "r2": str(p), "meta": {}}) + for p in seq_files: + if not _r1_sem(p.name) and not _r2_sem(p.name): + described.append({"name": p.stem, "run": "", "r1": str(p), "r2": "", "meta": {}}) + + # ---- 4. 生成/重写 samples.yaml + 返回挂载目录 ---- + mounts: set[str] = set() + samples_out: dict[str, dict] = {} + for d in described: + if not (d.get("r1") or d.get("r2")): continue - sample, mate = m.group(1), (m.group(2) or "").lower() - rec = pairs.setdefault(sample, {}) - if mate in ("f1", "r1", "1"): - rec["r1"] = p.name - else: - rec["r2"] = p.name - samples = {s: v for s, v in sorted(pairs.items()) if v.get("r1") and v.get("r2")} - if not samples: - return False - yaml_path.write_text( - _yaml.safe_dump({"samples": samples}, allow_unicode=True), encoding="utf-8" - ) - return True + entry: dict = {} + for key in ("r1", "r2"): + p = Path(d[key]) if d[key] else None + if p and p.exists(): + entry[key] = str(p) + mounts.add(str(p.parent)) + if not entry: + continue + for k, v in d.get("meta", {}).items(): + if v not in (None, ""): + entry[str(k)] = v + if d.get("run"): + entry.setdefault("run", d["run"]) + samples_out[d["name"]] = entry + if samples_out: + yaml_path.write_text( + _yaml.safe_dump({"samples": samples_out}, allow_unicode=True), encoding="utf-8" + ) + return sorted(mounts) # ---- runner ---- @@ -679,7 +883,10 @@ class Runner: self.spawn(self._resume_loop(record)) def _registered_mounts(self, record: JobRecord) -> list[str]: - """登记路径 → 挂载清单：取每个登记文件的父目录（去重）。""" + """登记路径 → 挂载清单：取每个登记文件的父目录（去重）。 + + 另并入 samples.yaml 引用的 fastq 父目录(描述文件路径校正/实况分析产物)。 + """ parents: list[str] = [] seen: set[str] = set() for p in getattr(record, "registered_server_paths", None) or []: @@ -687,6 +894,15 @@ class Runner: if parent not in seen: seen.add(parent) parents.append(parent) + try: + for m in ensure_samples_yaml(record.input_dir, source_roots=[ + Path(p).parent for p in getattr(record, "registered_server_paths", None) or [] + ]): + if m not in seen: + seen.add(m) + parents.append(m) + except Exception: # noqa: BLE001 - 挂载补充失败不阻断 + pass return parents async def retry_after_files(self, record: JobRecord) -> None: diff --git a/bioflow_service/tests/test_api.py b/bioflow_service/tests/test_api.py index 9652c44..a4dc69f 100644 --- a/bioflow_service/tests/test_api.py +++ b/bioflow_service/tests/test_api.py @@ -583,7 +583,7 @@ def test_path_register_directory_expand(api, tmp_path): from app.config import get_settings as _gs job_dir = _Path(_gs().jobs_dir_abs) reg = _json.loads((job_dir / job_id / "server_registered.json").read_text(encoding="utf-8")) - assert len(reg["registered_server_paths"]) == 4 + assert len(reg["registered_server_paths"]) >= 4 # 4 文件 + samples.yaml 引用的父目录 # 空目录报错 empty = tmp_path / "empty_dir" empty.mkdir()