#!/usr/bin/env python3
"""Rates a lifecycle journey inventory against the audit's coverage rules.

Usage:
  python3 scripts/inventory_audit.py INVENTORY.csv --model saas|ecommerce|app|leadgen
      [--not-applicable trial_expiry,renewal_notice] [--global-holdout yes|no]
      [--policy-caps-all-optional yes|no] [--as-of YYYY-MM-DD] [--json]

INVENTORY.csv uses the header of assets/journey-inventory-template.csv: one row
per live journey, recurring campaign, paused journey with entries in the window
and external owner (billing system, store platform, app store, sales team).
sends_window and steps_not_sending are optional; extra columns are allowed.
Values are compared in lower case. Counts may use thousands separators (1,240). Python 3.9+, standard library only. Reads one
file, calls no API and writes nothing.

Rules applied mechanically (thresholds and reasons: references/journey-coverage.md):
  JC01 expected-journeys      required types for the model have a live or external row
  JC02 coverage-ratio         entries / expected, rated only when expected is 100 or more
  JC03 stale-live-journey     live with 0 entries in the window (FAIL when the source count shows
                              occurrences; a blank expected_window asks for the count)
  JC04 exit-on-goal           2+ messages or any delay, and no exit on the goal
  JC05 recheck-after-wait     delay over 1 hour with no send-time recheck
  JC06 re-entry-matches-unit  re-entry blank, unknown, or once for a per-object unit
  JC07 collisions             two journeys of one type, or two marketing journeys on one trigger
  JC08 ownership              no owner
  JC09 live-journey-not-sending  live with entries, and 0 sends or a step that does not send (WARN
                              when the longest delay is as long as the 30-day window)
  TD01 authoritative-source   commercial or account journey triggered client-side
  FC02 config-coverage        live marketing journey not covered by the sender's cap
  PC07 transactional-misuse   promotional type classed transactional or sent to unsubscribed
                              users; a usage_limit alert outside marketing is a WARN to read
  MH02 persistent-holdout     live marketing journeys with no holdout exclusion

It also prints a coverage map by journey type and by lifecycle stage. A row's
stage column wins over the type's default stage.
Exit status: 1 when any FAIL, 2 on bad input, 0 otherwise.
"""
import argparse
import csv
import json
import math
import re
import sys
from collections import defaultdict
from datetime import date, datetime

# Expected journey types per business model. Must match the lists in
# references/journey-coverage.md exactly.
# Required entry: type -> (criticality, external owner allowed, when N/A)
MODELS = {
    "saas": {
        "required": {
            "activation_first_value": ("revenue-critical", False, ""),
            "trial_expiry": ("revenue-critical", False, "no trials"),
            "payment_failure": ("revenue-critical", True, ""),
            "renewal_notice": ("revenue-critical", True, "no annual or contract terms"),
            "account_security": ("service", True, ""),
        },
        "recommended": [
            "signup_completion", "onboarding_setup", "invite_collaborators", "usage_limit",
            "upgrade_or_expansion", "usage_decline", "cancellation", "winback", "product_updates",
        ],
    },
    "ecommerce": {
        "required": {
            "welcome_subscriber": ("revenue-critical", False, ""),
            "cart_or_checkout_abandonment": ("revenue-critical", False, ""),
            "order_and_shipping": ("service", True, ""),
            "winback": ("revenue-critical", False, ""),
            "sunset_or_repermission": ("deliverability-critical", False, ""),
        },
        "recommended": [
            "browse_abandonment", "post_purchase_cross_sell", "replenishment", "review_request",
            "back_in_stock", "vip_or_loyalty",
        ],
    },
    "app": {
        "required": {
            "onboarding_and_permission": ("", False, ""),
            "activation_first_value": ("revenue-critical", False, ""),
            "subscription_conversion": ("revenue-critical", False, "no subscriptions"),
            "billing_issue": ("revenue-critical", True, ""),
            "lapsed_reactivation": ("", False, ""),
            "account_security": ("service", True, ""),
        },
        "recommended": ["habit_or_streak", "winback_after_cancel", "feature_announcement"],
    },
    "leadgen": {
        "required": {
            "lead_welcome": ("", False, ""),
            "nurture_by_stage": ("revenue-critical", False, ""),
            "sales_handoff": ("revenue-critical", True, ""),
            "customer_onboarding": ("", False, ""),
        },
        "recommended": ["event_follow_up", "recycle_closed_lost", "renewal_or_expansion", "newsletter"],
    },
}

# Default lifecycle stage per journey type, for the coverage-by-stage view. A row's
# stage column wins. Must match references/journey-coverage.md.
STAGE_ORDER = ["onboarding", "activation", "conversion", "adoption", "retention", "expansion", "winback",
               "service"]
DEFAULT_STAGE = {
    "signup_completion": "onboarding", "onboarding_setup": "onboarding", "welcome_subscriber": "onboarding",
    "onboarding_and_permission": "onboarding", "lead_welcome": "onboarding", "customer_onboarding": "onboarding",
    "activation_first_value": "activation",
    "trial_expiry": "conversion", "subscription_conversion": "conversion",
    "cart_or_checkout_abandonment": "conversion", "browse_abandonment": "conversion",
    "back_in_stock": "conversion", "nurture_by_stage": "conversion", "sales_handoff": "conversion",
    "event_follow_up": "conversion",
    "invite_collaborators": "adoption", "product_updates": "adoption", "feature_announcement": "adoption",
    "habit_or_streak": "adoption", "review_request": "adoption", "newsletter": "adoption",
    "payment_failure": "retention", "billing_issue": "retention", "renewal_notice": "retention",
    "usage_decline": "retention", "cancellation": "retention", "replenishment": "retention",
    "vip_or_loyalty": "retention",
    "upgrade_or_expansion": "expansion", "post_purchase_cross_sell": "expansion",
    "renewal_or_expansion": "expansion",
    "winback": "winback", "winback_after_cancel": "winback", "lapsed_reactivation": "winback",
    "recycle_closed_lost": "winback", "sunset_or_repermission": "winback",
    "account_security": "service", "order_and_shipping": "service", "usage_limit": "service",
}

PROMOTIONAL = {
    "winback", "winback_after_cancel", "post_purchase_cross_sell", "upgrade_or_expansion", "product_updates",
    "feature_announcement", "newsletter", "browse_abandonment", "cart_or_checkout_abandonment",
    "back_in_stock", "vip_or_loyalty", "review_request", "nurture_by_stage", "event_follow_up",
    "recycle_closed_lost", "habit_or_streak",
}
# Service alerts that often carry an upgrade pitch: read the copy before rating PC07.
READ_THE_COPY = {"usage_limit"}
COMMERCIAL = {
    "trial_expiry", "payment_failure", "renewal_notice", "subscription_conversion", "billing_issue",
    "order_and_shipping", "cancellation",
}
ACCOUNT_STATE = {
    "activation_first_value", "signup_completion", "onboarding_setup", "invite_collaborators",
    "usage_limit", "usage_decline", "onboarding_and_permission", "lapsed_reactivation",
    "customer_onboarding",
}
CLIENT_SOURCES = {"client", "sdk"}
PER_OBJECT_UNITS = {
    "workspace", "account", "company", "invoice", "order", "trial", "subscription", "cart",
    "checkout", "opportunity", "deal", "ticket", "booking",
}

TEMPLATE_COLUMNS = [
    "platform", "journey_id", "journey_name", "status", "journey_type", "stage", "message_class",
    "channels", "trigger_type", "trigger_name", "trigger_source", "entry_unit", "recipient_role",
    "re_entry", "exit_on_goal", "goal_event", "message_steps", "max_delay_hours",
    "recheck_after_delay", "frequency_capped", "sends_to_unsubscribed", "holdout", "owner",
    "last_entry_date", "entries_window", "expected_window", "sends_window", "steps_not_sending", "notes",
]
REQUIRED_COLUMNS = [
    "platform", "journey_id", "journey_name", "status", "journey_type", "message_class",
    "trigger_name", "trigger_source", "re_entry", "exit_on_goal", "message_steps",
    "max_delay_hours", "recheck_after_delay", "frequency_capped", "sends_to_unsubscribed",
    "holdout", "owner", "entries_window", "expected_window",
]
ALLOWED = {
    "status": {"live", "paused", "draft", "archived", "external"},
    "message_class": {"marketing", "transactional", "service", "internal"},
    "trigger_source": {"server", "billing", "store", "client", "sdk", "sync", "import", "na", "unknown", ""},
    "exit_on_goal": {"yes", "no", "na", ""},
    "recheck_after_delay": {"yes", "no", "na", ""},
    "frequency_capped": {"yes", "no", "exempt", "na", ""},
    "sends_to_unsubscribed": {"yes", "no", "na", ""},
    "holdout": {"global", "journey", "none", ""},
}
RE_ENTRY = re.compile(r"^(once|per_unit|always|unknown|cooldown_\d+d|)$")
NUMERIC = ["message_steps", "max_delay_hours", "entries_window", "expected_window", "sends_window",
           "steps_not_sending"]

RULES = [
    ("JC01", "expected-journeys"), ("JC02", "coverage-ratio"), ("JC03", "stale-live-journey"),
    ("JC04", "exit-on-goal"), ("JC05", "recheck-after-wait"), ("JC06", "re-entry-matches-unit"),
    ("JC07", "collisions"), ("JC08", "ownership"), ("JC09", "live-journey-not-sending"),
    ("TD01", "authoritative-source"),
    ("FC02", "config-coverage"), ("PC07", "transactional-misuse"), ("MH02", "persistent-holdout"),
]
RULE_NAME = dict(RULES)
RULE_ORDER = {rule: i for i, (rule, _) in enumerate(RULES)}
LEVEL_ORDER = {"FAIL": 0, "WARN": 1, "PASS": 2, "INFO": 3}
NOT_CHECKED = ("TD02 to TD08, SG01 to SG06, FC01, FC03 to FC09, PC01 to PC06, PC08 to PC10, "
               "DL01 to DL17, MH01, MH03 to MH10")
STALE_DAYS = 30


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


THOUSANDS = re.compile(r"-?\d{1,3}(,\d{3})+(\.\d+)?")


def number(value):
    """A count or number of hours. Thousands separators (1,240) are allowed; a decimal comma is not."""
    if value == "":
        return None
    if THOUSANDS.fullmatch(value):
        value = value.replace(",", "")
    result = float(value)
    if not math.isfinite(result):  # nan or inf from an export would otherwise read as a ratio above 1.25
        raise ValueError(value)
    return result


def fmt_num(value):
    return str(int(value)) if value is not None and float(value).is_integer() else str(value)


def read_rows(path):
    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_COLUMNS if c not in header]
            if missing:
                bad_input("missing required columns: " + ", ".join(missing))
            rows = []
            for raw in reader:
                row = {}
                for key, value in raw.items():
                    if key is None:
                        continue
                    row[key.strip().lower()] = (value or "").strip()
                for column in TEMPLATE_COLUMNS:
                    row.setdefault(column, "")
                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))
    if not rows:
        bad_input("no data rows in %s" % path)
    return rows


def normalise(rows):
    problems = []
    for index, row in enumerate(rows, start=2):
        for column in TEMPLATE_COLUMNS:
            if column not in ("journey_name", "journey_id", "owner", "notes", "goal_event", "trigger_name"):
                row[column] = row[column].lower()
        if row["trigger_source"] == "":
            row["trigger_source"] = "unknown"
        if row["holdout"] == "":
            row["holdout"] = "none"
        for column, allowed in ALLOWED.items():
            if row[column] not in allowed:
                problems.append("row %d %s '%s' is not one of %s" % (
                    index, column, row[column], ", ".join(sorted(a for a in allowed if a))))
        if not RE_ENTRY.match(row["re_entry"]):
            problems.append("row %d re_entry '%s' is not once, per_unit, cooldown_<n>d, always or unknown"
                            % (index, row["re_entry"]))
        for column in NUMERIC:
            try:
                row["_" + column] = number(row[column])
            except ValueError:
                problems.append("row %d %s '%s' is not a number" % (index, column, row[column]))
        if row["last_entry_date"]:
            try:
                row["_last_entry"] = datetime.strptime(row["last_entry_date"], "%Y-%m-%d").date()
            except ValueError:
                problems.append("row %d last_entry_date '%s' is not YYYY-MM-DD" % (index, row["last_entry_date"]))
        else:
            row["_last_entry"] = None
        if not row["journey_type"]:
            problems.append("row %d journey_type is blank (use other when no type fits)" % index)
    if problems:
        bad_input("bad values:\n  " + "\n  ".join(problems[:40]))


def add(findings, level, rule, detail, row=None):
    findings.append({
        "level": level,
        "rule": rule,
        "name": RULE_NAME[rule],
        "journey_id": row["journey_id"] if row else "",
        "journey_name": row["journey_name"] if row else "",
        "platform": row["platform"] if row else "",
        "detail": detail,
    })


def label(row):
    return '%s "%s" (%s, %s)' % (row["journey_id"], row["journey_name"], row["platform"], row["status"])


def audit(rows, args):
    findings = []
    model = MODELS[args.model]
    not_applicable = {t.strip().lower() for t in args.not_applicable.split(",") if t.strip()}
    live = [r for r in rows if r["status"] == "live"]
    active = [r for r in rows if r["status"] in ("live", "external")]
    as_of = args.as_of

    # JC01 expected journeys and the coverage map
    by_type = defaultdict(list)
    for row in rows:
        by_type[row["journey_type"]].append(row)
    coverage = []
    for jtype, (critical, external_ok, na_when) in model["required"].items():
        covering = [r for r in by_type.get(jtype, []) if r["status"] in ("live", "external")]
        tag = critical or "required"
        if jtype in not_applicable:
            coverage.append("required %s [%s]: N/A (state the reason in the report)" % (jtype, tag))
            continue
        if covering:
            coverage.append("required %s [%s]: %s" % (jtype, tag, "; ".join(label(r) for r in covering)))
            parts = []
            for r in covering:
                if r["status"] == "external":
                    parts.append("%s (%s, external; keep one sample message or setting as evidence)"
                                 % (r["journey_id"], r["platform"]))
                else:
                    parts.append("%s (%s, live)" % (r["journey_id"], r["platform"]))
            add(findings, "PASS", "JC01", "required type %s covered by %s." % (jtype, ", ".join(parts)))
        else:
            coverage.append("required %s [%s]: MISSING" % (jtype, tag))
            hint = "P1" if critical in ("revenue-critical", "deliverability-critical") else "P2"
            detail = "required type %s (%s) has no live journey and no external owner. Priority hint %s." % (
                jtype, critical or "required", hint)
            if external_ok:
                detail += (" If another system sends it (billing system, store platform, app store, sales team),"
                           " add an external row with its owner and one sample message.")
            if na_when:
                detail += " Not applicable only when there are %s: pass --not-applicable %s and say why." % (
                    na_when, jtype)
            add(findings, "WARN", "JC01", detail)
    for jtype in model["recommended"]:
        covering = [r for r in by_type.get(jtype, []) if r["status"] in ("live", "external")]
        if jtype in not_applicable:
            coverage.append("recommended %s: N/A" % jtype)
        elif covering:
            coverage.append("recommended %s: %s" % (jtype, "; ".join(label(r) for r in covering)))
        else:
            coverage.append("recommended %s: missing" % jtype)
            add(findings, "INFO", "JC01", "recommended type %s has no journey." % jtype)
    known = set(model["required"]) | set(model["recommended"])
    others = sorted(t for t in by_type if t not in known)
    for jtype in others:
        coverage.append("other type %s: %s" % (jtype, "; ".join(label(r) for r in by_type[jtype])))

    # JC02 coverage ratio and JC03 stale live journeys
    ratios = []
    for row in live:
        entries, expected = row["_entries_window"], row["_expected_window"]
        if entries is None and row["_last_entry"] and as_of:
            if (as_of - row["_last_entry"]).days > STALE_DAYS:
                entries = 0.0
        if entries == 0:
            last = ""
            if row["_last_entry"]:
                last = "; last entry %s" % row["last_entry_date"]
                if as_of:
                    last += " (%d days before %s)" % ((as_of - row["_last_entry"]).days, as_of.isoformat())
            if expected and expected > 0:
                add(findings, "FAIL", "JC03",
                    "is live with 0 entries in the window while the source shows %s qualifying occurrences%s."
                    " Check for a renamed or retired trigger event and a broken entry filter." % (fmt_num(expected), last),
                    row)
            elif expected is None:
                add(findings, "WARN", "JC03",
                    "is live with 0 entries and no source count (expected_window blank)%s: get the source count;"
                    " FAIL if it shows qualifying occurrences, else WARN (dead or seasonal)." % last, row)
            else:
                add(findings, "WARN", "JC03",
                    "is live with 0 entries and no qualifying occurrences in the source%s: dead or seasonal."
                    " Ask the owner whether to retire it." % last, row)
            ratios.append("%s: 0 entries (see JC03)" % row["journey_id"])
            continue
        if entries is None or expected is None:
            ratios.append("%s: not rated (no %s count)" % (
                row["journey_id"], "entry" if entries is None else "expected"))
            continue
        if expected < 100:
            ratios.append("%s %s / %s: not rated (low volume, expected below 100)" % (
                row["journey_id"], fmt_num(entries), fmt_num(expected)))
            continue
        ratio = entries / expected
        shown = "%.2f" % ratio
        ratios.append("%s %s / %s = %s" % (row["journey_id"], fmt_num(entries), fmt_num(expected), shown))
        base = "entered %s of %s expected (ratio %s" % (fmt_num(entries), fmt_num(expected), shown)
        if 0.90 <= ratio <= 1.10:
            add(findings, "PASS", "JC02", base + ").", row)
        elif 0.75 <= ratio < 0.90:
            add(findings, "WARN", "JC02", base + ", below 0.90). Trace 5 missing occurrences end to end.", row)
        elif 1.10 < ratio <= 1.25:
            add(findings, "WARN", "JC02", base + ", above 1.10). Trace 5 extra entries end to end.", row)
        elif ratio < 0.75:
            add(findings, "FAIL", "JC02", base + ", below 0.75): trigger leakage. Split entries by source"
                " and trace 5 missing occurrences end to end.", row)
        else:
            add(findings, "FAIL", "JC02", base + ", above 1.25): duplicate or unintended entries. Check"
                " producers, retries and imports (TD02 to TD04).", row)

    # JC09 live journeys that enter people and send nothing, or have a step that does not send
    sending, not_rated = [], []
    for row in live:
        entries = row["_entries_window"]
        if not entries:
            continue
        sends, idle = row["_sends_window"], row["_steps_not_sending"]
        if sends is None and idle is None:
            not_rated.append(row["journey_id"])
        elif sends == 0 and (row["_max_delay_hours"] or 0) >= STALE_DAYS * 24:
            add(findings, "WARN", "JC09", "entered %s in the window and sent 0 messages, but its longest delay (%s"
                " hours) is as long as the %d-day window, so nobody may have reached a message yet. Check entrants"
                " older than the delay; FAIL if they received nothing." % (
                    fmt_num(entries), fmt_num(row["_max_delay_hours"]), STALE_DAYS), row)
        elif sends == 0:
            add(findings, "FAIL", "JC09", "entered %s in the window but sent 0 messages. Read each message step's"
                " status in the platform (labels: verify live) and confirm with seed S01." % fmt_num(entries), row)
        elif idle:
            add(findings, "FAIL", "JC09", "entered %s in the window, but %s of %s message steps sent nothing or"
                " sit in a non-sending status. Read each step's status (labels: verify live) and confirm with"
                " seed S01." % (fmt_num(entries), fmt_num(idle), fmt_num(row["_message_steps"] or 0)), row)
        else:
            sending.append(row["journey_id"])
    if sending:
        add(findings, "PASS", "JC09", "%d live journey%s with entries sent messages and report%s no step that"
            " does not send: %s. Step statuses are configuration reads; a seed (S01) confirms a doubtful one." % (
                len(sending), "" if len(sending) == 1 else "s", "s" if len(sending) == 1 else "",
                ", ".join(sending)))
    if not_rated:
        ratios.append("JC09 not rated (no sends_window or steps_not_sending): %s" % ", ".join(not_rated))

    for row in live:
        steps = row["_message_steps"] or 0
        delay = row["_max_delay_hours"] or 0
        unit = row["entry_unit"]
        # JC04 exit on goal
        if steps >= 2 or delay > 0:
            if row["exit_on_goal"] == "yes":
                goal = ("goal %s" % row["goal_event"]) if row["goal_event"] else "its goal (goal_event not recorded)"
                add(findings, "PASS", "JC04", "exits on %s (configuration-only; confirm with seed S02 in Full depth)."
                    % goal, row)
            else:
                add(findings, "FAIL", "JC04",
                    "has %s message steps and up to %s hours of delay but no exit on its goal (exit_on_goal: %s)."
                    " Read the exit settings and look for sends after the goal event in the send log."
                    % (fmt_num(steps), fmt_num(delay), row["exit_on_goal"] or "blank"), row)
        # JC05 send-time recheck
        if delay > 1 and row["recheck_after_delay"] != "yes":
            add(findings, "WARN", "JC05",
                "waits up to %s hours before a message with no send-time recheck. Read the copy: FAIL if it"
                " states the unfinished state or offers a discount." % fmt_num(delay), row)
        # JC06 re-entry
        if row["re_entry"] in ("", "unknown"):
            add(findings, "WARN", "JC06", "re-entry is %s; record once, per_unit, cooldown_<n>d or always to match"
                " the journey unit (%s)." % (row["re_entry"] or "blank", unit or "unit not recorded"), row)
        elif row["re_entry"] == "once" and unit in PER_OBJECT_UNITS:
            add(findings, "WARN", "JC06", "re-entry is once but the unit is %s: a second %s for the same person"
                " cannot enter." % (unit, unit), row)
        # TD01 authoritative source
        jtype = row["journey_type"]
        if row["trigger_source"] in CLIENT_SOURCES and (jtype in COMMERCIAL or jtype in ACCOUNT_STATE):
            kind = "a commercial fact" if jtype in COMMERCIAL else "an account state"
            add(findings, "WARN", "TD01",
                "is triggered client-side (trigger_source: %s) on %s (%s). Trigger it from the system that"
                " commits the fact; FAIL if failed attempts enter or JC02 is above 1.10."
                % (row["trigger_source"], kind, row["trigger_name"] or "trigger not recorded"), row)
        # FC02 cap coverage
        if row["message_class"] == "marketing":
            capped = row["frequency_capped"]
            if capped == "no":
                if args.policy_caps_all_optional == "yes":
                    add(findings, "FAIL", "FC02", "is a live marketing journey with frequency_capped: no while the"
                        " policy caps all optional messages.", row)
                else:
                    add(findings, "WARN", "FC02", "is a live marketing journey not covered by the sender's cap;"
                        " confirm whether the policy counts it.", row)
            elif capped == "exempt":
                if row["notes"]:
                    add(findings, "INFO", "FC02", "is exempt from the cap: %s" % row["notes"], row)
                else:
                    add(findings, "WARN", "FC02", "is exempt from the cap with no reason in notes.", row)
            elif capped in ("", "na"):
                add(findings, "WARN", "FC02", "has frequency_capped %s on a marketing journey; record yes, no or"
                    " exempt." % (capped or "blank"), row)

    # PC07 promotional content outside marketing controls (live and external rows)
    for row in active:
        non_marketing = row["message_class"] in ("transactional", "service")
        to_unsubscribed = row["sends_to_unsubscribed"] == "yes"
        if row["journey_type"] in PROMOTIONAL and (row["message_class"] == "transactional" or to_unsubscribed
                                                   or non_marketing):
            reasons = []
            if non_marketing:
                reasons.append("classed %s" % row["message_class"])
            if to_unsubscribed:
                reasons.append("set to send to unsubscribed users")
            add(findings, "FAIL", "PC07", "is a promotional type (%s) %s, so it bypasses preferences and caps."
                " Reclass it as marketing or remove the promotion." % (row["journey_type"], " and ".join(reasons)),
                row)
        elif row["journey_type"] in READ_THE_COPY and (non_marketing or to_unsubscribed):
            add(findings, "WARN", "PC07", "is a %s alert (%s)%s. Read the copy: FAIL when it carries an upgrade"
                " offer or a discount; a plain notice about the account's standing passes." % (
                    row["message_class"], row["journey_type"],
                    " set to reach unsubscribed users" if to_unsubscribed else ""), row)
        elif to_unsubscribed or row["message_class"] == "transactional":
            add(findings, "PASS", "PC07", "is a %s message (%s)%s; confirm the copy carries no promotion." % (
                row["message_class"], row["journey_type"],
                " set to reach unsubscribed users" if to_unsubscribed else ""), row)

    # JC07 collisions
    reported = []
    for jtype, group in sorted(by_type.items()):
        members = [r for r in group if r["status"] in ("live", "external")]
        if jtype != "other" and len(members) >= 2:
            ids = {r["journey_id"] for r in members}
            reported.append(ids)
            add(findings, "WARN", "JC07",
                "%d live or external journeys share type %s: %s. Add a mutual exclusion or priority rule, or retire"
                " one; confirm with message_pressure.py journey pairs (FAIL if both reach the same people within"
                " 24 hours)." % (len(members), jtype, "; ".join(label(r) for r in members)))
    by_trigger = defaultdict(list)
    for row in live:
        if row["message_class"] == "marketing" and row["trigger_name"]:
            by_trigger[row["trigger_name"].lower()].append(row)
    for trigger, group in sorted(by_trigger.items()):
        ids = {r["journey_id"] for r in group}
        if len(group) >= 2 and not any(ids <= seen for seen in reported):
            add(findings, "WARN", "JC07",
                "%d live marketing journeys start on trigger %s: %s. Add a mutual exclusion or priority rule;"
                " confirm with message_pressure.py journey pairs." % (len(group), trigger,
                                                                      "; ".join(label(r) for r in group)))

    # JC08 ownership (every row, external rows included)
    for row in rows:
        if row["status"] in ("live", "external", "paused") and not row["owner"]:
            add(findings, "WARN", "JC08", "has no named owner. Name the person or role who can pause it and its"
                " outcome metric.", row)

    # MH02 holdout coverage
    uncovered = [r for r in live if r["message_class"] == "marketing" and r["holdout"] == "none"]
    if args.global_holdout == "yes":
        for row in uncovered:
            add(findings, "WARN", "MH02", "is a live marketing journey that does not exclude the global holdout"
                " (holdout: none). Exclude holdout members in this sender or record why not.", row)
    elif uncovered:
        add(findings, "WARN", "MH02", "no global holdout, and %d live marketing journeys have no holdout of their"
            " own: %s." % (len(uncovered), ", ".join(r["journey_id"] for r in uncovered)))

    findings.sort(key=lambda f: (LEVEL_ORDER[f["level"]], RULE_ORDER[f["rule"]]))
    return findings, coverage, ratios, stage_view(rows, model, not_applicable)


def stage_view(rows, model, not_applicable):
    """Live and external journeys per lifecycle stage, then the model's missing types."""
    stages = defaultdict(lambda: {"journeys": [], "missing_required": [], "missing_recommended": [], "na": []})
    covered = set()
    for row in rows:
        if row["status"] not in ("live", "external"):
            continue
        covered.add(row["journey_type"])
        stage = row["stage"] or DEFAULT_STAGE.get(row["journey_type"], "unstaged")
        stages[stage]["journeys"].append("%s (%s, %s)" % (row["journey_id"], row["platform"], row["status"]))
    for kind, types in (("required", list(model["required"])), ("recommended", model["recommended"])):
        for jtype in types:
            stage = DEFAULT_STAGE.get(jtype, "unstaged")
            if jtype in not_applicable:
                stages[stage]["na"].append(jtype)
            elif jtype not in covered:
                stages[stage]["missing_" + kind].append(jtype)
    order = STAGE_ORDER + sorted(s for s in stages if s not in STAGE_ORDER and s != "unstaged") + ["unstaged"]
    view = []
    for stage in order:
        if stage not in stages:
            continue
        entry = stages[stage]
        view.append({"stage": stage, "journeys": entry["journeys"], "missing_required": entry["missing_required"],
                     "missing_recommended": entry["missing_recommended"], "not_applicable": entry["na"]})
    return view


def stage_line(entry):
    text = "%s: %s." % (entry["stage"], "; ".join(entry["journeys"]) or "no live or external journey")
    if entry["missing_required"]:
        text += " Missing required: %s." % ", ".join(entry["missing_required"])
    if entry["missing_recommended"]:
        text += " Missing recommended: %s." % ", ".join(entry["missing_recommended"])
    if entry["not_applicable"]:
        text += " N/A: %s." % ", ".join(entry["not_applicable"])
    return text


def line(f):
    subject = ('journey "%s" [%s] ' % (f["journey_name"], f["journey_id"])) if f["journey_id"] else ""
    return "%s %s %s: %s%s" % (f["level"], f["rule"], f["name"], subject, f["detail"])


def main():
    parser = argparse.ArgumentParser(description="Rate a lifecycle journey inventory CSV.")
    parser.add_argument("inventory")
    parser.add_argument("--model", required=True, choices=sorted(MODELS))
    parser.add_argument("--not-applicable", default="", help="comma-separated journey types that do not apply")
    parser.add_argument("--global-holdout", choices=["yes", "no"], default="no")
    parser.add_argument("--policy-caps-all-optional", choices=["yes", "no"], default="no")
    parser.add_argument("--as-of", help="YYYY-MM-DD, default today")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()
    try:
        args.as_of = datetime.strptime(args.as_of, "%Y-%m-%d").date() if args.as_of else date.today()
    except ValueError:
        bad_input("--as-of must be YYYY-MM-DD")
    model_types = set(MODELS[args.model]["required"]) | set(MODELS[args.model]["recommended"])
    unknown = sorted({t.strip().lower() for t in args.not_applicable.split(",") if t.strip()} - model_types)
    if unknown:
        bad_input("--not-applicable names types that are not in the %s model: %s. Valid types: %s" % (
            args.model, ", ".join(unknown), ", ".join(sorted(model_types))))

    rows = read_rows(args.inventory)
    normalise(rows)
    findings, coverage, ratios, stages = audit(rows, args)
    counts = {level: sum(1 for f in findings if f["level"] == level) for level in LEVEL_ORDER}
    statuses = defaultdict(int)
    for row in rows:
        statuses[row["status"]] += 1
    summary = "%d FAIL, %d WARN, %d PASS, %d INFO. Nothing was changed." % (
        counts["FAIL"], counts["WARN"], counts["PASS"], counts["INFO"])

    if args.json:
        print(json.dumps({
            "file": args.inventory, "model": args.model, "as_of": args.as_of.isoformat(),
            "rows": len(rows), "statuses": dict(statuses), "findings": findings,
            "coverage_map": coverage, "coverage_by_stage": stages, "coverage_ratios": ratios, "counts": counts,
            "summary": summary, "not_checked": NOT_CHECKED,
        }, indent=2))
    else:
        print("Lifecycle inventory audit: %s" % args.inventory)
        print("Model: %s. As of: %s. Rows: %d (%s)." % (
            args.model, args.as_of.isoformat(), len(rows),
            ", ".join("%d %s" % (n, s) for s, n in sorted(statuses.items()))))
        print("Global holdout: %s. Policy caps all optional messages: %s." % (
            args.global_holdout, args.policy_caps_all_optional))
        print("")
        for f in findings:
            print(line(f))
        print("")
        print("Coverage map (%s):" % args.model)
        for entry in coverage:
            print("  " + entry)
        print("Coverage by stage (a row's stage column wins over the type's default):")
        for entry in stages:
            print("  " + stage_line(entry))
        print("Coverage ratios (live rows):")
        for entry in ratios:
            print("  " + entry)
        print("")
        print(summary)
        print("Not checked by this script: %s." % NOT_CHECKED)
    sys.exit(1 if counts["FAIL"] else 0)


if __name__ == "__main__":
    main()