src/uploads/validation.py

"""Validate-before-availability for a quarantined upload (ADR-0011 items 3, 4, 5).

A staged file stays quarantined until it clears the SAME grounding gate a
registered dataset clears:

    1. manifest schema validation  (src.datasets.manifest_schema.validate_manifest)
    2. against-data checks          (src.workflows.manifest_data_validation)

Item 4 ("never execute uploaded content") is honoured structurally: the file is
opened ONLY by a vetted reader — ``scanpy.read_h5ad`` for h5ad, ``pandas.read_csv``
for a flat matrix — never ``exec``/``eval``/``pickle``. Passing both gates flips
the record to ``validated``; failing keeps it quarantined with the errors surfaced.

Two file shapes are supported:

- **h5ad** — carries expression AND sample metadata (obs), so the full five-check
  ``validate_manifest_against_data`` runs (data level, feature IDs, metadata
  columns, group columns, contrasts).
- **flat matrix** (``.csv``/``.tsv``/``.txt`` + ``.gz``) — samples as rows, genes
  as columns (the convention ``decoupler_inspect_data`` uses). A bare matrix has
  no ``obs``, so only the two *file-content* checks that don't need metadata run
  (``data_level`` + ``feature_id_type``); the obs-dependent checks are recorded
  as skipped **caveats**, not silently passed. This verifies the file is what the
  manifest claims (correct units, correct feature IDs) — the security-relevant
  part — while stating honestly that grouping/metadata weren't validated.
"""

from __future__ import annotations

from datetime import datetime
from pathlib import Path
from typing import Any

from src.uploads.records import (
    STATUS_QUARANTINED,
    STATUS_VALIDATED,
    UploadRecord,
    persist_record,
)

# Flat-matrix suffixes (compound .gz handled in _kind_of / _read_tabular_matrix).
_TABULAR_SUFFIXES = (".csv", ".tsv", ".txt", ".csv.gz", ".tsv.gz", ".txt.gz")

# Cap the number of expression values fed to the data-level classifier, matching
# the dataset_validate_manifest_against_data tool.
_MAX_X = 50_000


def _report(status: str, errors: list[str], **extra: Any) -> dict:
    return {"status": status, "errors": errors, **extra}


def _kind_of(path: str) -> str | None:
    """Return 'h5ad' | 'tabular' | None for a staged file name."""
    name = Path(path).name.lower()
    if name.endswith(".h5ad"):
        return "h5ad"
    if name.endswith(_TABULAR_SUFFIXES):
        return "tabular"
    return None


def _cap_flat(X_flat):
    import numpy as np

    if len(X_flat) > _MAX_X:
        rng = np.random.default_rng(0)
        return rng.choice(X_flat, size=_MAX_X, replace=False)
    return X_flat


def _h5ad_report(staged_path: str, manifest_obj) -> dict:
    """Full five-check report from an h5ad, via the vetted scanpy loader."""
    import scanpy as sc

    from src.workflows.manifest_data_validation import validate_manifest_against_data

    adata = sc.read_h5ad(staged_path)  # vetted loader — never exec
    X = adata.X
    if hasattr(X, "toarray"):
        X = X.toarray()
    X_flat = _cap_flat(X.flatten())
    return validate_manifest_against_data(
        X_flat, list(adata.var.index), adata.obs.copy(), manifest_obj
    )


def _tabular_report(staged_path: str, manifest_obj) -> dict:
    """Content-only report from a flat matrix, via the vetted pandas reader.

    Samples as rows, genes as columns (index_col=0), matching
    ``decoupler_inspect_data``. Only the two checks that a bare matrix can support
    are run; the obs-dependent checks are surfaced as caveats.
    """
    import pandas as pd

    from src.workflows.manifest_data_validation import (
        check_data_level,
        check_feature_id_type,
    )

    name = Path(staged_path).name.lower()
    base = name[:-3] if name.endswith(".gz") else name
    sep = "\t" if base.endswith((".tsv", ".txt")) else ","

    # pandas infers .gz decompression from the filename. read_csv is a safe,
    # vetted reader — no code execution (unlike read_pickle).
    df = pd.read_csv(staged_path, index_col=0, sep=sep)
    X = df.to_numpy(dtype=float)  # raises if the matrix isn't purely numeric
    X_flat = _cap_flat(X.flatten())
    var_index = [str(c) for c in df.columns]

    checks = {
        "data_level": check_data_level(X_flat, manifest_obj.data_level),
        "feature_id_type": check_feature_id_type(var_index, manifest_obj.feature_id_type),
    }
    errors = [
        f"[{name}] {c.get('message', '')}"
        for name, c in checks.items()
        if c.get("status") == "error"
    ]
    warnings = [
        f"[{name}] {c.get('message', '')}"
        for name, c in checks.items()
        if c.get("status") == "warning"
    ]

    skipped = ["metadata_columns", "group_columns", "default_contrasts"]
    caveats = [
        "Flat matrix carries no sample metadata (obs), so the obs-dependent checks "
        f"({', '.join(skipped)}) were NOT run — re-upload as an .h5ad with embedded "
        "obs to validate grouping and contrasts.",
        "Orientation assumed samples-as-rows, genes-as-columns (index_col=0); if the "
        "feature_id_type check flags a mismatch, the matrix may be transposed.",
    ]

    return {
        "dataset_id": manifest_obj.dataset_id,
        "n_samples": int(df.shape[0]),
        "n_features": int(df.shape[1]),
        "overall_valid": len(errors) == 0,
        "n_errors": len(errors),
        "n_warnings": len(warnings),
        "checks": checks,
        "errors": errors,
        "warnings": warnings,
        "skipped_checks": skipped,
        "caveats": caveats,
    }


def validate_upload(record: UploadRecord) -> tuple[UploadRecord, dict]:
    """Run the manifest-schema + against-data gate on a quarantined upload.

    Returns ``(record, report)``. The record's ``status`` becomes ``validated``
    only when both gates pass; otherwise it stays ``quarantined`` and the reasons
    are in ``record.errors`` / ``report['errors']``. A flat-matrix upload that
    validates carries ``record.caveats`` naming the checks that could not run.
    """
    if record.status != STATUS_QUARANTINED:
        return record, _report(
            "error",
            [f"Upload is '{record.status}', not 'quarantined' — nothing to validate."],
        )
    if not record.staged_path or not Path(record.staged_path).is_file():
        record.errors = [f"Staged file missing: {record.staged_path}"]
        persist_record(record)
        return record, _report("error", record.errors)

    # ── Gate 0: content scan must run and clear before validation ────────────
    # ADR-0011: "the optional malware scan runs on the staged object *before*
    # validation." The structural magic-byte check is unconditional; the AV pass
    # is best-effort (skipped-with-caveat unless UPLOAD_SCAN_REQUIRED). An
    # infected/disguised file never reaches the loaders below.
    from src.uploads.scanning import SCAN_INFECTED, scan_upload

    if record.scan_status is None:
        record, scan_report = scan_upload(record)
    if record.scan_status == SCAN_INFECTED:
        return record, _report("error", record.errors, scan=record.scan_detail)
    scan_caveats = list(record.caveats)  # a skipped-AV caveat, if any

    # ── Gate 1: manifest schema validity ─────────────────────────────────────
    from src.datasets.manifest_schema import DatasetManifest, validate_manifest

    schema_result = validate_manifest(record.manifest)
    if not schema_result.valid:
        record.errors = [f"manifest schema: {e}" for e in schema_result.errors]
        persist_record(record)
        return record, _report("error", record.errors, schema_warnings=schema_result.warnings)

    manifest_obj = DatasetManifest.from_dict(record.manifest)

    # ── Gate 2: against-data validation via the VETTED reader only (item 4) ──
    kind = _kind_of(record.staged_path)
    if kind is None:
        record.errors = [f"Unsupported file type for validation: {Path(record.staged_path).name}."]
        persist_record(record)
        return record, _report("error", record.errors)

    try:
        if kind == "h5ad":
            data_report = _h5ad_report(record.staged_path, manifest_obj)
        else:
            data_report = _tabular_report(record.staged_path, manifest_obj)
    except Exception as exc:  # noqa: BLE001 — a load/validate failure is a rejection
        record.errors = [f"against-data validation failed to run: {exc}"]
        persist_record(record)
        return record, _report("error", record.errors)

    caveats = scan_caveats + [c for c in data_report.get("caveats", []) if c not in scan_caveats]

    if not data_report.get("overall_valid"):
        record.errors = list(data_report.get("errors", [])) or [
            "against-data validation reported the manifest inconsistent with the file."
        ]
        record.caveats = caveats
        persist_record(record)
        return record, _report(
            "error",
            record.errors,
            against_data=data_report,
            schema_warnings=schema_result.warnings,
        )

    # ── Passed both gates ────────────────────────────────────────────────────
    record.status = STATUS_VALIDATED
    record.validated_at = datetime.now().isoformat(timespec="seconds")
    record.errors = []
    record.caveats = caveats
    persist_record(record)
    return record, _report(
        "pass",
        [],
        against_data=data_report,
        schema_warnings=schema_result.warnings,
        caveats=caveats,
    )