diff --git a/bioflow_service/app/routers/biojobs.py b/bioflow_service/app/routers/biojobs.py index 182261b..2cd8251 100644 --- a/bioflow_service/app/routers/biojobs.py +++ b/bioflow_service/app/routers/biojobs.py @@ -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) diff --git a/bioflow_service/app/services/bioflow/runner.py b/bioflow_service/app/services/bioflow/runner.py index 3f8894b..eaddb4e 100644 --- a/bioflow_service/app/services/bioflow/runner.py +++ b/bioflow_service/app/services/bioflow/runner.py @@ -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]] = {}