Validate AnVIL metadata records at load (#161)

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test lint classify classify-hprc classify-and-report download classify-bam classify-vcf classify-fastq classify-fasta classify-gfa classify-headers classify-bed coverage-report validation-report all-reports download-hprc validate-hprc clean help
+.PHONY: test lint validate-metadata classify classify-hprc classify-and-report download classify-bam classify-vcf classify-fastq classify-fasta classify-gfa classify-headers classify-bed coverage-report validation-report all-reports download-hprc validate-hprc clean help
 
 help:
 	@echo "meta-disco — AnVIL file metadata classification"
@@ -8,6 +8,7 @@ help:
 	@echo "  make classify           Run full classification pipeline (all file types, parallel)"
 	@echo "  make classify-and-report Run classify + regenerate all reports"
 	@echo "  make download           Download fresh AnVIL metadata from API"
+	@echo "  make validate-metadata  Check downloaded metadata against the record schema"
 	@echo ""
 	@echo "  make classify-bam       Classify BAM/CRAM files (network required)"
 	@echo "  make classify-vcf       Classify VCF files (network required)"
@@ -42,6 +43,9 @@ classify-and-report: classify classify-hprc all-reports
 download:
 	python scripts/download_anvil_metadata.py
 
+validate-metadata:
+	python scripts/validate_metadata.py
+
 classify-headers: classify-bam classify-vcf classify-fastq classify-fasta classify-gfa
 
 classify-bam:

scripts/validate_metadata.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Validate downloaded AnVIL metadata against the record schema.
+
+Run this after `make download` and before a multi-hour classification run: a
+renamed key or a newly-nullable column otherwise surfaces as a crash somewhere
+deep in the corpus, or as a silently wrong classification.
+
+    python scripts/validate_metadata.py
+    python scripts/validate_metadata.py --input data/anvil/anvil_files_metadata.json
+
+Exit code 1 when the records do not match the schema.
+"""
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.meta_disco.metadata_schema import (  # noqa: E402
+    ANVIL_RECORD_SCHEMA,
+    validate_anvil_records,
+)
+
+DEFAULT_INPUT = Path("data/anvil/anvil_files_metadata.json")
+
+
+def load_records(path: Path) -> list[dict]:
+    """Read the records array from a metadata file (JSON or NDJSON)."""
+    if path.suffix == ".ndjson":
+        with open(path) as f:
+            return [json.loads(line) for line in f if line.strip()]
+
+    with open(path) as f:
+        data = json.load(f)
+    if not isinstance(data, dict):
+        raise TypeError(f"Expected a JSON object, got {type(data).__name__}")
+    records = data.get("files") or data.get("results")
+    if records is None:
+        raise ValueError("JSON object must contain a 'files' or 'results' key")
+    return records
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input", type=Path, default=DEFAULT_INPUT,
+                        help=f"Metadata file to check (default: {DEFAULT_INPUT})")
+    args = parser.parse_args()
+
+    if not args.input.is_file():
+        print(f"No such file: {args.input}", file=sys.stderr)
+        return 1
+
+    records = load_records(args.input)
+    report = validate_anvil_records(records)
+
+    print(f"Schema: {len(ANVIL_RECORD_SCHEMA)} fields")
+    print(f"Input : {args.input}")
+    print(report.summary())
+
+    if report.is_valid:
+        return 0
+
+    print(
+        "\nThe AnVIL record shape has changed, or the download is incomplete.\n"
+        "Fix the schema in src/meta_disco/metadata_schema.py if the change is\n"
+        "intended — do NOT coerce the values, or the drift is hidden rather than\n"
+        "caught.",
+        file=sys.stderr,
+    )
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/meta_disco/metadata_schema.py

@@ -0,0 +1,173 @@
+"""Shape validation for AnVIL file-metadata records.
+
+Nothing between the AnVIL API and the classifiers checks the shape of a record.
+A renamed key, a newly-nullable column, or a stringified size flows silently into
+758k records and surfaces as a crash — or a wrong classification — somewhere deep
+in a multi-hour run.
+
+Two contracts, deliberately separate:
+
+* :data:`ANVIL_RECORD_SCHEMA` — the full 12-field record the download script
+  produces. Checked by ``scripts/validate_metadata.py`` and at download time.
+* :func:`validate_pipeline_records` — the far smaller promise ``ClassifyPipeline``
+  relies on. Derived inputs (a filtered subset, a hand-built test fixture) carry
+  only a few fields, so requiring the full AnVIL record here would be wrong.
+
+Nothing here mutates a record. A ``None`` is never coerced to ``""``: that would
+hide the drift this exists to catch, and it is what makes ``rec.get(k, "")``
+dangerous — the default fires only when the key is *absent*, so a present-but-null
+value returns ``None`` and crashes the first ``.lower()`` downstream.
+"""
+
+import re
+from collections import Counter
+from dataclasses import dataclass, field
+
+MD5_PATTERN = re.compile(r"^[0-9a-f]{32}$")
+
+# Bounded so a systematically broken download reports a summary, not 758k lines.
+MAX_SAMPLES_PER_PROBLEM = 3
+
+
+@dataclass(frozen=True)
+class FieldSpec:
+    """The contract for one record field."""
+
+    types: tuple[type, ...]
+    nullable: bool = False
+    allow_empty: bool = False
+    pattern: re.Pattern | None = None
+    min_value: int | None = None
+
+
+# Measured against the 758,658-record corpus. Two constraints are easy to get
+# wrong and are called out rather than inferred:
+#   * file_size may be 0 — three records are. `min_value=0`, not 1.
+#   * data_modality and reference_assembly are AnVIL's own declarations and are
+#     null for ~99% of files. They are nullable by design, not by accident.
+ANVIL_RECORD_SCHEMA: dict[str, FieldSpec] = {
+    "entry_id": FieldSpec((str,)),
+    "file_id": FieldSpec((str,)),
+    "file_name": FieldSpec((str,)),
+    "file_format": FieldSpec((str,)),
+    "file_size": FieldSpec((int,), min_value=0),
+    "file_md5sum": FieldSpec((str,), pattern=MD5_PATTERN),
+    "drs_uri": FieldSpec((str,)),
+    "dataset_id": FieldSpec((str,)),
+    "dataset_title": FieldSpec((str,)),
+    "is_supplementary": FieldSpec((bool,)),
+    "data_modality": FieldSpec((str,), nullable=True),
+    "reference_assembly": FieldSpec((str,), nullable=True),
+}
+
+# Fields ClassifyPipeline reads. Each may be absent (a derived input need not
+# carry it) but must never be present-and-null, which is the failure this guards.
+PIPELINE_STRING_FIELDS = ("file_md5sum", "file_name", "file_format", "dataset_title")
+
+
+@dataclass
+class ValidationReport:
+    """Problems found, counted by kind, with a bounded sample of each."""
+
+    total: int = 0
+    counts: Counter = field(default_factory=Counter)
+    samples: dict[str, list[str]] = field(default_factory=dict)
+
+    @property
+    def is_valid(self) -> bool:
+        return not self.counts
+
+    @property
+    def problem_records(self) -> int:
+        """Number of distinct problems recorded, not of distinct records.
+
+        One record may contribute several problems, so this is an upper bound on
+        the number of bad records — it is a signal of scale, not a count.
+        """
+        return sum(self.counts.values())
+
+    def add(self, kind: str, detail: str) -> None:
+        self.counts[kind] += 1
+        bucket = self.samples.setdefault(kind, [])
+        if len(bucket) < MAX_SAMPLES_PER_PROBLEM:
+            bucket.append(detail)
+
+    def summary(self) -> str:
+        if self.is_valid:
+            return f"{self.total:,} records, no problems."
+        lines = [f"{self.total:,} records, {len(self.counts)} kinds of problem:"]
+        for kind, n in self.counts.most_common():
+            lines.append(f"  {kind}: {n:,}")
+            for detail in self.samples.get(kind, []):
+                lines.append(f"      e.g. {detail}")
+        return "\n".join(lines)
+
+
+def _check_value(name: str, value, spec: FieldSpec) -> str | None:
+    """The problem this value has against its spec, or None."""
+    if value is None:
+        return None if spec.nullable else "null"
+
+    # bool is a subclass of int, so an `is_supplementary` in a `file_size` slot
+    # would silently pass an isinstance(int) check.
+    if isinstance(value, bool) and bool not in spec.types:
+        return "type"
+    if not isinstance(value, spec.types):
+        return "type"
+
+    if isinstance(value, str):
+        if not value and not spec.allow_empty:
+            return "empty"
+        if spec.pattern and not spec.pattern.match(value):
+            return "pattern"
+    if spec.min_value is not None and value < spec.min_value:
+        return "range"
+    return None
+
+
+def validate_anvil_records(
+    records: list[dict], schema: dict[str, FieldSpec] = ANVIL_RECORD_SCHEMA
+) -> ValidationReport:
+    """Validate records against the full AnVIL contract.
+
+    Unknown fields are *not* a problem: the API may add a column, and that should
+    not fail a download. A renamed column shows up as `missing:<old_name>`.
+    """
+    report = ValidationReport(total=len(records))
+    for i, rec in enumerate(records):
+        if not isinstance(rec, dict):
+            report.add("not_an_object", f"record {i}: {type(rec).__name__}")
+            continue
+        for name, spec in schema.items():
+            if name not in rec:
+                report.add(f"missing:{name}", f"record {i}")
+                continue
+            problem = _check_value(name, rec[name], spec)
+            if problem:
+                report.add(f"{problem}:{name}", f"record {i}: {rec[name]!r}")
+    return report
+
+
+def validate_pipeline_records(records: list[dict]) -> ValidationReport:
+    """Validate the small promise ClassifyPipeline relies on.
+
+    Each of PIPELINE_STRING_FIELDS may be absent, but must not be present-and-null:
+    `rec.get("file_name", "")` returns None for a null, and the first `.lower()` or
+    slice downstream then raises. `file_size`, when present, must be a non-negative
+    int (or null — some derived inputs omit the size rather than the key).
+    """
+    report = ValidationReport(total=len(records))
+    for i, rec in enumerate(records):
+        if not isinstance(rec, dict):
+            report.add("not_an_object", f"record {i}: {type(rec).__name__}")
+            continue
+        for name in PIPELINE_STRING_FIELDS:
+            if name in rec and not isinstance(rec[name], str):
+                kind = "null" if rec[name] is None else "type"
+                report.add(f"{kind}:{name}", f"record {i}: {rec[name]!r}")
+        size = rec.get("file_size")
+        if size is not None and (isinstance(size, bool) or not isinstance(size, int)):
+            report.add("type:file_size", f"record {i}: {size!r}")
+        elif isinstance(size, int) and not isinstance(size, bool) and size < 0:
+            report.add("range:file_size", f"record {i}: {size!r}")
+    return report

src/meta_disco/pipeline.py

@@ -14,6 +14,7 @@
 
 from .fetchers import FetchError
 from .header_classifier import classify_without_content
+from .metadata_schema import validate_pipeline_records
 
 
 @dataclass(frozen=True)
@@ -204,21 +205,35 @@ def classify_single(
     # --- Internal ---
 
     def _load_input(self) -> list[dict]:
-        """Load NDJSON or JSON input, extracting records array."""
+        """Load NDJSON or JSON input, extract the records array, and validate it.
+
+        Validates only the fields this pipeline reads (``validate_pipeline_records``),
+        not the full AnVIL record: a derived input may legitimately carry a subset.
+        Raises rather than dropping the offenders — a dropped record is
+        indistinguishable from a file that was never seen (see #155).
+        """
         with open(self.input_path) as f:
             if self.input_path.suffix == ".ndjson":
-                return [json.loads(line) for line in f if line.strip()]
-            data = json.load(f)
-            if not isinstance(data, dict):
-                raise TypeError(
-                    f"Expected JSON object with 'results' key, got {type(data).__name__}"
-                )
-            records = data.get("results") or data.get("files")
-            if records is None:
-                raise ValueError(
-                    "JSON object must contain a 'results' or 'files' key"
-                )
-            return records
+                records = [json.loads(line) for line in f if line.strip()]
+            else:
+                data = json.load(f)
+                if not isinstance(data, dict):
+                    raise TypeError(
+                        f"Expected JSON object with 'results' key, got {type(data).__name__}"
+                    )
+                records = data.get("results") or data.get("files")
+                if records is None:
+                    raise ValueError(
+                        "JSON object must contain a 'results' or 'files' key"
+                    )
+
+        report = validate_pipeline_records(records)
+        if not report.is_valid:
+            raise ValueError(
+                f"{self.input_path} holds records this pipeline cannot read.\n"
+                f"{report.summary()}"
+            )
+        return records
 
     def _filter_records(self, records: list[dict]) -> list[dict]:
         """Filter to records matching file_type.extensions with valid MD5."""