admin/PlantAI-rag-new
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(bioflow): 样本配置智能分析——描述文件优先+实况分析+解压

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
admin
2026-09-16 19:54:21 +08:00
co-authored by Claude Fable 5
parent 750e218541
commit f2c101ec65
3 changed files with 262 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bioflow_service/app/routers/biojobs.py
+17 -4
View File
@@ -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)
bioflow_service/app/services/bioflow/runner.py
+244 -28
View File
@@ -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:
bioflow_service/tests/test_api.py
+1 -1
View File
@@ -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()