#!/usr/bin/env python3
"""Consolidate public complete-WES audit artifacts without re-probing URLs.

The input audits intentionally answer different questions.  This script turns
them into one object-level inventory while preserving the distinctions among:

* source rows (the same object can occur in several audit artifacts),
* public alignment objects (one normalized alignment URL),
* exact content groups (official MD5 + byte size, when both are available), and
* biological samples/people (which must not be inferred from file count).

The script never changes an input artifact.  It waits for the broad public
probe sentinel to report ``complete``, checks embedded input hashes when they
are available, and requires the inputs to remain unchanged for a configurable
stability interval before reading them.
"""

from __future__ import annotations

import argparse
import base64
import csv
import hashlib
import json
import math
import statistics
import sys
import time
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.parse import parse_qsl, unquote, urlencode, urlsplit


DEFAULT_DATE = "2026-08-01"
MIB = 1024**2
GIB = 1024**3

OUTPUT_FIELDS = [
    "record_id",
    "repository",
    "collection",
    "dataset_id",
    "study_accession",
    "run_accession",
    "sample_id",
    "individual_id",
    "sample_alias",
    "sample_type",
    "technology",
    "capture_kit",
    "reference_build",
    "alignment_format",
    "access_class",
    "analysis_prerequisites",
    "file_name",
    "alignment_url",
    "alternate_alignment_urls",
    "index_url",
    "alternate_index_urls",
    "alignment_bytes",
    "alignment_gib",
    "size_bin",
    "size_review_flag",
    "index_bytes",
    "alignment_md5",
    "index_md5",
    "alignment_download_verified",
    "index_download_verified",
    "indexed_pair_verified",
    "alignment_http_status",
    "index_http_status",
    "alignment_content_range",
    "index_content_range",
    "alignment_magic_ok",
    "index_magic_ok",
    "verification_method",
    "verified_at_utc",
    "mean_target_coverage_x",
    "pct_target_bases_20x",
    "pct_target_bases_30x",
    "official_source_url",
    "source_artifacts",
    "source_row_count",
    "canonical_url_key",
    "source_logical_alignment_key",
    "logical_alignment_group_id",
    "logical_alignment_alias_count",
    "is_logical_alignment_alias",
    "content_group_id",
    "content_alias_count",
    "is_content_alias",
    "preferred_object_in_content_group",
    "whole_exome_claim_basis",
    "whole_exome_confidence",
    "brca_truth_status",
    "brca_truth_event",
    "truth_evidence_url",
    "validation_role",
    "notes",
]

REPOSITORY_FIELDS = [
    "repository",
    "candidate_objects",
    "alignment_download_verified_objects",
    "verified_indexed_pair_objects",
    "downloadable_alignment_without_verified_index",
    "failed_or_unverified_objects",
    "repository_scoped_sample_ids",
    "objects_with_known_size",
    "known_alignment_bytes",
    "min_alignment_bytes",
    "median_alignment_bytes",
    "max_alignment_bytes",
    "verified_indexed_objects_with_known_size",
    "verified_indexed_alignment_bytes",
    "verified_indexed_min_alignment_bytes",
    "verified_indexed_median_alignment_bytes",
    "verified_indexed_max_alignment_bytes",
    "objects_with_official_md5",
    "exact_content_groups_among_md5_objects",
    "extra_url_aliases_to_exact_content",
    "objects_flagged_for_size_review",
]

SIZE_FIELDS = [
    "size_bin",
    "candidate_objects",
    "alignment_download_verified_objects",
    "verified_indexed_pair_objects",
    "known_alignment_bytes",
    "alignment_download_verified_bytes",
    "verified_indexed_alignment_bytes",
    "fraction_of_candidate_objects",
]

REPOSITORY_SIZE_FIELDS = [
    "repository",
    "size_bin",
    "verified_indexed_pair_objects",
    "verified_indexed_alignment_bytes",
    "fraction_of_repository_verified_indexed_pairs",
]

REPOSITORY_NAMES = {
    "DepMap_CCLE_AWS": "DepMap CCLE (AWS Open Data)",
    "ENA": "European Nucleotide Archive (ENA)",
    "GATK_test_data_AWS": "Broad GATK test data (AWS Open Data)",
    "Google_Brain_GIAB_benchmark": "Google Brain GIAB benchmark",
    "NIST_GIAB": "NIST Genome in a Bottle / NCBI",
    "NCBI_SEQC2": "NCBI SEQC2",
}

HOST_ALIASES = {
    "ftp-trace.ncbi.nlm.nih.gov": "ftp.ncbi.nlm.nih.gov",
    "ftp-trace.ncbi.nih.gov": "ftp.ncbi.nlm.nih.gov",
    "ftp.ncbi.nih.gov": "ftp.ncbi.nlm.nih.gov",
}


def utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def read_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as handle:
        value = json.load(handle)
    if not isinstance(value, dict):
        raise ValueError(f"Expected a JSON object: {path}")
    return value


def read_csv(path: Path) -> list[dict[str, str]]:
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def text_value(value: Any) -> str:
    if value is None:
        return ""
    return str(value).strip()


def first(row: dict[str, Any], *names: str) -> str:
    for name in names:
        value = text_value(row.get(name))
        if value:
            return value
    return ""


def as_bool(value: Any) -> bool:
    return text_value(value).lower() in {"1", "true", "yes", "y", "pass"}


def as_int(value: Any) -> int | None:
    raw = text_value(value).replace(",", "")
    if not raw:
        return None
    try:
        return int(float(raw))
    except ValueError:
        return None


def normalized_md5(value: Any) -> str:
    """Return a lower-case hexadecimal MD5 or an empty string.

    Some source artifacts name a field ``*_md5_base64``.  Supporting both
    encodings prevents a mirror from escaping exact-content grouping merely
    because the transport metadata used Base64 rather than hexadecimal.
    """

    raw = text_value(value)
    lowered = raw.lower()
    if len(lowered) == 32 and all(char in "0123456789abcdef" for char in lowered):
        return lowered
    try:
        decoded = base64.b64decode(raw, validate=True)
    except Exception:
        return ""
    return decoded.hex() if len(decoded) == 16 else ""


def canonical_url(url: str) -> str:
    """Normalize protocol/host aliases but retain path and meaningful query."""

    raw = text_value(url)
    if not raw:
        return ""
    parts = urlsplit(raw)
    host = (parts.hostname or "").lower()
    host = HOST_ALIASES.get(host, host)
    port = parts.port
    if port and port not in {80, 443}:
        host = f"{host}:{port}"
    path = unquote(parts.path).replace("//", "/").rstrip("/")
    query = urlencode(sorted(parse_qsl(parts.query, keep_blank_values=True)))
    return f"{host}{path}" + (f"?{query}" if query else "")


def basename_from_url(url: str) -> str:
    return Path(unquote(urlsplit(text_value(url)).path)).name


def stable_id(prefix: str, value: str, length: int = 16) -> str:
    return prefix + hashlib.sha256(value.encode("utf-8")).hexdigest()[:length]


def size_bin(byte_count: int | None) -> str:
    if byte_count is None:
        return "Unknown"
    if byte_count < 100 * MIB:
        return "<100 MiB"
    if byte_count < GIB:
        return "100 MiB-<1 GiB"
    if byte_count < 5 * GIB:
        return "1-<5 GiB"
    if byte_count < 10 * GIB:
        return "5-<10 GiB"
    if byte_count < 20 * GIB:
        return "10-<20 GiB"
    if byte_count < 50 * GIB:
        return "20-<50 GiB"
    return ">=50 GiB"


SIZE_BIN_ORDER = [
    "<100 MiB",
    "100 MiB-<1 GiB",
    "1-<5 GiB",
    "5-<10 GiB",
    "10-<20 GiB",
    "20-<50 GiB",
    ">=50 GiB",
    "Unknown",
]


def size_review_flag(byte_count: int | None) -> str:
    # Size is only a triage signal.  It is never used here as proof that a
    # repository-labelled WES file is complete or incomplete.
    if byte_count is None:
        return "review_unknown_size"
    if byte_count < 100 * MIB:
        return "review_extremely_small"
    if byte_count < GIB:
        return "review_low_volume"
    if byte_count >= 50 * GIB:
        return "review_very_large"
    return "plausible_by_volume_only"


def infer_sample_type(repository: str, row: dict[str, Any]) -> str:
    explicit = first(row, "sample_type")
    if explicit:
        return explicit
    if "DepMap" in repository:
        return "cancer cell line"
    if "GIAB" in repository or "Genome in a Bottle" in repository or "GATK" in repository:
        return "reference material / benchmark"
    if "1000 Genomes" in repository:
        return "population sample"
    if "ENA" in repository:
        return "human biosample; inspect study metadata"
    return "human sample; inspect source metadata"


def claim_metadata(repository: str) -> tuple[str, str, str]:
    if "1000 Genomes" in repository:
        return (
            "Official Phase 3 exome alignment index",
            "official_curated_exome_index",
            "population background / independently annotated controls",
        )
    if "GIAB" in repository or "Genome in a Bottle" in repository:
        return (
            "Official GIAB exome index or benchmark WES prefix",
            "benchmark_curated",
            "technical reproducibility and reference-sample testing",
        )
    if "SEQC2" in repository:
        return (
            "Official SEQC2 WES directory and checksum manifest",
            "benchmark_curated",
            "matched tumor-normal technical-replicate stress test",
        )
    if "DepMap" in repository:
        return (
            "DepMap Open Data record with datatype=wes",
            "repository_metadata",
            "cancer-cell-line stress testing; no germline BRCA exon-CNV truth assumed",
        )
    if "ENA" in repository:
        return (
            "ENA human record with library_strategy=WXS and submitted BAM+index",
            "repository_metadata_needs_header_and_coverage_qc",
            "discovery / background until study, header, and coverage QC are reviewed",
        )
    if "GATK" in repository:
        return (
            "Public GATK test-data object labelled exome",
            "test_data_metadata",
            "software smoke test; underlying individual may duplicate another collection",
        )
    return ("Source labelled complete WES", "source_metadata", "research use after QC")


def access_evidence(
    *, pair_verified: bool, alignment_status: str, index_status: str, index_url: str
) -> tuple[bool, bool, bool, str]:
    """Translate source probe evidence into non-overlapping access classes."""

    # The source audit's pair boolean also checks expected size and any format
    # criteria it implements.  A bare 206 status must not override a failed
    # source audit, because it may have returned the wrong object length.
    alignment_verified = pair_verified
    index_verified = pair_verified
    if pair_verified:
        access_class = "ready_indexed_pair"
    elif alignment_verified and not index_url:
        access_class = "downloadable_alignment_no_submitted_index"
    elif alignment_verified:
        access_class = "downloadable_alignment_index_failed_or_unverified"
    else:
        access_class = "alignment_download_failed_or_unverified"
    return alignment_verified, index_verified, pair_verified, access_class


def blank_record() -> dict[str, Any]:
    record = {field: "" for field in OUTPUT_FIELDS}
    record.update(
        {
            "_source_artifacts": set(),
            "_alternate_alignment_urls": set(),
            "_alternate_index_urls": set(),
            "_notes": set(),
            "_source_row_count": 1,
        }
    )
    return record


def baseline_adapter(row: dict[str, str], source: str) -> dict[str, Any]:
    repository = first(row, "repository")
    alignment_url = first(row, "alignment_url")
    byte_count = as_int(row.get("alignment_bytes"))
    claim, confidence, role = claim_metadata(repository)
    alignment_status = first(row, "alignment_http_status")
    index_status = first(row, "index_http_status")
    index_url = first(row, "index_url")
    alignment_ok, index_ok, pair_ok, access_class = access_evidence(
        pair_verified=as_bool(first(row, "downloadable", "verified_downloadable")),
        alignment_status=alignment_status,
        index_status=index_status,
        index_url=index_url,
    )
    record = blank_record()
    record.update(
        {
            "repository": repository,
            "collection": first(row, "collection"),
            "dataset_id": first(row, "dataset_id"),
            "sample_id": first(row, "sample_id"),
            "individual_id": first(row, "individual_key", "sample_id"),
            "sample_alias": first(row, "sample_alias"),
            "sample_type": infer_sample_type(repository, row),
            "technology": first(row, "technology"),
            "capture_kit": first(row, "capture_kit"),
            "reference_build": first(row, "reference_build"),
            "alignment_format": first(row, "alignment_format") or "BAM",
            "access_class": access_class,
            "analysis_prerequisites": "Matching BAI",
            "file_name": first(row, "file_name") or basename_from_url(alignment_url),
            "alignment_url": alignment_url,
            "index_url": index_url,
            "alignment_bytes": byte_count,
            "index_bytes": as_int(row.get("index_bytes")),
            "alignment_md5": normalized_md5(
                first(row, "alignment_md5_from_official_index", "alignment_md5")
            ),
            "index_md5": normalized_md5(
                first(row, "index_md5_from_official_index", "index_md5")
            ),
            "alignment_download_verified": alignment_ok,
            "index_download_verified": index_ok,
            "indexed_pair_verified": pair_ok,
            "alignment_http_status": alignment_status,
            "index_http_status": index_status,
            "alignment_content_range": first(row, "alignment_content_range"),
            "index_content_range": first(row, "index_content_range"),
            "alignment_magic_ok": first(row, "alignment_magic_ok"),
            "index_magic_ok": first(row, "index_magic_ok"),
            "verification_method": (
                "Anonymous byte-range GET of alignment and index; magic checked "
                "when recorded by source audit"
            ),
            "verified_at_utc": first(row, "verified_at_utc"),
            "mean_target_coverage_x": first(
                row, "mean_target_coverage", "mean_target_coverage_x"
            ),
            "pct_target_bases_20x": first(row, "pct_target_bases_20x"),
            "pct_target_bases_30x": first(row, "pct_target_bases_30x"),
            "official_source_url": first(row, "official_source_index_url"),
            "source_logical_alignment_key": first(row, "logical_alignment_key"),
            "whole_exome_claim_basis": claim,
            "whole_exome_confidence": confidence,
            "validation_role": role,
        }
    )
    record["_source_artifacts"].add(source)
    for value in (
        first(row, "alternate_alignment_mirror_url"),
        first(row, "alternate_alignment_url"),
    ):
        if value:
            record["_alternate_alignment_urls"].add(value)
    for value in (
        first(row, "alternate_index_mirror_url"),
        first(row, "alternate_index_url"),
    ):
        if value:
            record["_alternate_index_urls"].add(value)
    note = first(row, "deduplication_and_mirror_note")
    if note:
        record["_notes"].add(note)
    return record


def broad_adapter(row: dict[str, str], source: str) -> dict[str, Any]:
    repository = REPOSITORY_NAMES.get(first(row, "repository"), first(row, "repository"))
    alignment_url = first(row, "alignment_url")
    collection = first(row, "collection")
    byte_count = as_int(row.get("alignment_bytes"))
    claim, confidence, role = claim_metadata(repository)
    alignment_status = first(row, "alignment_http_status")
    index_status = first(row, "index_http_status")
    index_url = first(row, "index_url")
    alignment_format = first(row, "alignment_format") or "BAM"
    alignment_ok, index_ok, pair_ok, access_class = access_evidence(
        pair_verified=as_bool(first(row, "verified_downloadable")),
        alignment_status=alignment_status,
        index_status=index_status,
        index_url=index_url,
    )
    if alignment_format.upper() == "CRAM" and pair_ok:
        access_class = "ready_indexed_pair_reference_required"
    record = blank_record()
    record.update(
        {
            "repository": repository,
            "collection": collection,
            "dataset_id": collection,
            "study_accession": collection if collection.startswith("PRJ") else "",
            "run_accession": first(row, "run_accession"),
            "sample_id": first(row, "sample_id"),
            "individual_id": first(row, "sample_id"),
            "sample_type": infer_sample_type(repository, row),
            "technology": first(row, "platform_or_kit"),
            "capture_kit": first(row, "platform_or_kit"),
            "reference_build": first(row, "reference_build"),
            "alignment_format": alignment_format,
            "access_class": access_class,
            "analysis_prerequisites": (
                "CRAI plus the exact reference sequence identified by CRAM header MD5s"
                if alignment_format.upper() == "CRAM"
                else "Matching BAI"
            ),
            "file_name": basename_from_url(alignment_url),
            "alignment_url": alignment_url,
            "index_url": index_url,
            "alignment_bytes": byte_count,
            "index_bytes": as_int(row.get("index_bytes")),
            "alignment_md5": normalized_md5(
                first(row, "alignment_md5", "alignment_checksum")
            ),
            "index_md5": normalized_md5(first(row, "index_md5", "index_checksum")),
            "alignment_download_verified": alignment_ok,
            "index_download_verified": index_ok,
            "indexed_pair_verified": pair_ok,
            "alignment_http_status": alignment_status,
            "index_http_status": index_status,
            "alignment_content_range": first(row, "alignment_content_range"),
            "index_content_range": first(row, "index_content_range"),
            "verification_method": "Anonymous HTTP GET Range bytes=0-0 on alignment and index",
            "whole_exome_claim_basis": claim,
            "whole_exome_confidence": confidence,
            "validation_role": role,
        }
    )
    record["_source_artifacts"].add(source)
    origin = first(row, "origin_note")
    if origin:
        record["_notes"].add(origin)
    return record


def unindexed_adapter(row: dict[str, str], source: str) -> dict[str, Any]:
    """Adapt a live-probed ENA BAM for which no submitted BAI exists."""

    repository = "European Nucleotide Archive (ENA)"
    alignment_url = first(row, "alignment_url")
    alignment_status = first(row, "alignment_http_status")
    alignment_ok = as_bool(first(row, "verified_downloadable"))
    claim, confidence, role = claim_metadata(repository)
    record = blank_record()
    record.update(
        {
            "repository": repository,
            "collection": first(row, "collection"),
            "dataset_id": first(row, "collection"),
            "study_accession": first(row, "collection"),
            "run_accession": first(row, "run_accession"),
            "sample_id": first(row, "sample_id"),
            "individual_id": first(row, "sample_id"),
            "sample_type": infer_sample_type(repository, row),
            "technology": first(row, "capture"),
            "capture_kit": first(row, "capture"),
            "reference_build": first(row, "reference_build"),
            "alignment_format": first(row, "alignment_format") or "BAM",
            "access_class": (
                "downloadable_alignment_no_submitted_index"
                if alignment_ok
                else "alignment_download_failed_or_unverified"
            ),
            "analysis_prerequisites": "Full BAM download followed by local BAI creation",
            "file_name": basename_from_url(alignment_url),
            "alignment_url": alignment_url,
            "index_url": "",
            "alignment_bytes": as_int(row.get("alignment_bytes")),
            "index_bytes": None,
            "alignment_md5": normalized_md5(first(row, "alignment_md5")),
            "index_md5": "",
            "alignment_download_verified": alignment_ok,
            "index_download_verified": False,
            "indexed_pair_verified": False,
            "alignment_http_status": alignment_status,
            "index_http_status": "",
            "alignment_content_range": first(row, "alignment_content_range"),
            "index_content_range": "",
            "verification_method": "Anonymous HTTP GET Range bytes=0-0 on alignment only",
            "whole_exome_claim_basis": claim,
            "whole_exome_confidence": confidence,
            "validation_role": role,
        }
    )
    record["_source_artifacts"].add(source)
    limitation = first(row, "limitation")
    if limitation:
        record["_notes"].add(limitation)
    return record


def seqc2_adapter(row: dict[str, str], source: str) -> dict[str, Any]:
    repository = "NCBI SEQC2"
    alignment_url = first(row, "alignment_url")
    sample_type = first(row, "sample_type")
    sample_id = "HCC1395BL" if "HCC1395BL" in sample_type else "HCC1395"
    claim, confidence, role = claim_metadata(repository)
    alignment_status = first(row, "alignment_http_status")
    index_status = first(row, "index_http_status")
    index_url = first(row, "index_url")
    alignment_ok, index_ok, pair_ok, access_class = access_evidence(
        pair_verified=as_bool(first(row, "verified_downloadable")),
        alignment_status=alignment_status,
        index_status=index_status,
        index_url=index_url,
    )
    record = blank_record()
    record.update(
        {
            "repository": repository,
            "collection": first(row, "collection"),
            "dataset_id": "SEQC2_HCC1395_WES",
            "sample_id": sample_id,
            "individual_id": sample_id,
            "sample_alias": ":".join(
                part for part in [first(row, "center_code"), first(row, "replicate")] if part
            ),
            "sample_type": sample_type,
            "technology": first(row, "capture"),
            "capture_kit": first(row, "capture"),
            "reference_build": first(row, "reference_build"),
            "alignment_format": "BAM",
            "access_class": access_class,
            "analysis_prerequisites": "Matching BAI",
            "file_name": basename_from_url(alignment_url),
            "alignment_url": alignment_url,
            "index_url": index_url,
            "alignment_bytes": as_int(row.get("alignment_bytes")),
            "index_bytes": as_int(row.get("index_bytes")),
            "alignment_md5": normalized_md5(first(row, "alignment_md5")),
            "index_md5": normalized_md5(first(row, "index_md5")),
            "alignment_download_verified": alignment_ok,
            "index_download_verified": index_ok,
            "indexed_pair_verified": pair_ok,
            "alignment_http_status": alignment_status,
            "index_http_status": index_status,
            "verification_method": "Anonymous HTTP GET Range bytes=0-0 on alignment and index",
            "whole_exome_claim_basis": claim,
            "whole_exome_confidence": confidence,
            "validation_role": role,
        }
    )
    record["_source_artifacts"].add(source)
    return record


def consistency_value(field: str, left: Any, right: Any, url: str) -> Any:
    if left in {None, ""}:
        return right
    if right in {None, ""}:
        return left
    if field in {"alignment_bytes", "index_bytes", "alignment_md5", "index_md5"}:
        if left != right:
            raise ValueError(
                f"Conflicting {field} for the same normalized URL {url}: {left!r} vs {right!r}"
            )
    return left


def merge_same_object(left: dict[str, Any], right: dict[str, Any]) -> dict[str, Any]:
    url = left["canonical_url_key"]
    for field in OUTPUT_FIELDS:
        if field in {
            "record_id",
            "source_artifacts",
            "source_row_count",
            "canonical_url_key",
            "logical_alignment_group_id",
            "logical_alignment_alias_count",
            "is_logical_alignment_alias",
            "content_group_id",
            "content_alias_count",
            "is_content_alias",
            "preferred_object_in_content_group",
            "notes",
        }:
            continue
        if field in {
            "alignment_download_verified",
            "index_download_verified",
            "indexed_pair_verified",
        }:
            left[field] = bool(left[field]) or bool(right[field])
        else:
            left[field] = consistency_value(field, left.get(field), right.get(field), url)
    left["access_class"] = access_evidence(
        pair_verified=bool(left["indexed_pair_verified"]),
        alignment_status=text_value(left["alignment_http_status"]),
        index_status=text_value(left["index_http_status"]),
        index_url=text_value(left["index_url"]),
    )[3]
    if text_value(left.get("alignment_format")).upper() == "CRAM" and bool(
        left["indexed_pair_verified"]
    ):
        left["access_class"] = "ready_indexed_pair_reference_required"
    left["_source_artifacts"].update(right["_source_artifacts"])
    left["_alternate_alignment_urls"].update(right["_alternate_alignment_urls"])
    left["_alternate_index_urls"].update(right["_alternate_index_urls"])
    left["_notes"].update(right["_notes"])
    left["_source_row_count"] += right["_source_row_count"]
    return left


def prepare_record(record: dict[str, Any]) -> dict[str, Any]:
    url_key = canonical_url(text_value(record["alignment_url"]))
    if not url_key:
        raise ValueError(f"Missing alignment URL in {record['_source_artifacts']}")
    record["canonical_url_key"] = url_key
    record["record_id"] = stable_id("wesobj_", url_key)
    byte_count = record.get("alignment_bytes")
    record["alignment_gib"] = (
        f"{byte_count / GIB:.6f}" if isinstance(byte_count, int) else ""
    )
    record["size_bin"] = size_bin(byte_count)
    record["size_review_flag"] = size_review_flag(byte_count)
    return record


def truth_rank(status: str) -> int:
    lowered = status.lower()
    if "positive" in lowered or "pathogenic" in lowered:
        return 3
    if "negative" in lowered:
        return 2
    if lowered and lowered != "unlabeled":
        return 1
    return 0


def truth_keys(row: dict[str, str]) -> list[str]:
    keys: list[str] = []
    url_key = canonical_url(first(row, "bam_url", "alignment_url"))
    if url_key:
        keys.append("url:" + url_key)
    md5 = normalized_md5(first(row, "bam_md5_base64", "alignment_md5"))
    byte_count = as_int(first(row, "bam_bytes", "alignment_bytes"))
    if md5 and byte_count is not None:
        keys.append(f"md5:{md5}:{byte_count}")
    sample = first(row, "sample_id")
    name = basename_from_url(first(row, "bam_url", "alignment_url"))
    if sample and byte_count is not None and name:
        keys.append(f"fallback:{sample}:{byte_count}:{name}")
    return keys


def enrich_truth(records: list[dict[str, Any]], truth_rows: list[dict[str, str]]) -> int:
    annotations: dict[str, dict[str, str]] = {}
    for row in truth_rows:
        status = first(row, "brca_truth_status", "truth_status")
        candidate = {
            "brca_truth_status": status,
            "brca_truth_event": first(row, "brca_truth_event", "paper_candidate_event"),
            "truth_evidence_url": first(row, "truth_evidence_url"),
            "validation_role": first(row, "validation_use"),
        }
        for key in truth_keys(row):
            previous = annotations.get(key)
            if previous is None or truth_rank(status) > truth_rank(
                previous.get("brca_truth_status", "")
            ):
                annotations[key] = candidate

    enriched = 0
    for record in records:
        keys = ["url:" + record["canonical_url_key"]]
        md5 = text_value(record.get("alignment_md5"))
        byte_count = record.get("alignment_bytes")
        if md5 and isinstance(byte_count, int):
            keys.append(f"md5:{md5}:{byte_count}")
        if record.get("sample_id") and isinstance(byte_count, int) and record.get("file_name"):
            keys.append(
                f"fallback:{record['sample_id']}:{byte_count}:{record['file_name']}"
            )
        matches = [annotations[key] for key in keys if key in annotations]
        if not matches:
            continue
        best = max(matches, key=lambda item: truth_rank(item["brca_truth_status"]))
        if truth_rank(best["brca_truth_status"]) == 0:
            continue
        for field, value in best.items():
            if value:
                record[field] = value
        enriched += 1
    return enriched


def assign_content_groups(records: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
    groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for record in records:
        md5 = text_value(record.get("alignment_md5"))
        byte_count = record.get("alignment_bytes")
        if md5 and isinstance(byte_count, int):
            raw_key = f"md5:{md5}:{byte_count}"
            group_id = stable_id("content_", raw_key)
        else:
            raw_key = "url:" + record["canonical_url_key"]
            group_id = stable_id("unknown_content_", raw_key)
        record["content_group_id"] = group_id
        groups[group_id].append(record)

    for members in groups.values():
        preferred = min(
            members,
            key=lambda row: (
                not bool(row["indexed_pair_verified"]),
                not bool(row["alignment_download_verified"]),
                row["repository"],
                row["alignment_url"],
            ),
        )
        for record in members:
            record["content_alias_count"] = len(members)
            record["is_content_alias"] = len(members) > 1
            record["preferred_object_in_content_group"] = record is preferred
    return groups


def assign_logical_alignment_groups(
    records: list[dict[str, Any]],
) -> dict[str, list[dict[str, Any]]]:
    """Group only source-declared aliases or byte-identical content.

    This layer is intentionally more conservative than sample-name grouping.
    It handles, for example, the NIST NA12878 b37/hg19 legacy reheader alias
    while retaining true technical replicates as separate logical alignments.
    """

    groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for record in records:
        source_key = text_value(record.get("source_logical_alignment_key"))
        if source_key:
            raw_key = f"source:{record['repository']}:{source_key}"
        elif text_value(record.get("alignment_md5")):
            raw_key = "content:" + record["content_group_id"]
        else:
            raw_key = "object:" + record["record_id"]
        group_id = stable_id("logical_", raw_key)
        record["logical_alignment_group_id"] = group_id
        groups[group_id].append(record)
    for members in groups.values():
        for record in members:
            record["logical_alignment_alias_count"] = len(members)
            record["is_logical_alignment_alias"] = len(members) > 1
    return groups


def finalize_record(record: dict[str, Any]) -> dict[str, Any]:
    record["alternate_alignment_urls"] = ";".join(
        sorted(record.pop("_alternate_alignment_urls"))
    )
    record["alternate_index_urls"] = ";".join(
        sorted(record.pop("_alternate_index_urls"))
    )
    record["source_artifacts"] = ";".join(sorted(record.pop("_source_artifacts")))
    record["source_row_count"] = record.pop("_source_row_count")
    record["notes"] = " | ".join(sorted(record.pop("_notes")))
    for field in OUTPUT_FIELDS:
        value = record.get(field, "")
        if isinstance(value, bool):
            record[field] = "True" if value else "False"
        elif value is None:
            record[field] = ""
    return {field: record.get(field, "") for field in OUTPUT_FIELDS}


def median_int(values: list[int]) -> int | None:
    if not values:
        return None
    return int(statistics.median(values))


def repository_summary(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for repository in sorted({str(row["repository"]) for row in records}):
        group = [row for row in records if row["repository"] == repository]
        sizes = [
            int(row["alignment_bytes"])
            for row in group
            if isinstance(row.get("alignment_bytes"), int)
        ]
        indexed_sizes = [
            int(row["alignment_bytes"])
            for row in group
            if bool(row["indexed_pair_verified"])
            and isinstance(row.get("alignment_bytes"), int)
        ]
        md5_rows = [row for row in group if text_value(row.get("alignment_md5"))]
        content_keys = {
            (row["alignment_md5"], row.get("alignment_bytes")) for row in md5_rows
        }
        rows.append(
            {
                "repository": repository,
                "candidate_objects": len(group),
                "alignment_download_verified_objects": sum(
                    bool(row["alignment_download_verified"]) for row in group
                ),
                "verified_indexed_pair_objects": sum(
                    bool(row["indexed_pair_verified"]) for row in group
                ),
                "downloadable_alignment_without_verified_index": sum(
                    bool(row["alignment_download_verified"])
                    and not bool(row["indexed_pair_verified"])
                    for row in group
                ),
                "failed_or_unverified_objects": sum(
                    not bool(row["alignment_download_verified"]) for row in group
                ),
                "repository_scoped_sample_ids": len(
                    {row["sample_id"] for row in group if row.get("sample_id")}
                ),
                "objects_with_known_size": len(sizes),
                "known_alignment_bytes": sum(sizes),
                "min_alignment_bytes": min(sizes) if sizes else "",
                "median_alignment_bytes": median_int(sizes) or "",
                "max_alignment_bytes": max(sizes) if sizes else "",
                "verified_indexed_objects_with_known_size": len(indexed_sizes),
                "verified_indexed_alignment_bytes": sum(indexed_sizes),
                "verified_indexed_min_alignment_bytes": (
                    min(indexed_sizes) if indexed_sizes else ""
                ),
                "verified_indexed_median_alignment_bytes": (
                    median_int(indexed_sizes) or ""
                ),
                "verified_indexed_max_alignment_bytes": (
                    max(indexed_sizes) if indexed_sizes else ""
                ),
                "objects_with_official_md5": len(md5_rows),
                "exact_content_groups_among_md5_objects": len(content_keys),
                "extra_url_aliases_to_exact_content": len(md5_rows) - len(content_keys),
                "objects_flagged_for_size_review": sum(
                    row["size_review_flag"] != "plausible_by_volume_only" for row in group
                ),
            }
        )
    return rows


def size_summary(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    total = len(records)
    rows: list[dict[str, Any]] = []
    for label in SIZE_BIN_ORDER:
        group = [row for row in records if row["size_bin"] == label]
        rows.append(
            {
                "size_bin": label,
                "candidate_objects": len(group),
                "alignment_download_verified_objects": sum(
                    bool(row["alignment_download_verified"]) for row in group
                ),
                "verified_indexed_pair_objects": sum(
                    bool(row["indexed_pair_verified"]) for row in group
                ),
                "known_alignment_bytes": sum(
                    row["alignment_bytes"]
                    for row in group
                    if isinstance(row.get("alignment_bytes"), int)
                ),
                "alignment_download_verified_bytes": sum(
                    row["alignment_bytes"]
                    for row in group
                    if bool(row["alignment_download_verified"])
                    and isinstance(row.get("alignment_bytes"), int)
                ),
                "verified_indexed_alignment_bytes": sum(
                    row["alignment_bytes"]
                    for row in group
                    if bool(row["indexed_pair_verified"])
                    and isinstance(row.get("alignment_bytes"), int)
                ),
                "fraction_of_candidate_objects": f"{len(group) / total:.8f}" if total else "0",
            }
        )
    return rows


def repository_size_summary(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for repository in sorted({str(row["repository"]) for row in records}):
        verified = [
            row
            for row in records
            if row["repository"] == repository and bool(row["indexed_pair_verified"])
        ]
        denominator = len(verified)
        for label in SIZE_BIN_ORDER:
            group = [row for row in verified if row["size_bin"] == label]
            rows.append(
                {
                    "repository": repository,
                    "size_bin": label,
                    "verified_indexed_pair_objects": len(group),
                    "verified_indexed_alignment_bytes": sum(
                        row["alignment_bytes"]
                        for row in group
                        if isinstance(row.get("alignment_bytes"), int)
                    ),
                    "fraction_of_repository_verified_indexed_pairs": (
                        f"{len(group) / denominator:.8f}" if denominator else "0"
                    ),
                }
            )
    return rows


def write_csv(path: Path, fields: list[str], rows: Iterable[dict[str, Any]]) -> None:
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)


def required_paths(analysis: Path, date: str) -> dict[str, Path]:
    return {
        "baseline_csv": analysis / f"public_complete_wes_alignment_inventory_{date}.csv",
        "baseline_json": analysis / f"public_complete_wes_alignment_inventory_{date}.json",
        "broad_csv": analysis / f"public_complete_wes_alignment_inventory_verified_{date}.csv",
        "broad_json": analysis / f"public_complete_wes_repository_audit_{date}.json",
        "probe_progress": analysis / f"public_complete_wes_alignment_probe_progress_{date}.json",
        "seqc2_csv": analysis / f"seqc2_complete_wes_alignment_inventory_verified_{date}.csv",
        "seqc2_json": analysis / f"seqc2_complete_wes_repository_audit_{date}.json",
        "cram_csv": analysis / f"ena_human_wxs_cram_all_candidates_{date}.csv",
        "cram_json": analysis / f"ena_human_wxs_cram_repository_audit_{date}.json",
        "cram_progress": analysis / f"ena_wxs_cram_probe_progress_{date}.json",
        "unindexed_csv": analysis / f"ena_human_wxs_bam_without_index_all_candidates_{date}.csv",
        "unindexed_json": analysis / f"ena_human_wxs_bam_without_index_audit_{date}.json",
        "unindexed_progress": analysis / f"ena_wxs_bam_unindexed_probe_progress_{date}.json",
        "truth_csv": analysis / f"public_full_wes_brca_specific_alignment_inventory_{date}.csv",
        "truth_json": analysis / f"public_full_wes_brca_specific_audit_{date}.json",
        "raw_fastq_csv": analysis / f"public_brca_candidate_raw_wes_inventory_{date}.csv",
    }


def validate_input_set(paths: dict[str, Path]) -> list[str]:
    problems: list[str] = []
    for name, path in paths.items():
        if not path.exists():
            problems.append(f"missing {name}: {path}")
    if problems:
        return problems

    try:
        for label, progress_name in (
            ("broad BAM/index", "probe_progress"),
            ("ENA CRAM/index", "cram_progress"),
            ("ENA unindexed BAM", "unindexed_progress"),
        ):
            progress = read_json(paths[progress_name])
            if progress.get("status") != "complete":
                problems.append(
                    f"{label} probe status is {progress.get('status')!r}, expected 'complete'"
                )
    except Exception as error:
        problems.append(f"cannot read probe sentinel: {error}")

    hash_checks = [
        ("baseline_json", "baseline_csv", "inventory_csv_sha256"),
    ]
    for json_name, csv_name, field in hash_checks:
        try:
            metadata = read_json(paths[json_name])
            expected = text_value(metadata.get(field)).lower()
            if expected and sha256_file(paths[csv_name]).lower() != expected:
                problems.append(f"hash mismatch: {csv_name} does not match {json_name}")
        except Exception as error:
            problems.append(f"cannot validate {csv_name}: {error}")

    # The official release has 2,535 main Phase 3 exome alignments plus 16
    # additional QA-passed exome-only alignments.  Refuse to consolidate the
    # older partial artifact that omitted the technical-directory files.
    try:
        baseline_metadata = read_json(paths["baseline_json"])
        one_kgp = (
            baseline_metadata.get("summary", {})
            .get("repositories", {})
            .get("1000 Genomes Project", {})
        )
        listed = int(one_kgp.get("listed_alignment_files", 0))
        if listed < 2551:
            problems.append(
                f"1000 Genomes baseline lists {listed} alignments; expected at least "
                "2,551 after adding the 16 official exome-only files"
            )
    except Exception as error:
        problems.append(f"cannot validate 1000 Genomes completeness: {error}")

    try:
        truth_metadata = read_json(paths["truth_json"])
        expected = text_value(
            truth_metadata.get("artifacts", {}).get("alignment_inventory_sha256")
        ).lower()
        if expected and sha256_file(paths["truth_csv"]).lower() != expected:
            problems.append("hash mismatch: truth_csv does not match truth_json")
        raw_expected = text_value(
            truth_metadata.get("artifacts", {}).get("raw_candidate_inventory_sha256")
        ).lower()
        if raw_expected and sha256_file(paths["raw_fastq_csv"]).lower() != raw_expected:
            problems.append("hash mismatch: raw_fastq_csv does not match truth_json")
    except Exception as error:
        problems.append(f"cannot validate truth inventory: {error}")
    return problems


def wait_for_final_inputs(
    paths: dict[str, Path], timeout_seconds: int, stability_seconds: int
) -> None:
    deadline = time.monotonic() + timeout_seconds
    previous: dict[Path, tuple[int, int]] | None = None
    stable_since: float | None = None
    while True:
        problems = validate_input_set(paths)
        signatures = {
            path: (path.stat().st_size, path.stat().st_mtime_ns)
            for path in paths.values()
            if path.exists()
        }
        now = time.monotonic()
        if not problems and signatures == previous:
            stable_since = stable_since or now
            if now - stable_since >= stability_seconds:
                return
        else:
            stable_since = None
        if now >= deadline:
            detail = "; ".join(problems) if problems else "inputs did not remain stable"
            raise TimeoutError(f"Final inputs not ready: {detail}")
        previous = signatures
        time.sleep(min(5, max(1, stability_seconds)))


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    project = Path(__file__).resolve().parents[1]
    parser.add_argument("--analysis-dir", type=Path, default=project / "analysis")
    parser.add_argument("--date", default=DEFAULT_DATE)
    parser.add_argument("--wait-timeout-seconds", type=int, default=900)
    parser.add_argument("--stability-seconds", type=int, default=30)
    parser.add_argument(
        "--dry-run", action="store_true", help="Validate and summarize without writing outputs"
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    analysis = args.analysis_dir.resolve()
    paths = required_paths(analysis, args.date)
    wait_for_final_inputs(paths, args.wait_timeout_seconds, args.stability_seconds)

    inputs: list[tuple[str, list[dict[str, str]], Any]] = [
        (
            paths["baseline_csv"].name,
            read_csv(paths["baseline_csv"]),
            baseline_adapter,
        ),
        (paths["broad_csv"].name, read_csv(paths["broad_csv"]), broad_adapter),
        (paths["seqc2_csv"].name, read_csv(paths["seqc2_csv"]), seqc2_adapter),
        (paths["cram_csv"].name, read_csv(paths["cram_csv"]), broad_adapter),
        (
            paths["unindexed_csv"].name,
            read_csv(paths["unindexed_csv"]),
            unindexed_adapter,
        ),
    ]

    source_row_counts: dict[str, int] = {}
    objects: dict[str, dict[str, Any]] = {}
    for source, rows, adapter in inputs:
        source_row_counts[source] = len(rows)
        for raw in rows:
            record = prepare_record(adapter(raw, source))
            key = record["canonical_url_key"]
            if key in objects:
                objects[key] = merge_same_object(objects[key], record)
            else:
                objects[key] = record

    records = sorted(objects.values(), key=lambda row: (row["repository"], row["alignment_url"]))
    truth_rows = read_csv(paths["truth_csv"])
    truth_enriched = enrich_truth(records, truth_rows)
    content_groups = assign_content_groups(records)
    logical_groups = assign_logical_alignment_groups(records)
    repo_rows = repository_summary(records)
    bin_rows = size_summary(records)
    repo_bin_rows = repository_size_summary(records)
    raw_fastq_rows = read_csv(paths["raw_fastq_csv"])

    objects_with_md5 = [row for row in records if text_value(row.get("alignment_md5"))]
    md5_content_groups = {
        (row["alignment_md5"], row.get("alignment_bytes")) for row in objects_with_md5
    }
    alignment_verified = [
        row for row in records if bool(row["alignment_download_verified"])
    ]
    indexed_pair_verified = [row for row in records if bool(row["indexed_pair_verified"])]
    alignment_only_verified = [
        row
        for row in records
        if bool(row["alignment_download_verified"]) and not bool(row["indexed_pair_verified"])
    ]
    curated_confidence = {"official_curated_exome_index", "benchmark_curated"}
    curated_complete = [
        row for row in records if row.get("whole_exome_confidence") in curated_confidence
    ]
    metadata_only = [
        row for row in records if row.get("whole_exome_confidence") not in curated_confidence
    ]
    raw_fastq_downloadable = [row for row in raw_fastq_rows if as_bool(row.get("downloadable"))]
    brca_positive_objects = [
        row
        for row in records
        if "positive" in text_value(row.get("brca_truth_status")).lower()
    ]
    known_sizes = [
        row["alignment_bytes"]
        for row in records
        if isinstance(row.get("alignment_bytes"), int)
    ]
    source_rows_total = sum(source_row_counts.values())
    summary = {
        "schema_version": "1.0",
        "generated_at_utc": utc_now(),
        "scope": (
            "Public repository-labelled complete human WES BAM/CRAM alignment objects, "
            "including indexed pairs and separately classified downloadable BAMs with "
            "no submitted index. Raw FASTQ-only files are summarized separately and are "
            "not inserted into the alignment inventory."
        ),
        "counting_denominators": {
            "source_rows": source_rows_total,
            "unique_normalized_alignment_objects": len(records),
            "conservatively_deduplicated_logical_alignments": len(logical_groups),
            "extra_public_object_aliases_to_logical_alignments": len(records)
            - len(logical_groups),
            "alignment_download_verified_objects": len(alignment_verified),
            "verified_indexed_alignment_pairs": len(indexed_pair_verified),
            "downloadable_alignment_objects_without_verified_index": len(
                alignment_only_verified
            ),
            "failed_or_unverified_alignment_objects": len(records)
            - len(alignment_verified),
            "downloadable_raw_wes_fastq_files_reported_separately": len(
                raw_fastq_downloadable
            ),
            "curated_or_benchmark_complete_wes_objects": len(curated_complete),
            "repository_metadata_only_objects_requiring_content_qc": len(metadata_only),
            "objects_with_official_alignment_md5": len(objects_with_md5),
            "exact_content_groups_among_md5_objects": len(md5_content_groups),
            "extra_url_aliases_to_exact_content": len(objects_with_md5)
            - len(md5_content_groups),
            "objects_without_official_alignment_md5": len(records) - len(objects_with_md5),
            "content_group_count_including_url_fallback_for_unknown_md5": len(content_groups),
            "truth_enriched_alignment_objects": truth_enriched,
            "independently_supported_brca_positive_alignment_objects": len(
                brca_positive_objects
            ),
            "independently_supported_brca_positive_people": len(
                {
                    row["individual_id"] or row["sample_id"]
                    for row in brca_positive_objects
                    if row["individual_id"] or row["sample_id"]
                }
            ),
        },
        "important_interpretation": [
            "A verified indexed pair is one public alignment URL and its index URL that both passed the source audit's anonymous byte-range test. Complete files were not downloaded during this audit.",
            "A downloadable unindexed BAM passed an alignment byte-range test but has no submitted BAI. It requires full download and local indexing and is not included in the ready indexed-pair total.",
            "Raw FASTQ files are counted separately from alignment objects because they require alignment before LGASieve can use them.",
            "Object count is not patient count. Technical replicates, alternative alignments, and repeated submissions are retained as separate objects.",
            "Exact-content deduplication is performed only with an official MD5 plus byte size. Objects without an MD5 are not guessed to be duplicates.",
            "Repository WXS/WES metadata establishes the discovery claim, not clinical suitability. ENA records in particular require BAM-header, capture-footprint, contamination, depth, and sample-level review.",
            "File size is a triage signal only. Small or very large objects are flagged for review rather than automatically included or excluded as complete WES.",
        ],
        "source_row_counts": source_row_counts,
        "known_alignment_bytes_total": sum(known_sizes),
        "known_alignment_bytes_download_verified": sum(
            row["alignment_bytes"]
            for row in alignment_verified
            if isinstance(row.get("alignment_bytes"), int)
        ),
        "known_alignment_bytes_verified_indexed": sum(
            row["alignment_bytes"]
            for row in indexed_pair_verified
            if isinstance(row.get("alignment_bytes"), int)
        ),
        "resource_class_summary": {
            "verified_indexed_alignment_pairs": len(indexed_pair_verified),
            "verified_downloadable_alignments_without_verified_index": len(
                alignment_only_verified
            ),
            "downloadable_raw_wes_fastq_files": len(raw_fastq_downloadable),
            "raw_wes_fastq_runs": len(
                {first(row, "run_accession") for row in raw_fastq_rows if first(row, "run_accession")}
            ),
            "raw_wes_fastq_known_bytes": sum(
                as_int(row.get("file_bytes")) or 0 for row in raw_fastq_rows
            ),
            "raw_fastq_inventory": paths["raw_fastq_csv"].name,
            "curated_or_benchmark_complete_wes_alignment_objects": len(curated_complete),
            "repository_metadata_only_alignment_objects_requiring_qc": len(metadata_only),
        },
        "repository_summary": repo_rows,
        "size_bin_summary": bin_rows,
        "repository_verified_indexed_size_bin_summary": repo_bin_rows,
        "input_sha256": {name: sha256_file(path) for name, path in paths.items()},
        "outputs": {},
    }

    printable = {
        "source_rows": source_rows_total,
        "unique_objects": len(records),
        "alignment_download_verified_objects": len(alignment_verified),
        "verified_indexed_pairs": len(indexed_pair_verified),
        "downloadable_without_verified_index": len(alignment_only_verified),
        "downloadable_raw_fastq_files": len(raw_fastq_downloadable),
        "md5_exact_content_groups": len(md5_content_groups),
        "truth_enriched_objects": truth_enriched,
    }
    if args.dry_run:
        print(json.dumps(printable, indent=2))
        return 0

    prefix = f"consolidated_public_complete_wes_{args.date}"
    inventory_path = analysis / f"{prefix}.csv"
    repository_path = analysis / f"{prefix}_repository_summary.csv"
    size_path = analysis / f"{prefix}_size_bins.csv"
    repository_size_path = analysis / f"{prefix}_repository_verified_indexed_size_bins.csv"
    summary_path = analysis / f"{prefix}_summary.json"

    finalized = [finalize_record(row) for row in records]
    write_csv(inventory_path, OUTPUT_FIELDS, finalized)
    write_csv(repository_path, REPOSITORY_FIELDS, repo_rows)
    write_csv(size_path, SIZE_FIELDS, bin_rows)
    write_csv(repository_size_path, REPOSITORY_SIZE_FIELDS, repo_bin_rows)
    summary["outputs"] = {
        "inventory_csv": inventory_path.name,
        "inventory_csv_sha256": sha256_file(inventory_path),
        "repository_summary_csv": repository_path.name,
        "repository_summary_csv_sha256": sha256_file(repository_path),
        "size_bins_csv": size_path.name,
        "size_bins_csv_sha256": sha256_file(size_path),
        "repository_verified_indexed_size_bins_csv": repository_size_path.name,
        "repository_verified_indexed_size_bins_csv_sha256": sha256_file(
            repository_size_path
        ),
    }
    with summary_path.open("w", encoding="utf-8") as handle:
        json.dump(summary, handle, indent=2, ensure_ascii=False)
        handle.write("\n")
    print(json.dumps({**printable, "summary": str(summary_path)}, indent=2))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (TimeoutError, ValueError, OSError) as error:
        print(f"ERROR: {error}", file=sys.stderr)
        raise SystemExit(2)
