#!/usr/bin/env python3
"""Read-only probe for the epic #3873 derived-field consistency invariant (Metric 2).

Checks that stored derived fields match their definitions, in three sections:

  quote-money    Quote rollup headers (TCV_Amount_c__c, NetFees__c) vs the
                 aggregate their installed RollupConfig declares.
  order-headers  Order rollup headers (Net_Fees__c, TotalAmount,
                 Count_Product_Lines__c) vs their installed RollupConfig.
  opp-rate       Non-USD Opportunity saved after --since must have Rate__c set
                 (the before-save rate flow stamps it on every save). Without
                 --since the blank non-USD count is reported as a dormant
                 backlog, not a violation.

The probe's coverage is the explicit field list in CHECKED_ROLLUPS — nothing
more. For those fields, the child object, summary column, relationship field,
and calc type are read from the runtime (`_System_Field.RollupConfig`) at
probe time, never restated here: a re-source that changes a checked rollup's
summary column changes what this probe verifies, with no probe edit. Fields
absent from CHECKED_ROLLUPS (e.g. the filtered Cash_Collected accumulators
and the Invoice_Collection/Revenue-sourced Order fields) are not verified at
all. The probe fails closed when a checked field has no
parseable rollup config, an unsupported calc type, or a filtered rollup (it
cannot apply raw SQL filters client-side).

Joins are computed client-side after paging the raw rows out over the REST
query endpoint, so the probe does not depend on SOQL aggregate/semi-join
support. Exit code is 1 when any section reports violations, 0 otherwise.

First run after the order-header backfill is the backfill acceptance; monthly
runs are the regression sweep. Quote/Order sections always sweep the full
tables (parent-side date scoping would let child edits on older parents drift
undetected); --since only classifies the opp-rate section.

Usage:
  scripts/probe_derived_field_consistency.py --base-url https://... \
      --username admin@example.com --password ... [--since 2026-08-01] [--json]
"""

from __future__ import annotations

import argparse
import datetime
import json
import os
import sys
from dataclasses import dataclass, field
from decimal import Decimal, InvalidOperation
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen

OAUTH_CLIENT_ID = "nexuscrm-ui"
OAUTH_CLIENT_SECRET = "nexuscrm-ui-internal"
DATA_BASE = "/services/data/v63.0"
PAGE_SIZE = 2000
MONEY_TOLERANCE = Decimal("0.01")


# --------------------------------------------------------------------------
# Pure comparison core (unit-tested; no I/O)
# --------------------------------------------------------------------------


def as_decimal(value: object) -> Decimal | None:
    """Parse a REST money value; None/'' stay None (blank is not zero here)."""
    if value is None or value == "":
        return None
    try:
        return Decimal(str(value))
    except InvalidOperation:
        return None


def display_id(record_id: str, show_ids: bool) -> str:
    """Redact record ids by default so probe output is safe to share or archive.

    Eight hex characters are enough for an operator holding DB access to
    re-locate the row (`Id LIKE '<prefix>%'`) without the output carrying a
    full customer record identifier.
    """
    if show_ids or len(record_id) <= 8:
        return record_id
    return f"{record_id[:8]}…"


@dataclass
class Violation:
    section: str
    record_id: str
    check: str
    header: object
    expected: object

    def as_dict(self, show_ids: bool = False) -> dict:
        return {
            "section": self.section,
            "id": display_id(self.record_id, show_ids),
            "check": self.check,
            "header": None if self.header is None else str(self.header),
            "expected": None if self.expected is None else str(self.expected),
        }


@dataclass
class SectionResult:
    section: str
    parents_checked: int = 0
    violations: list[Violation] = field(default_factory=list)
    info: dict = field(default_factory=dict)


def _sum_check(
    section: str,
    parent_id: str,
    check: str,
    header_raw: object,
    child_values: list[Decimal | None],
    tolerance: Decimal,
) -> Violation | None:
    """Header must equal the SUM over non-null child values, within tolerance.

    When every child value is null the engine may legitimately store either
    null or 0 in the header, so both are accepted.
    """
    header = as_decimal(header_raw)
    present = [v for v in child_values if v is not None]
    if not present:
        if header is None or header == 0:
            return None
        return Violation(section, parent_id, check, header_raw, "0 or blank")
    expected = sum(present, Decimal(0))
    if header is None or abs(header - expected) >= tolerance:
        return Violation(section, parent_id, check, header_raw, expected)
    return None


def _count_check(
    section: str,
    parent_id: str,
    check: str,
    header_raw: object,
    expected: int,
) -> Violation | None:
    """Header must equal the COUNT the rollup declares, exactly.

    A blank header is acceptable only when the expected count is 0 (the rollup
    may never have been computed for an empty child set); otherwise exact
    Decimal comparison, so a malformed fractional stored count (e.g. 1.5) is a
    violation, not truncated to a match.
    """
    header = as_decimal(header_raw)
    if header is None:
        if expected == 0:
            return None
        return Violation(section, parent_id, check, header_raw, expected)
    if header != expected:
        return Violation(section, parent_id, check, header_raw, expected)
    return None


_IDENTIFIER = __import__("re").compile(r"^[A-Za-z][A-Za-z0-9_]*$")


def _require_identifier(parent_object: str, field_name: str, kind: str, value: str) -> str:
    """Config values are interpolated into SOQL; anything that is not a plain
    identifier (whitespace, quotes, empty) must stop the probe loudly rather
    than produce a syntactically-broken or silently-wrong query."""
    if not _IDENTIFIER.match(value):
        raise SystemExit(
            f"{parent_object}.{field_name}: RollupConfig {kind} {value!r} is not a"
            " plain identifier; refusing to build a query from it"
        )
    return value


def _minmax_check(
    section: str,
    parent_id: str,
    check: str,
    header_raw: object,
    child_values: list[object],
    calc_type: str,
) -> Violation | None:
    """Header must equal the MIN/MAX over non-null child values, exactly.

    An empty (or all-null) child set stores null for MIN/MAX (engine default),
    so only a blank header passes then — a stale value after the last child
    was removed is a violation. Values compare in the Decimal domain when
    every operand parses as one (currency/number rollups), else as strings
    (ISO dates order lexicographically).
    """
    present = [v for v in child_values if v not in (None, "")]
    header_blank = header_raw in (None, "")
    if not present:
        if header_blank:
            return None
        return Violation(section, parent_id, check, header_raw, "blank")
    decimals = [as_decimal(v) for v in present]
    header_dec = as_decimal(header_raw)
    if all(d is not None for d in decimals) and header_dec is not None:
        expected = min(decimals) if calc_type == "MIN" else max(decimals)
        if header_dec != expected:
            return Violation(section, parent_id, check, header_raw, expected)
        return None
    values = [str(v) for v in present]
    expected = min(values) if calc_type == "MIN" else max(values)
    if header_blank or str(header_raw) != expected:
        return Violation(section, parent_id, check, header_raw, expected)
    return None


@dataclass(frozen=True)
class RollupSpec:
    """One checked rollup header, as declared by the runtime's RollupConfig."""

    parent_object: str
    parent_field: str
    calc_type: str  # SUM | COUNT
    summary_object: str
    summary_field: str  # "*" allowed for COUNT (row count)
    relationship_field: str


def parse_rollup_specs(
    parent_object: str,
    checked_fields: tuple[str, ...],
    field_rows: list[dict],
) -> list[RollupSpec]:
    """Turn `_System_Field` rows into RollupSpecs for the checked fields.

    Fails closed (SystemExit) on any condition under which the probe could
    otherwise verify something different from what the engine computes:
    a checked field with no rollup config, a calc type the probe does not
    implement, or a filtered rollup (raw SQL filters cannot be applied
    client-side, and ignoring one would misstate the expected aggregate).
    """
    if not checked_fields:
        raise SystemExit(f"{parent_object}: CHECKED_ROLLUPS lists no fields")
    names = [row.get("ApiName") for row in field_rows]
    duplicated = sorted({n for n in names if names.count(n) > 1})
    if duplicated:
        raise SystemExit(
            f"{parent_object}: multiple live _System_Field rows for"
            f" {', '.join(duplicated)}; refusing to pick one arbitrarily"
        )
    by_name = {row.get("ApiName"): row for row in field_rows}
    specs: list[RollupSpec] = []
    for field_name in checked_fields:
        row = by_name.get(field_name)
        if row is None:
            raise SystemExit(
                f"{parent_object}.{field_name}: no _System_Field row came back —"
                " either the field is gone (update CHECKED_ROLLUPS) or the"
                " probe user cannot read it"
            )
        config = row.get("RollupConfig")
        if isinstance(config, str):
            # The runtime may serialize the JSON column as a string (the
            # staging-inspection reader normalizes the same way).
            try:
                config = json.loads(config)
            except ValueError:
                raise SystemExit(
                    f"{parent_object}.{field_name}: RollupConfig is not valid JSON"
                ) from None
        if not isinstance(config, dict):
            raise SystemExit(
                f"{parent_object}.{field_name}: no RollupConfig on the runtime —"
                " the field is not a rollup; update CHECKED_ROLLUPS"
            )
        calc_type = str(config.get("calcType") or "").upper()
        summary_object = str(config.get("summaryObject") or "")
        summary_field = str(config.get("summaryField") or "")
        relationship_field = str(config.get("relationshipField") or "")
        if calc_type not in ("SUM", "COUNT", "MIN", "MAX"):
            raise SystemExit(
                f"{parent_object}.{field_name}: calcType {calc_type or '<blank>'!s}"
                " is not supported by this probe (SUM/COUNT/MIN/MAX only); update the probe"
            )
        if config.get("filter"):
            raise SystemExit(
                f"{parent_object}.{field_name}: rollup declares a filter, which this"
                " probe cannot apply client-side; update the probe before trusting it"
            )
        maintenance_mode = str(config.get("maintenanceMode") or "").lower()
        if maintenance_mode and maintenance_mode != "synchronous":
            raise SystemExit(
                f"{parent_object}.{field_name}: maintenanceMode {maintenance_mode!r}"
                " rides a delta queue whose legitimate lag is indistinguishable"
                " from drift; this probe only verifies synchronous rollups"
            )
        if not summary_object or not relationship_field:
            raise SystemExit(
                f"{parent_object}.{field_name}: RollupConfig is missing"
                " summaryObject/relationshipField; refusing to guess"
            )
        if not summary_field:
            raise SystemExit(
                f"{parent_object}.{field_name}: RollupConfig has no summaryField;"
                " refusing to guess"
            )
        if calc_type != "COUNT" and summary_field == "*":
            raise SystemExit(
                f"{parent_object}.{field_name}: {calc_type} over '*' is not a"
                " checkable aggregate; refusing to guess"
            )
        specs.append(
            RollupSpec(
                parent_object=parent_object,
                parent_field=field_name,
                calc_type=calc_type,
                summary_object=_require_identifier(
                    parent_object, field_name, "summaryObject", summary_object
                ),
                summary_field=(
                    summary_field
                    if summary_field == "*"
                    else _require_identifier(
                        parent_object, field_name, "summaryField", summary_field
                    )
                ),
                relationship_field=_require_identifier(
                    parent_object, field_name, "relationshipField", relationship_field
                ),
            )
        )
    return specs


def check_rollup_headers(
    section: str,
    parents: list[dict],
    children_by_parent: dict[str, list[dict]],
    specs: list[RollupSpec],
    tolerance: Decimal = MONEY_TOLERANCE,
) -> SectionResult:
    """Verify each parent's rollup headers against the spec-declared aggregates.

    Childless parents are checked too: an empty child set means a SUM header
    must be blank or 0 and a COUNT header blank or 0 — a stale nonzero header
    after the last child was removed is a violation, not a skip.
    """
    result = SectionResult(section)
    for parent in parents:
        children = children_by_parent.get(parent["Id"], [])
        result.parents_checked += 1
        for spec in specs:
            if spec.calc_type == "SUM":
                v = _sum_check(
                    section,
                    parent["Id"],
                    spec.parent_field,
                    parent.get(spec.parent_field),
                    [as_decimal(c.get(spec.summary_field)) for c in children],
                    tolerance,
                )
            elif spec.calc_type in ("MIN", "MAX"):
                v = _minmax_check(
                    section,
                    parent["Id"],
                    spec.parent_field,
                    parent.get(spec.parent_field),
                    [c.get(spec.summary_field) for c in children],
                    spec.calc_type,
                )
            else:
                # COUNT mirrors the engine's recompute exactly: COUNT(*) over
                # the child rows regardless of summaryField (the engine ignores
                # the named column for COUNT and installs canonicalize it to
                # '*'; see rollup_repository.go aggExpression handling).
                expected = len(children)
                v = _count_check(
                    section,
                    parent["Id"],
                    spec.parent_field,
                    parent.get(spec.parent_field),
                    expected,
                )
            if v:
                result.violations.append(v)
    return result


def check_opp_rate(opportunities: list[dict], since: str | None) -> SectionResult:
    """Non-USD opps saved on/after `since` must carry Rate__c (flow stamps on save).

    Without `since` there is no way to know which rows the flow has touched,
    so blanks are reported as a dormant backlog instead of violations.
    """
    result = SectionResult("opp-rate")
    dormant = 0
    for opp in opportunities:
        currency = opp.get("CurrencyIsoCode")
        if not currency or currency == "USD":
            continue
        result.parents_checked += 1
        if as_decimal(opp.get("Rate__c")) is not None:
            continue
        modified = str(opp.get("LastModifiedDate") or "")
        if since is not None and modified >= since:
            result.violations.append(
                Violation("opp-rate", opp["Id"], "Rate__c", None, f"set (saved {modified})")
            )
        else:
            dormant += 1
    result.info["dormant_blank_non_usd"] = dormant
    return result


def group_children(children: list[dict], parent_field: str) -> dict[str, list[dict]]:
    grouped: dict[str, list[dict]] = {}
    for child in children:
        parent_id = child.get(parent_field)
        if parent_id:
            grouped.setdefault(parent_id, []).append(child)
    return grouped


# --------------------------------------------------------------------------
# REST plumbing
# --------------------------------------------------------------------------


def _request(url: str, *, data: bytes | None, headers: dict[str, str], method: str) -> str:
    req = Request(url, data=data, headers=headers, method=method)
    try:
        with urlopen(req, timeout=60) as resp:
            return resp.read().decode()
    except HTTPError as exc:
        body = exc.read().decode(errors="replace")
        raise SystemExit(f"{method} {url} failed: http {exc.code}: {body[:500]}") from exc
    except URLError as exc:
        raise SystemExit(f"{method} {url} failed: {exc}") from exc


def login(base_url: str, username: str, password: str) -> str:
    payload = urlencode(
        {
            "grant_type": "password",
            "client_id": OAUTH_CLIENT_ID,
            "client_secret": OAUTH_CLIENT_SECRET,
            "username": username,
            "password": password,
        }
    ).encode()
    body = _request(
        f"{base_url.rstrip('/')}/services/oauth2/token",
        data=payload,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    token = (json.loads(body) or {}).get("access_token", "")
    if not token:
        raise SystemExit("oauth password grant returned no access_token")
    return str(token)


def query_all(base_url: str, token: str, soql_template: str) -> list[dict]:
    """Keyset-paginate `soql_template` (must contain {cursor}) ordered by Id.

    Termination is an empty page, never a short page: the server may cap a
    batch below our LIMIT, and inferring doneness from len(page) < LIMIT
    would silently truncate larger tables into a false-green probe.
    """
    rows: list[dict] = []
    cursor = ""
    while True:
        soql = soql_template.format(cursor=f"Id > '{cursor}' AND " if cursor else "")
        body = _request(
            f"{base_url.rstrip('/')}{DATA_BASE}/query?q={quote(soql)}",
            data=None,
            headers={"Authorization": f"Bearer {token}"},
            method="GET",
        )
        records = (json.loads(body) or {}).get("records") or []
        records = [r for r in records if isinstance(r, dict) and r.get("Id")]
        if not records:
            return rows
        rows.extend(records)
        cursor = records[-1]["Id"]


def _require_keys(rows: list[dict], keys: list[str], context: str) -> None:
    """A selected column that the REST layer drops (FLS, rename, projection
    quirk) would read as None everywhere and turn a drifted table into a false
    green; a missing key is a probe failure, not an empty value."""
    for row in rows:
        for key in keys:
            if key not in row:
                raise SystemExit(
                    f"{context}: column {key} was selected but is absent from a"
                    " returned row — cannot verify against it"
                )


# Which rollup headers the probe verifies, as {object: (section, fields)}.
# This list IS the probe's coverage. Only the field NAMES are stated here;
# everything about how each one aggregates (child object, summary column,
# relationship field, calc type) is read from the runtime at probe time so a
# re-source cannot silently diverge from what this probe checks.
CHECKED_ROLLUPS: dict[str, tuple[str, tuple[str, ...]]] = {
    "Quote": (
        "quote-money",
        ("TCV_Amount_c__c", "NetFees__c", "Overall_Start_Date__c", "Overall_End_Date__c"),
    ),
    "Order": (
        "order-headers",
        (
            "Net_Fees__c",
            "TotalAmount",
            "Count_Product_Lines__c",
            "Overall_Start_Date__c",
            "Overall_End_Date__c",
        ),
    ),
}


def fetch_rollup_specs(base_url: str, token: str) -> dict[str, list[RollupSpec]]:
    """Read the checked fields' RollupConfig from the runtime, fail-closed.

    One `_System_Field` sweep covers every checked object; rows are partitioned
    by ObjectId client-side. The name filter is the union of all checked field
    names, which may match unchecked same-named fields on another checked
    object — `parse_rollup_specs` only consults the rows for its own checked
    list, so extras are inert. Both metadata queries filter IsDeleted so a
    soft-deleted tombstone from an earlier re-source can neither shadow the
    live object row nor supply a retired rollup definition.
    """
    object_names = list(CHECKED_ROLLUPS)
    quoted_names = ", ".join(f"'{name}'" for name in object_names)
    object_rows = query_all(
        base_url,
        token,
        f"SELECT Id, ApiName FROM _System_Object WHERE {{cursor}}IsDeleted = false"
        f" AND ApiName IN ({quoted_names}) ORDER BY Id LIMIT {PAGE_SIZE}",
    )
    object_row_names = [row.get("ApiName") for row in object_rows]
    duplicated_objects = sorted({n for n in object_row_names if object_row_names.count(n) > 1})
    if duplicated_objects:
        raise SystemExit(
            "multiple live _System_Object rows for"
            f" {', '.join(duplicated_objects)}; refusing to pick one arbitrarily"
        )
    _require_keys(object_rows, ["Id", "ApiName"], "_System_Object")
    id_by_name = {row.get("ApiName"): row["Id"] for row in object_rows}
    missing = [name for name in object_names if name not in id_by_name]
    if missing:
        raise SystemExit(f"_System_Object rows not found for: {', '.join(missing)}")

    all_field_names = sorted(
        {name for _, fields in CHECKED_ROLLUPS.values() for name in fields}
    )
    quoted_fields = ", ".join(f"'{name}'" for name in all_field_names)
    quoted_object_ids = ", ".join(f"'{id_by_name[name]}'" for name in object_names)
    field_rows = query_all(
        base_url,
        token,
        "SELECT Id, ApiName, ObjectId, RollupConfig FROM _System_Field"
        f" WHERE {{cursor}}IsDeleted = false AND ObjectId IN ({quoted_object_ids})"
        f" AND ApiName IN ({quoted_fields}) ORDER BY Id LIMIT {PAGE_SIZE}",
    )
    # Key presence, not value: a checked non-rollup field returns RollupConfig
    # as JSON null with the key present, while an FLS-dropped column omits the
    # key entirely — the latter must not read as "not a rollup".
    _require_keys(field_rows, ["Id", "ApiName", "ObjectId", "RollupConfig"], "_System_Field")
    rows_by_object_id: dict[str, list[dict]] = {}
    for row in field_rows:
        rows_by_object_id.setdefault(str(row.get("ObjectId")), []).append(row)

    return {
        object_name: parse_rollup_specs(
            object_name,
            CHECKED_ROLLUPS[object_name][1],
            rows_by_object_id.get(id_by_name[object_name], []),
        )
        for object_name in object_names
    }


def run_probe(base_url: str, token: str, since: str | None) -> list[SectionResult]:
    # Quote/Order sections always sweep the full tables: scoping parents by
    # CreatedDate would let a child edit on an older parent drift undetected
    # (and a missed recompute does not bump the parent's LastModifiedDate, so
    # no parent timestamp scope is sound). The full sweep is a few thousand
    # rows; `since` only classifies the opp-rate section, where pre-deploy
    # rows are legitimately blank until their next save.
    specs_by_object = fetch_rollup_specs(base_url, token)

    results: list[SectionResult] = []
    for object_name, (section, _) in CHECKED_ROLLUPS.items():
        specs = specs_by_object[object_name]
        parent_fields = ", ".join(spec.parent_field for spec in specs)
        parents = query_all(
            base_url,
            token,
            f"SELECT Id, {parent_fields} FROM {object_name}"
            f" WHERE {{cursor}}IsDeleted = false AND Id != null ORDER BY Id LIMIT {PAGE_SIZE}",
        )
        _require_keys(parents, [spec.parent_field for spec in specs], object_name)
        # Checked rollups may aggregate different child tables (a legitimate
        # metadata evolution, not a probe-shape change): fetch each distinct
        # (summary_object, relationship_field) source once and merge the
        # violations into one section, counting each parent exactly once.
        merged = SectionResult(section)
        merged.parents_checked = len(parents)
        groups: dict[tuple[str, str], list[RollupSpec]] = {}
        for spec in specs:
            groups.setdefault(
                (spec.summary_object, spec.relationship_field), []
            ).append(spec)
        for (summary_object, relationship_field), group_specs in groups.items():
            summary_fields = sorted(
                {spec.summary_field for spec in group_specs if spec.summary_field != "*"}
            )
            # dict.fromkeys dedupes while preserving order: the summary field
            # may itself be Id (the pack-source shape of a COUNT rollup) or the
            # relationship field, and a duplicated projection must not reach
            # the query parser.
            child_select = ", ".join(
                dict.fromkeys(["Id", relationship_field, *summary_fields])
            )
            children = query_all(
                base_url,
                token,
                f"SELECT {child_select} FROM {summary_object}"
                f" WHERE {{cursor}}IsDeleted = false AND Id != null ORDER BY Id LIMIT {PAGE_SIZE}",
            )
            _require_keys(
                children,
                [relationship_field, *summary_fields],
                f"{object_name}<-{summary_object}",
            )
            partial = check_rollup_headers(
                section, parents, group_children(children, relationship_field), group_specs
            )
            merged.violations.extend(partial.violations)
        results.append(merged)

    opps = query_all(
        base_url,
        token,
        "SELECT Id, CurrencyIsoCode, Rate__c, LastModifiedDate FROM Opportunity"
        f" WHERE {{cursor}}IsDeleted = false AND Id != null ORDER BY Id LIMIT {PAGE_SIZE}",
    )
    results.append(check_opp_rate(opps, f"{since}T00:00:00" if since else None))
    return results


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------


def iso_date(raw: str) -> str:
    """Fail closed on a malformed --since: an unparsed cutoff compared
    lexicographically against timestamps would misclassify every blank
    non-USD opportunity as dormant and return a false green."""
    try:
        return datetime.date.fromisoformat(raw).isoformat()
    except ValueError as exc:
        raise argparse.ArgumentTypeError(
            f"--since must be an ISO date (YYYY-MM-DD), got {raw!r}"
        ) from exc


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--base-url", required=True)
    parser.add_argument("--username", default=os.environ.get("NEXUSCRM_PROBE_USERNAME", ""))
    parser.add_argument("--password", default=os.environ.get("NEXUSCRM_PROBE_PASSWORD", ""))
    parser.add_argument(
        "--since",
        type=iso_date,
        help="ISO date (YYYY-MM-DD): treat blank Rate__c on non-USD opportunities"
        " as a violation only when the opp was saved on/after this date (the rate"
        " flow stamps on save). Quote/Order sections always sweep the full tables.",
    )
    parser.add_argument("--json", action="store_true", help="emit machine-readable JSON")
    parser.add_argument(
        "--show-ids",
        action="store_true",
        help="opt in to full record ids in output (default: redacted to an 8-char prefix)",
    )
    args = parser.parse_args(argv)
    if not args.username or not args.password:
        parser.error(
            "--username/--password required (or NEXUSCRM_PROBE_USERNAME/NEXUSCRM_PROBE_PASSWORD)"
        )

    token = login(args.base_url, args.username, args.password)
    results = run_probe(args.base_url, token, args.since)

    total = sum(len(r.violations) for r in results)
    if args.json:
        print(
            json.dumps(
                {
                    "since": args.since,
                    "total_violations": total,
                    "sections": [
                        {
                            "section": r.section,
                            "parents_checked": r.parents_checked,
                            "violations": [v.as_dict(args.show_ids) for v in r.violations],
                            **({"info": r.info} if r.info else {}),
                        }
                        for r in results
                    ],
                },
                indent=2,
            )
        )
    else:
        for r in results:
            print(f"[{r.section}] parents checked: {r.parents_checked}, violations: {len(r.violations)}")
            for v in r.violations[:20]:
                print(
                    f"  {display_id(v.record_id, args.show_ids)} {v.check}:"
                    f" header={v.header!r} expected={v.expected!r}"
                )
            if len(r.violations) > 20:
                print(f"  ... and {len(r.violations) - 20} more")
            for key, value in r.info.items():
                print(f"  info {key}: {value}")
        print(f"TOTAL violations: {total}")
    return 1 if total else 0


if __name__ == "__main__":
    sys.exit(main())