#!/usr/bin/env python3
"""Message pressure, sends after opt-out, stale writes, holdout contamination and
quiet hours across every sender, from pseudonymous logs.

Usage:
  python3 scripts/message_pressure.py SENDS.csv [--prefs PREFS.csv] [--holdout HOLDOUT.csv]
      [--cap-24h N] [--cap-7d N] [--window rolling|calendar] [--quiet HH:MM-HH:MM]
      [--classes marketing] [--channels email,sms,push] [--trust-sync-on]
      [--grace-hours 1] [--latency-fail-hours 48] [--json]

SENDS.csv columns: profile_key, sent_at, channel, sender, journey_id,
message_class (marketing, transactional, service or internal, set by purpose from
the journey inventory, never from a platform's transactional flag; any other
value stops the run), topic (needed when PREFS.csv has topic scopes) and
optionally recipient_tz (an IANA name such as Europe/London). One row per
message sent, from every sender.
PREFS.csv columns: profile_key, changed_at, scope (global, topic:<id> or
channel:<name>), choice (off or on), source.
HOLDOUT.csv column: profile_key, and optionally assigned_at and released_at, so
that only sends inside each member's membership count. For an account-level
holdout, list every profile in each held-out account.
Timestamps: ISO 8601 with Z or an offset, Snowflake's default TIMESTAMP_TZ text
(2026-10-08 07:00:00.000 +0000) or BigQuery's CSV text (2026-10-08 07:00:00 UTC).
A timestamp with no offset is read as UTC and the output says so.

profile_key must be an internal ID or a salted hash. A key that contains @,
starts with + or has phone formatting stops the run with exit status 2, and so
does a bare 10 to 15 digit key on an SMS or WhatsApp row. The value is never
printed. Python 3.9+, standard library only. Reads files, calls no API, writes
nothing.

Checks (thresholds and reasons: references/caps-and-suppression.md,
references/consent-and-preferences.md, references/measurement-and-holdouts.md):
  FC05 observed-pressure    peak counted messages per recipient in 24 hours and 7 days
  FC04 cross-sender-budget  over-cap recipients that every single sender kept within the cap
  PC02 sends-after-opt-out  marketing sends more than --latency-fail-hours after an 'off' in a
                            matching scope, or at any time after a stale write reversed it
  PC03 latency evidence     marketing sends between --grace-hours and --latency-fail-hours
                            after an 'off' (sends inside the grace are listed as in flight)
  PC08 sms-consent-and-stop SMS or WhatsApp marketing sends after an opt-out, at any delay, and
                            texts of any other class after STOP (source sms_keyword) beyond one
                            confirmation within 5 minutes
  PC01 authority            an 'on' from a sync or import after an 'off' with no customer action
  MH03 contamination        marketing sends to holdout members while assigned
  FC07 quiet-hours          counted sends inside the quiet window in recipient local time
  JC07 evidence             journey pairs that reached the same profile within 24 hours

--classes and --channels set what the cap policy counts (FC05, FC04, FC07, JC07).
PC02, PC03, PC08 and MH03 always read every marketing-class send on every channel.
NOT CHECKED replaces a PASS the data cannot support: a topic opt-out that no
marketing send carries (topic IDs differ, or no topic column), or a preference
log or holdout list with no key in common with the send log (keys or salts
differ). The output prints how many keys each file shares with the send log.
Without --cap-24h and --cap-7d, recipients above a screen of 2 per rolling 24
hours or 5 per rolling 7 days are a WARN (a screen, not a recommendation).
Local-time checks need the system time zone database; where it is missing (some
Windows installs), sends are counted as having no usable recipient_tz and the
output says so.
Exit status: 1 when any FAIL, 2 on bad input or a contact detail, 0 otherwise.
"""
import argparse
import csv
import json
import math
import re
import sys
from collections import defaultdict
from datetime import datetime, timedelta, timezone

try:
    from zoneinfo import ZoneInfo
except ImportError:  # Python 3.8 or older
    ZoneInfo = None

CUSTOMER_SOURCES = {"preference_centre", "one_click", "link", "support", "in_app", "sms_keyword"}
SYNC_SOURCES = {"crm_sync", "warehouse_sync", "import", "api_sync"}
SCREEN_24H, SCREEN_7D = 2, 5
DAY, WEEK = timedelta(hours=24), timedelta(days=7)
SEPARATORS = re.compile(r"[\s\-().]")
BARE_PHONE = re.compile(r"\d{10,15}")
PHONE_CHANNELS = {"sms", "whatsapp"}
PHONE_SCOPES = {"channel:sms", "channel:whatsapp"}
CLASSES = {"marketing", "transactional", "service", "internal"}
CLASS_NAMES = "marketing, transactional, service or internal"
CONFIRM_MINUTES = 5  # one opt-out confirmation text this soon after STOP is not counted (audit default)
NOT_CHECKED = ("JC01 to JC06, JC08, JC09, TD01 to TD08, SG01 to SG06, FC01 to FC03, FC06, FC08, FC09, "
               "PC03 per sender (seed S04), PC04 to PC07, PC08 opt-in records, PC09, PC10, DL01 to DL17, MH01, MH02, "
               "MH04 to MH10")
NOTES = []


def bad_input(message):
    print("message_pressure: " + message, file=sys.stderr)
    sys.exit(2)


def looks_like_contact(value):
    """True for anything that looks like an email address or a phone number."""
    if "@" in value:
        return True
    compact = SEPARATORS.sub("", value)
    if re.fullmatch(r"\+\d{8,15}", compact):
        return True
    return bool(SEPARATORS.search(value.strip()) and re.fullmatch(r"\d{8,15}", compact))


def read_csv(path, required, label, phone_row=None):
    try:
        with open(path, newline="", encoding="utf-8-sig") as handle:
            reader = csv.DictReader(handle)
            header = [h.strip().lower() for h in (reader.fieldnames or [])]
            missing = [c for c in required if c not in header]
            if missing:
                bad_input("%s is missing columns: %s" % (label, ", ".join(missing)))
            rows = []
            for number, raw in enumerate(reader, start=2):
                row = {(k or "").strip().lower(): (v or "").strip() for k, v in raw.items() if k is not None}
                key = row.get("profile_key", "")
                if not key:
                    bad_input("row %d of %s has no profile_key" % (number, label))
                if looks_like_contact(key):
                    bad_input("row %d of %s: profile_key looks like an email address or phone number. Replace"
                              " contact details with internal IDs or salted hashes first. The value was not"
                              " printed or stored." % (number, label))
                if phone_row and phone_row(row) and BARE_PHONE.fullmatch(key):
                    bad_input("row %d of %s: profile_key on an SMS or WhatsApp row is 10 to 15 digits, which may"
                              " be a phone number that lost its + sign in a spreadsheet. Hash SMS keys before"
                              " export. The value was not printed or stored." % (number, label))
                row["_row"] = number
                rows.append(row)
    except (OSError, UnicodeDecodeError, csv.Error) as exc:
        bad_input("cannot read %s as a UTF-8 CSV: %s (save Excel files as CSV UTF-8)" % (path, exc))
    return rows


def parse_ts(text, where):
    """ISO 8601, plus the default text of Snowflake TIMESTAMP_TZ and BigQuery CSV exports."""
    value = text.strip()
    value = re.sub(r"\s+(UTC|GMT)$", "+00:00", value, flags=re.I)
    if value[-1:] in ("Z", "z"):
        value = value[:-1] + "+00:00"
    value = re.sub(r"\s+([+-]\d{2}:?\d{2})$", r"\1", value)
    value = re.sub(r"([+-]\d{2})(\d{2})$", r"\1:\2", value)
    value = re.sub(r"\.(\d+)", lambda m: "." + (m.group(1) + "000000")[:6], value, count=1)
    try:
        stamp = datetime.fromisoformat(value)
    except ValueError:
        bad_input("%s: '%s' is not a timestamp this script reads (ISO 8601 with an offset, Snowflake"
                  " TIMESTAMP_TZ text or BigQuery CSV text)" % (where, text))
    if stamp.tzinfo is None:
        if "naive" not in NOTES:
            NOTES.append("naive")
        stamp = stamp.replace(tzinfo=timezone.utc)
    return stamp.astimezone(timezone.utc)


_ZONES = {}


def zone(name):
    if not name or ZoneInfo is None:
        return None
    if name not in _ZONES:
        try:
            _ZONES[name] = ZoneInfo(name)
        except Exception:  # unknown zone name or no tz database on this machine
            _ZONES[name] = None
    return _ZONES[name]


def iso(stamp):
    return stamp.strftime("%Y-%m-%dT%H:%MZ")


def rolling_peak(times, span):
    peak, start = 0, 0
    for end in range(len(times)):
        while times[end] - times[start] >= span:
            start += 1
        peak = max(peak, end - start + 1)
    return peak


def calendar_peaks(sends):
    """Peak per local calendar day and per 7 local calendar days."""
    per_day = defaultdict(int)
    for send in sends:
        tz = zone(send["tz"])
        local = send["ts"].astimezone(tz) if tz else send["ts"]
        per_day[local.date()] += 1
    if not per_day:
        return 0, 0
    days = sorted(per_day)
    peak_day = max(per_day.values())
    peak_week = 0
    for day in days:
        total = sum(n for d, n in per_day.items() if timedelta(0) <= day - d < timedelta(days=7))
        peak_week = max(peak_week, total)
    return peak_day, peak_week


def peaks(sends, window):
    if window == "calendar":
        return calendar_peaks(sends)
    times = sorted(s["ts"] for s in sends)
    return rolling_peak(times, DAY), rolling_peak(times, WEEK)


def percentile(values, pct):
    if not values:
        return 0
    ordered = sorted(values)
    rank = max(1, math.ceil(pct / 100.0 * len(ordered)))
    return ordered[rank - 1]


def parse_quiet(text):
    match = re.fullmatch(r"(\d{1,2}):(\d{2})-(\d{1,2}):(\d{2})", text.strip())
    if not match:
        bad_input("--quiet must look like 21:00-08:00")
    h1, m1, h2, m2 = (int(x) for x in match.groups())
    if h1 > 23 or h2 > 23 or m1 > 59 or m2 > 59:
        bad_input("--quiet must look like 21:00-08:00")
    return h1 * 60 + m1, h2 * 60 + m2


def in_quiet(minutes, start, end):
    if start > end:
        return minutes >= start or minutes < end
    return start <= minutes < end


class Report:
    def __init__(self):
        self.lines = []
        self.findings = []

    def text(self, value=""):
        self.lines.append(value)

    def finding(self, level, rule, name, detail, examples=None):
        self.findings.append({"level": level, "rule": rule, "name": name, "detail": detail,
                              "examples": examples or []})
        self.lines.append("%s %s %s: %s" % (level, rule, name, detail))
        for example in examples or []:
            self.lines.append("  " + example)


def main():
    parser = argparse.ArgumentParser(description="Cross-sender message pressure and preference checks.")
    parser.add_argument("sends")
    parser.add_argument("--prefs")
    parser.add_argument("--holdout")
    parser.add_argument("--cap-24h", type=int)
    parser.add_argument("--cap-7d", type=int)
    parser.add_argument("--window", choices=["rolling", "calendar"], default="rolling")
    parser.add_argument("--quiet", help="quiet window in recipient local time, for example 21:00-08:00")
    parser.add_argument("--classes", default="marketing", help="comma-separated message classes the policy counts")
    parser.add_argument("--channels", default="", help="comma-separated channels the policy counts (default all)")
    parser.add_argument("--trust-sync-on", action="store_true",
                        help="let an 'on' from a sync or import lift an earlier 'off'")
    parser.add_argument("--grace-hours", type=float, default=1.0,
                        help="sends this soon after an 'off' are listed as in flight (default 1)")
    parser.add_argument("--latency-fail-hours", type=float, default=48.0,
                        help="sends later than this after an 'off' are PC02 FAIL (default 48)")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()
    if args.grace_hours < 0 or args.latency_fail_hours <= args.grace_hours:
        bad_input("--latency-fail-hours must be above --grace-hours, and --grace-hours 0 or more")

    classes = {c.strip().lower() for c in args.classes.split(",") if c.strip()}
    channels = {c.strip().lower() for c in args.channels.split(",") if c.strip()}
    for value in (args.cap_24h, args.cap_7d):
        if value is not None and value < 1:
            bad_input("caps must be 1 or more")
    policy = args.cap_24h is not None or args.cap_7d is not None
    cap24 = args.cap_24h if args.cap_24h is not None else (None if policy else SCREEN_24H)
    cap7 = args.cap_7d if args.cap_7d is not None else (None if policy else SCREEN_7D)
    over_level = "FAIL" if policy else "WARN"

    raw = read_csv(args.sends, ["profile_key", "sent_at", "channel", "sender", "journey_id", "message_class"],
                   args.sends, lambda row: row.get("channel", "").lower() in PHONE_CHANNELS)
    if not raw:
        bad_input("no data rows in %s" % args.sends)
    bad_classes = sorted(classes - CLASSES)
    if bad_classes:
        bad_input("--classes takes %s; got %s" % (CLASS_NAMES, ", ".join(bad_classes)))
    unknown = defaultdict(list)
    for row in raw:
        if row["message_class"].lower() not in CLASSES:
            unknown[row["message_class"].lower()].append(row["_row"])
    if unknown:
        bad_input("message_class must be %s, set by purpose from the journey inventory (never from a platform's"
                  " transactional flag). Found: %s. Classify those journeys in the inventory and rebuild the send"
                  " log." % (CLASS_NAMES, "; ".join(
                      "%s (%d row%s, first row %d)" % (
                          "blank" if not value else ("a value that looks like a contact detail, not printed"
                                                     if looks_like_contact(value) else "'%s'" % value),
                          len(rows), "" if len(rows) == 1 else "s", rows[0])
                      for value, rows in sorted(unknown.items()))))
    has_topic_column = "topic" in raw[0]
    sends = []
    for row in raw:
        sends.append({
            "key": row["profile_key"],
            "ts": parse_ts(row["sent_at"], "row %d of %s" % (row["_row"], args.sends)),
            "channel": row["channel"].lower(),
            "sender": row["sender"].lower(),
            "journey": row["journey_id"],
            "cls": row["message_class"].lower(),
            "topic": row.get("topic", "").lower(),
            "tz": row.get("recipient_tz", ""),
        })
    counted = [s for s in sends if s["cls"] in classes and (not channels or s["channel"] in channels)]
    by_profile = defaultdict(list)
    for send in counted:
        by_profile[send["key"]].append(send)
    for values in by_profile.values():
        values.sort(key=lambda s: s["ts"])
    optional = [s for s in sends if s["cls"] == "marketing"]
    optional_by_profile = defaultdict(list)
    for send in optional:
        optional_by_profile[send["key"]].append(send)
    profiles = sorted({s["key"] for s in sends})

    out = Report()
    first, last = min(s["ts"] for s in sends), max(s["ts"] for s in sends)
    senders = sorted({s["sender"] for s in sends})
    out.text("message_pressure.py: %d sends from %s to %s (UTC), %d profiles, senders %s" % (
        len(sends), first.date().isoformat(), last.date().isoformat(), len(profiles), ", ".join(senders)))
    class_counts = defaultdict(int)
    for send in sends:
        class_counts[send["cls"]] += 1
    out.text("Counted: classes %s%s, %d sends. Not counted: %s." % (
        ", ".join(sorted(classes)), (" on channels " + ", ".join(sorted(channels))) if channels else "",
        len(counted), ", ".join("%s %d" % (c, n) for c, n in sorted(class_counts.items()) if c not in classes)
        or "none"))
    out.text("PC02, PC03, PC08 and MH03 read every marketing-class send on every channel: %d sends. PC08 also reads"
             " texts of every class after STOP." % len(optional))
    day = "a rolling 24 hours" if args.window == "rolling" else "a local calendar day"
    week = "a rolling 7 days" if args.window == "rolling" else "7 local calendar days"
    if policy:
        out.text("Caps (policy): %s per %s, %s per %s." % (
            cap24 if cap24 else "no cap", day.replace("a ", "", 1), cap7 if cap7 else "no cap",
            week.replace("a ", "", 1)))
    else:
        out.text("No caps passed: screening at %d per %s and %d per %s (a screen, not a recommendation)."
                 % (SCREEN_24H, day.replace("a ", "", 1), SCREEN_7D, week.replace("a ", "", 1)))
    if args.window == "calendar":
        missing_tz = sum(1 for s in counted if not zone(s["tz"]))
        if missing_tz:
            out.text("Calendar mode: %d counted sends had no usable recipient_tz and were counted on UTC days."
                     % missing_tz)

    # FC05 and FC04
    out.text("")
    out.text("FC05 observed-pressure")
    totals, peak24, peak7 = {}, {}, {}
    for key in profiles:
        mine = by_profile.get(key, [])
        totals[key] = len(mine)
        peak24[key], peak7[key] = peaks(mine, args.window) if mine else (0, 0)
    reached = [k for k in profiles if totals[k]]
    out.text("  Recipients with counted sends: %d of %d profiles in the log (distribution below covers these)."
             % (len(reached), len(profiles)))
    for title, values in (("Counted messages per recipient in the log", totals),
                          ("Peak in %s" % day.replace("a ", "any ", 1), peak24),
                          ("Peak in any %s" % week.replace("a ", "", 1), peak7)):
        series = [values[k] for k in reached]
        out.text("  %s: p50 %d, p90 %d, p99 %d, max %d" % (
            title, percentile(series, 50), percentile(series, 90), percentile(series, 99), max(series or [0])))
    over24 = sorted(k for k in profiles if cap24 and peak24[k] > cap24)
    over7 = sorted(k for k in profiles if cap7 and peak7[k] > cap7)
    if cap24:
        if over24:
            out.finding(over_level, "FC05", "observed-pressure", "%d recipient%s above %d in %s: %s" % (
                len(over24), "" if len(over24) == 1 else "s", cap24, day,
                ", ".join("%s (%d)" % (k, peak24[k]) for k in over24)))
        else:
            out.finding("PASS", "FC05", "observed-pressure", "no recipient above %d in %s."
                        % (cap24, day))
    if cap7:
        if over7:
            out.finding(over_level, "FC05", "observed-pressure", "%d recipient%s above %d in %s: %s" % (
                len(over7), "" if len(over7) == 1 else "s", cap7, week,
                ", ".join("%s (%d)" % (k, peak7[k]) for k in over7)))
        else:
            out.finding("PASS", "FC05", "observed-pressure", "no recipient above %d in %s."
                        % (cap7, week))

    out.text("")
    out.text("FC04 cross-sender-budget")
    over_any = sorted(set(over24) | set(over7))
    within_all = 0
    examples = []
    for key in over_any:
        per_sender = defaultdict(list)
        for send in by_profile[key]:
            per_sender[send["sender"]].append(send)
        parts, inside = [], True
        for sender in sorted(per_sender):
            s24, s7 = peaks(per_sender[sender], args.window)
            parts.append("%s %d in 24h, %d in 7d" % (sender, s24, s7))
            if (key in over24 and s24 > cap24) or (key in over7 and s7 > cap7):
                inside = False
        within_all += 1 if inside else 0
        examples.append("%s: %s" % (key, "; ".join(parts)))
    if not over_any:
        out.text("  No over-cap recipients, so no cross-sender split to read.")
    elif len(senders) < 2:
        out.text("  Only one sender in the log; the cross-sender check needs every sender's sends.")
    elif within_all:
        out.finding(over_level, "FC04", "cross-sender-budget",
                    "cross-sender: %d of %d over-cap recipients were within every sender's own cap (no shared"
                    " budget across senders)" % (within_all, len(over_any)), examples)
    else:
        out.finding("PASS", "FC04", "cross-sender-budget",
                    "cross-sender: 0 of %d over-cap recipients were within every sender's own cap; in each case"
                    " one sender went over alone, so fix that sender's cap first (FC02, FC05)" % len(over_any),
                    examples)

    # PC02, PC03 evidence, PC08 and PC01
    out.text("")
    out.text("PC02 sends-after-opt-out, PC03 latency evidence, PC08 texts after opt-out and PC01 authority")
    prefs = None
    if not args.prefs:
        out.text("NOT CHECKED PC02, PC03, PC08 and PC01: no --prefs file.")
    else:
        prefs = read_csv(args.prefs, ["profile_key", "changed_at", "scope", "choice", "source"], args.prefs,
                         lambda row: row.get("scope", "").lower() in PHONE_SCOPES)
        if not prefs:
            out.text("NOT CHECKED PC02, PC03, PC08 and PC01: %s has no data rows." % args.prefs)
    if prefs:
        events = defaultdict(list)
        for row in prefs:
            scope = row["scope"].lower()
            if not (scope == "global" or re.fullmatch(r"(topic|channel):\S+", scope)):
                bad_input("row %d of %s: scope '%s' must be global, topic:<id> or channel:<name>"
                          % (row["_row"], args.prefs, row["scope"]))
            choice = row["choice"].lower()
            if choice not in ("off", "on"):
                bad_input("row %d of %s: choice must be off or on" % (row["_row"], args.prefs))
            events[row["profile_key"]].append({
                "ts": parse_ts(row["changed_at"], "row %d of %s" % (row["_row"], args.prefs)),
                "scope": scope, "choice": choice, "source": row["source"].lower(), "kind": 0})
        pref_keys = set(events)
        shared_keys = pref_keys & set(profiles)
        out.text("  Keys: %d of %d preference-log profiles appear in the send log." % (len(shared_keys), len(pref_keys)))
        off_topics = sorted({e["scope"] for changes in events.values() for e in changes
                             if e["choice"] == "off" and e["scope"].startswith("topic:")})
        send_topics = {s["topic"] for s in optional if s["topic"]}
        unmatched_topics = [scope for scope in off_topics if scope.split(":", 1)[1] not in send_topics]
        other_texts = defaultdict(list)
        for send in sends:
            if send["cls"] != "marketing" and send["channel"] in PHONE_CHANNELS:
                other_texts[send["key"]].append(send)
        late, slow, in_flight, texts, stale, trusted = [], [], [], [], [], []
        confirmations = 0
        for key in sorted(set(events) | set(optional_by_profile) | set(other_texts)):
            timeline = sorted(events.get(key, []) + [dict(s, kind=1) for s in optional_by_profile.get(key, [])]
                              + [dict(s, kind=1) for s in other_texts.get(key, [])],
                              key=lambda e: (e["ts"], e["kind"]))
            active = {}
            for item in timeline:
                if item["kind"] == 0:
                    if item["choice"] == "off":
                        entry = active.setdefault(item["scope"], {"since": item["ts"], "source": item["source"],
                                                                  "reversed": None, "stop_at": None,
                                                                  "confirmed": False})
                        if item["source"] == "sms_keyword" and entry["stop_at"] is None:
                            entry["stop_at"] = item["ts"]
                    elif item["scope"] in active:
                        entry = active[item["scope"]]
                        if item["source"] in CUSTOMER_SOURCES:
                            del active[item["scope"]]
                            continue
                        kind = "sync or import" if item["source"] in SYNC_SOURCES else "unclassified source"
                        text = "%s %s: off %s %s, on %s %s (%s, %.1f hours later)" % (
                            key, item["scope"], iso(entry["since"]), entry["source"], iso(item["ts"]),
                            item["source"], kind, (item["ts"] - entry["since"]).total_seconds() / 3600)
                        if args.trust_sync_on:
                            trusted.append(text)
                            del active[item["scope"]]
                        else:
                            stale.append(text)
                            if entry["reversed"] is None:
                                entry["reversed"] = item["ts"]
                    continue
                if item["cls"] != "marketing":
                    # A text of another class: only a STOP in its channel scope or the global scope covers it.
                    scope = next((s for s in ("channel:" + item["channel"], "global")
                                  if s in active and active[s]["stop_at"] is not None), None)
                    if not scope:
                        continue
                    entry = active[scope]
                    hours = (item["ts"] - entry["stop_at"]).total_seconds() / 3600
                    if hours * 60 <= CONFIRM_MINUTES and not entry["confirmed"]:
                        entry["confirmed"] = True
                        confirmations += 1
                        continue
                    texts.append({"item": item, "scope": scope, "since": entry["stop_at"], "source": "sms_keyword",
                                  "reversed": None, "hours": hours})
                    continue
                scopes = [s for s in ("global", "topic:" + item["topic"] if item["topic"] else "",
                                      "channel:" + item["channel"]) if s in active]
                if not scopes:
                    continue
                # The worst matching scope decides: a stale write first, then the earliest opt-out, so a
                # fresh global opt-out cannot hide a topic opt-out that was ignored for days.
                matched = min(scopes, key=lambda s: (active[s]["reversed"] is None, active[s]["since"]))
                entry = active[matched]
                hours = (item["ts"] - entry["since"]).total_seconds() / 3600
                record = {"item": item, "scope": matched, "since": entry["since"], "source": entry["source"],
                          "reversed": entry["reversed"], "hours": hours}
                if item["channel"] in PHONE_CHANNELS:
                    texts.append(record)
                elif entry["reversed"] is not None or hours > args.latency_fail_hours:
                    late.append(record)
                elif hours > args.grace_hours:
                    slow.append(record)
                else:
                    in_flight.append(record)

        def per_sender(records):
            counts = defaultdict(int)
            for record in records:
                counts[record["item"]["sender"]] += 1
            return ", ".join("%s %d" % (sender, n) for sender, n in sorted(counts.items()))

        def example(record):
            item, hours = record["item"], record["hours"]
            delay = ("%d minutes" % round(hours * 60)) if hours < 1 else ("%.1f hours" % hours)
            text = "%s %s %s %s%s, %s after the change at %s (%s)" % (
                item["key"], item["sender"], item["journey"], record["scope"],
                "" if item["cls"] == "marketing" else " [%s]" % item["cls"], delay, iso(record["since"]),
                record["source"])
            if record["reversed"] is not None:
                text += "; a stale write reversed it at %s" % iso(record["reversed"])
            return text

        limit = "%g" % args.latency_fail_hours
        if not shared_keys:
            out.finding("NOT CHECKED", "PC02", "sends-after-opt-out",
                        "none of the %d preference-log profiles appears in the send log, so PC02, PC03 and PC08"
                        " cannot be read: the two files use different keys or salts, or none of these people"
                        " received anything. Compare one key in both files, then rerun." % len(pref_keys))
        else:
            for scope in unmatched_topics:
                out.finding("NOT CHECKED", "PC02", scope,
                            "no marketing send in the log carries this topic%s; map send topics to the preference"
                            " IDs, then rerun." % ("" if has_topic_column else " (the send log has no topic column)"))
            if late:
                out.finding("FAIL", "PC02", "sends-after-opt-out",
                            "%d marketing send%s more than %s hours after a restrictive change in %s scope, or after"
                            " a stale write reversed it (%s)" % (
                                len(late), "" if len(late) == 1 else "s", limit,
                                "its" if len(late) == 1 else "their", per_sender(late)),
                            [example(r) for r in late[:5]])
            else:
                out.finding("PASS", "PC02", "sends-after-opt-out", "%sno marketing send more than %s hours after a"
                            " restrictive change in its scope, and none after a stale write%s." % (
                                "for global and channel scopes only: " if unmatched_topics else "", limit,
                                "; the topic scopes above are NOT CHECKED" if unmatched_topics else ""))
            if slow:
                out.finding("WARN", "PC03", "unsubscribe-latency",
                            "%d marketing send%s between %g and %s hours after a restrictive change (%s): suppression"
                            " is slow in that sender. Measure it with seed S04." % (
                                len(slow), "" if len(slow) == 1 else "s", args.grace_hours, limit, per_sender(slow)),
                            [example(r) for r in slow[:5]])
            if in_flight:
                out.finding("INFO", "PC02", "in-flight",
                            "%d marketing send%s within %g hour%s of a restrictive change, read as a batch already in"
                            " flight, not a failure (%s)" % (
                                len(in_flight), "" if len(in_flight) == 1 else "s", args.grace_hours,
                                "" if args.grace_hours == 1 else "s", per_sender(in_flight)),
                            [example(r) for r in in_flight[:5]])
            if texts:
                marketing_texts = sum(1 for r in texts if r["item"]["cls"] == "marketing")
                out.finding("FAIL", "PC08", "sms-consent-and-stop",
                            "%d SMS or WhatsApp send%s after an opt-out in a matching scope, with no grace period:"
                            " %d marketing at any delay, %d of other classes after STOP beyond one confirmation"
                            " within %d minutes (%s)" % (
                                len(texts), "" if len(texts) == 1 else "s", marketing_texts,
                                len(texts) - marketing_texts, CONFIRM_MINUTES, per_sender(texts)),
                            [example(r) for r in texts[:5]])
            elif any(s["channel"] in PHONE_CHANNELS for s in sends):
                out.finding("PASS", "PC08", "sms-consent-and-stop", "no SMS or WhatsApp marketing send after an"
                            " opt-out in a matching scope, and no other text after STOP beyond one confirmation"
                            " within %d minutes." % CONFIRM_MINUTES)
            if confirmations:
                out.text("  %d text%s within %d minutes of a STOP read as the opt-out confirmation and not counted."
                         % (confirmations, "" if confirmations == 1 else "s", CONFIRM_MINUTES))
        if stale:
            out.finding("FAIL", "PC01", "authority",
                        "%d suspected stale write%s: an 'on' from a sync, import or unclassified source after an"
                        " 'off' with no customer action between; restriction kept" % (
                            len(stale), "" if len(stale) == 1 else "s"), stale[:5])
        else:
            out.finding("PASS", "PC01", "authority", "no 'on' from a sync or import reversed an 'off'.")
        if trusted:
            out.text("  --trust-sync-on: %d sync or import 'on' change%s lifted a restriction:" % (
                len(trusted), "" if len(trusted) == 1 else "s"))
            for text in trusted[:5]:
                out.text("    " + text)

    # MH03
    out.text("")
    out.text("MH03 contamination")
    if not args.holdout:
        out.text("NOT CHECKED MH03: no --holdout file.")
    else:
        rows = read_csv(args.holdout, ["profile_key"], args.holdout)
        if not rows:
            out.text("NOT CHECKED MH03: %s has no data rows." % args.holdout)
        dated = bool(rows) and ("assigned_at" in rows[0] or "released_at" in rows[0])
        members = defaultdict(list)
        for row in rows:
            where = "row %d of %s" % (row["_row"], args.holdout)
            start = parse_ts(row["assigned_at"], where) if row.get("assigned_at") else None
            end = parse_ts(row["released_at"], where) if row.get("released_at") else None
            if start and end and end <= start:
                bad_input("%s: released_at must be after assigned_at" % where)
            members[row["profile_key"]].append((start, end))

        def assigned(send):
            return any((start is None or send["ts"] >= start) and (end is None or send["ts"] < end)
                       for start, end in members.get(send["key"], []))

        listed = [s for s in optional if s["key"] in members]
        hits = [s for s in listed if assigned(s)]
        seen = set(members) & set(profiles)
        if rows:
            out.text("  Keys: %d of %d holdout members appear in the send log." % (len(seen), len(members)))
            if hits:
                counts = defaultdict(int)
                for send in hits:
                    counts[send["sender"]] += 1
                out.finding("FAIL", "MH03", "contamination", "%d marketing send%s to holdout members%s (%s; holdout"
                            " list of %d)" % (len(hits), "" if len(hits) == 1 else "s",
                                             " while assigned" if dated else "",
                                             ", ".join("%s %d" % (s, n) for s, n in sorted(counts.items())),
                                             len(members)),
                            ["%s %s %s %s" % (s["key"], s["sender"], s["journey"], iso(s["ts"])) for s in hits[:5]])
            elif not seen:
                out.finding("NOT CHECKED", "MH03", "contamination",
                            "none of the %d holdout members appears in the send log: the two files use different keys"
                            " or salts, or no member received anything, not even a required message (check MH04)."
                            " Compare one key in both files, then rerun." % len(members))
            else:
                out.finding("PASS", "MH03", "contamination", "no marketing send to any of %d holdout members%s." % (
                    len(members), " while assigned" if dated else ""))
            if dated and len(listed) > len(hits):
                out.text("  %d marketing send%s to listed members fell outside their membership dates and were not"
                         " counted." % (len(listed) - len(hits), "" if len(listed) - len(hits) == 1 else "s"))
            if not dated:
                out.text("  Membership dates not given: check the earliest flagged send against the assignment date"
                         " (assigned_at and released_at columns limit the count to membership).")

    # FC07
    out.text("")
    out.text("FC07 quiet-hours")
    if not args.quiet:
        out.text("NOT CHECKED FC07: no --quiet window.")
    else:
        start, end = parse_quiet(args.quiet)
        inside, unknown = [], 0
        for send in counted:
            tz = zone(send["tz"])
            if not tz:
                unknown += 1
                continue
            local = send["ts"].astimezone(tz)
            if in_quiet(local.hour * 60 + local.minute, start, end):
                inside.append((send, local))
        sms = [x for x in inside if x[0]["channel"] == "sms"]
        examples = ["%s %s %s %s %s = %s %s" % (s["key"], s["sender"], s["journey"], s["channel"], iso(s["ts"]),
                                                local.strftime("%Y-%m-%d %H:%M"), s["tz"]) for s, local in inside[:5]]
        if sms:
            out.finding("FAIL", "FC07", "quiet-hours", "%d SMS inside %s recipient local time" % (
                len(sms), args.quiet), examples)
        if len(inside) > len(sms):
            out.finding("WARN", "FC07", "quiet-hours", "%d counted non-SMS send%s inside %s recipient local time"
                        % (len(inside) - len(sms), "" if len(inside) - len(sms) == 1 else "s", args.quiet),
                        [] if sms else examples)
        if not inside:
            out.finding("PASS", "FC07", "quiet-hours", "no counted send inside %s recipient local time%s." % (
                args.quiet, " among sends with a timezone" if unknown else ""))
        if unknown:
            out.text("  %d counted sends had no usable recipient_tz and were not checked (TD08)." % unknown)

    # JC07 evidence
    out.text("")
    out.text("JC07 collisions (evidence for inventory findings)")
    pairs = defaultdict(set)
    for key, mine in by_profile.items():
        for i, send in enumerate(mine):
            j = i - 1
            while j >= 0 and send["ts"] - mine[j]["ts"] < DAY:
                if mine[j]["journey"] != send["journey"]:
                    pairs[tuple(sorted((mine[j]["journey"], send["journey"])))].add(key)
                j -= 1
    ranked = sorted(pairs.items(), key=lambda item: (-len(item[1]), item[0]))
    if ranked:
        out.text("  Journey pairs that reached the same profile within 24 hours (counted sends):")
        for (a, b), keys in ranked[:5]:
            out.text("    %s + %s: %d profile%s" % (a, b, len(keys), "" if len(keys) == 1 else "s"))
    else:
        out.text("  No two journeys reached the same profile within 24 hours.")

    if "naive" in NOTES:
        out.text("")
        out.text("Note: some timestamps had no offset and were read as UTC.")
    counts = {level: sum(1 for f in out.findings if f["level"] == level)
              for level in ("FAIL", "WARN", "PASS", "NOT CHECKED")}
    summary = "%d FAIL, %d WARN, %d PASS%s. Nothing was changed." % (
        counts["FAIL"], counts["WARN"], counts["PASS"],
        (", %d NOT CHECKED" % counts["NOT CHECKED"]) if counts["NOT CHECKED"] else "")
    out.text("")
    out.text(summary)
    out.text("Not checked by this script: %s." % NOT_CHECKED)
    if args.json:
        print(json.dumps({"findings": out.findings, "counts": counts, "summary": summary,
                          "journey_pairs": [{"pair": list(p), "profiles": len(k)} for p, k in ranked[:5]],
                          "not_checked": NOT_CHECKED}, indent=2))
    else:
        print("\n".join(out.lines))
    sys.exit(1 if counts["FAIL"] else 0)


if __name__ == "__main__":
    main()