fix(bioflow): 描述文件候选择优+去除登记文件数上限

- 登记目录不再截断 200 文件上限(用户原始数据目录可能很大)
- 描述文件发现改为候选全收→逐份解析→按样本条目数择优(修复
  此前选到无样本语义 tsv 导致语义丢失);tsv/csv 平手优先
- samples.yaml 保留描述文件语义(样本名+分组字段),路径经实况校正
- 176 rice 目录终验:登记 244 文件,样本名 HH11_Control_0h_rep1 等
  分组语义保留,r1/r2 指向真实 fastq;任务 running
- 340 用例 + 4 场景全绿

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-16 20:03:49 +08:00
co-authored by Claude Fable 5
parent f2c101ec65
commit 36d3d8d62a
2 changed files with 96 additions and 79 deletions
-2
View File
@@ -265,8 +265,6 @@ async def _register_path(record: Any, server_path: str, settings: Any) -> dict[s
if _blocked_reason(p, settings):
continue
files.append(p)
if len(files) >= settings.max_register_files_per_job:
break
if not files:
return _err("VALIDATION",
f"目录内无可登记文件(允许扩展名 {settings.upload_ext_allow[:80]}…)", 400)
+96 -77
View File
@@ -112,6 +112,75 @@ def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = N
import re as _re
import yaml as _yaml
# 描述文件解析函数(候选择优用;定义在循环前供嵌套闭包使用)
def _parse_tsv_csv(desc: Path, seq_exts_local: tuple) -> list[dict]:
import csv as _csv
out: list[dict] = []
with desc.open(encoding="utf-8", errors="replace", newline="") as fh:
reader = _csv.DictReader(
fh, delimiter="\t" if desc.name.lower().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:
for v in row.values():
if str(v).strip().lower().endswith(seq_exts_local):
if not r1:
r1 = str(v).strip()
elif not r2:
r2 = str(v).strip()
if r1 or r2 or run:
out.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")},
})
return out
def _parse_xlsx(desc: Path) -> list[dict]:
import openpyxl as _xl
out: list[dict] = []
wb = _xl.load_workbook(desc, read_only=True, data_only=True)
try:
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
out.append({
"name": run, "run": run, "r1": "", "r2": "",
"meta": {k: v for k, v in d.items() if k != "run"},
})
finally:
wb.close()
return out
def _parse_yaml(desc: Path) -> list[dict]:
raw = _yaml.safe_load(desc.read_text(encoding="utf-8", errors="replace"))
out: list[dict] = []
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:
out.append({
"name": str(sname), "run": run, "r1": r1, "r2": r2, "meta": {},
})
return out
yaml_path = job_input_dir / "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()]
@@ -171,92 +240,42 @@ def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = N
seq_files.append(p)
seq_files = sorted({str(p): p for p in seq_files}.values(), key=lambda p: str(p))
# ---- 1. 描述文件发现(限深度 4,先源目录后 input) ----
# ---- 1. 描述文件发现:候选全收 → 逐份解析 → 按解析出的样本条目数择优 ----
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
candidates: list[Path] = []
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((".tsv", ".csv", ".xlsx", ".xls", ".yaml", ".yml")):
candidates.append(p)
# ---- 解析描述文件 → 样本清单 [{name, run, r1, r2, meta}] ----
described: list[dict] = []
if descriptor is not None:
name = descriptor.name.lower()
best: tuple[int, list[dict]] | None = None
for desc in candidates:
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 = []
if desc.name.lower().endswith((".tsv", ".csv")):
parsed = _parse_tsv_csv(desc, seq_exts)
elif desc.name.lower().endswith((".xlsx", ".xls")):
parsed = _parse_xlsx(desc)
else:
parsed = _parse_yaml(desc)
except Exception: # noqa: BLE001
parsed = []
# 样本条目数最多者胜;平手时 tsv/csv 优先(语义最全),再按路径排序稳定
is_table = desc.name.lower().endswith((".tsv", ".csv"))
if best is None or len(parsed) > best[0] or (len(parsed) == best[0] and is_table and not best_is_table):
best = (len(parsed), parsed)
best_is_table = is_table
if best and best[0] > 0:
described = best[1]
# ---- 3. 描述文件路径校正(实况为准,按 run 标识,不做变体猜测) ----
seq_by_run: dict[str, list[Path]] = {}