#!/usr/bin/env python3
"""Apply GA4's default channel rules to source and medium rows and check them.

Usage:
  python3 scripts/classify_channels.py SESSIONS.csv
      [--registry REGISTRY.csv] [--warehouse-map MAP.csv]
      [--source-categories CATEGORIES.csv] [--json]

The GA4 BigQuery export has no channel group field, so this script predicts
the default channel group from session source, medium and campaign using the
manual-traffic rules on Google's default channel group page (checked
2026-11-01), evaluated in the page's order. Matching is case-insensitive.

Inputs:
  SESSIONS.csv     source, medium, campaign (optional), sessions,
                   key_events (optional), click_id_sessions (optional),
                   as produced by query Q15 or a GA4 export
  REGISTRY.csv     source, medium, expected_channel, note; "*" as source
                   approves a medium for any source (exact, case-sensitive).
                   allowed-pairs.csv from the utm-convention-and-registry
                   skill also works (utm_source, utm_medium,
                   expected_custom_channel, expected_default_channel).
                   GA4's automatic pairs, (direct) / (none) and any source
                   with medium organic, referral or ai-assistant, never count
                   as unapproved, listed or not.
  MAP.csv          source, medium, campaign (optional), warehouse_channel:
                   the channel the warehouse's own CASE logic assigns
  CATEGORIES.csv   Google's downloadable list of sources and categories
                   (a source column and a category column such as
                   SOURCE_CATEGORY_SEARCH). Without it a small built-in seed
                   list is used and rows that depend on it are flagged.

Checks: CHN-01 vocabulary, CHN-02 Unassigned share, CHN-03 click-ID sessions
in paid channels, CHN-04 case and format variants, CHN-05 unresolved macros,
CHN-07 agreement with the warehouse logic. Suggested statuses use the
default thresholds in references/checks-channels-and-controls.md.

What it cannot see: Google Ads auto-tagged traffic is classified by GA4 from
the ad network type, which is not in the export, so google / cpc rows are
predicted with the manual rules and flagged. It cannot see spend (CHN-05 uses
a paid channel as the proxy for a campaign with spend; CHN-06 needs query
Q18). Standard library only. Exit status: 0 when no suggested status is
FAIL, 1 otherwise, 2 on input errors, including a file that is not UTF-8 CSV,
a row with more cells than the header, and a file with no sessions.
"""
import argparse
import csv
import json
import re
import sys
from collections import OrderedDict
from decimal import ROUND_HALF_UP, Decimal

PAID_MEDIUM = re.compile(r"^(.*cp.*|ppc|retargeting|paid.*)$", re.I)
SHOPPING_CAMPAIGN = re.compile(r"^(.*(([^a-df-z]|^)shop|shopping).*)$", re.I)
DISPLAY_MEDIA = {"display", "banner", "expandable", "interstitial", "cpm"}
SOCIAL_MEDIA = {"social", "social-network", "social-media", "sm", "social network", "social media"}
EMAIL_VALUES = {"email", "e-mail", "e_mail", "e mail"}
REFERRAL_MEDIA = {"referral", "app", "link"}
PAID_CHANNELS = {"Paid Search", "Paid Social", "Paid Video", "Paid Shopping", "Paid Other",
                 "Cross-network", "Display"}
ORGANIC_OR_DIRECT = {"Organic Search", "Organic Social", "Organic Video", "Organic Shopping", "Direct"}
DEFAULT_CHANNELS = PAID_CHANNELS | ORGANIC_OR_DIRECT | {
    "AI Assistant", "Referral", "Email", "Affiliates", "Audio", "SMS", "Mobile Push Notifications", "Unassigned"}
# Written by GA4, not by a tagged link, so a UTM registry has no reason to list them.
AUTOMATIC_MEDIA = {"organic", "referral", "ai-assistant"}
# Channels reached only after the category rules, so a source category could change them.
AFTER_CATEGORY_RULES = {"Display", "Paid Other", "AI Assistant", "Referral", "Email", "Affiliates",
                        "Audio", "SMS", "Mobile Push Notifications", "Unassigned"}
NOT_A_SITE = {"(direct)", "(not set)", "", "sms", "firebase"} | EMAIL_VALUES
OBJECTIVE_MEDIA = {"leadgen", "lead_gen", "lead-gen", "awareness", "boost", "boosted", "reach", "engagement"}
AI_LIKE_MEDIA = {"ai", "chatgpt", "llm", "genai", "assistant", "ai_assistant", "ai assistant", "ai-chat"}
# Unresolved template tokens. Single square brackets are left out on purpose:
# names such as "[Brand] Search - US" are a common naming convention, not a macro.
MACRO_PATTERNS = [
    re.compile(r"\{|\}"),
    re.compile(r"%7[bd]", re.I),
    re.compile(r"\[\[[^\]]+\]\]"),
    re.compile(r"\*\|[^|]+\|\*"),
    re.compile(r"%%[^%\s]+%%"),
    re.compile(r"__[A-Z][A-Z0-9_]*__"),
    re.compile(r"(?<![A-Za-z0-9])(CAMPAIGN_ID|CAMPAIGN_NAME|ADGROUP_ID|ADSET_ID|ADSET_NAME|AD_ID|AD_NAME|CREATIVE_ID)(?![A-Za-z0-9])"),
    re.compile(r"(?<![A-Za-z0-9])(campaign\.name|adset\.name|ad\.name|site_source_name)(?![A-Za-z0-9])", re.I),
]

# Small seed of source categories, used only when --source-categories is not
# given. Re-verify against Google's downloadable list on the default channel
# group page; the full list has many more sources and domain variants.
SEED_CATEGORIES = {
    "search": ["google", "google.com", "bing", "bing.com", "yahoo", "search.yahoo.com", "duckduckgo",
               "duckduckgo.com", "baidu", "baidu.com", "yandex", "yandex.ru", "ecosia", "ecosia.org",
               "naver", "naver.com"],
    "social": ["facebook", "facebook.com", "m.facebook.com", "l.facebook.com", "lm.facebook.com",
               "instagram", "instagram.com", "linkedin", "linkedin.com", "lnkd.in", "twitter",
               "twitter.com", "t.co", "x.com", "reddit", "reddit.com", "pinterest", "pinterest.com",
               "tiktok", "tiktok.com", "snapchat", "quora", "quora.com", "threads.net"],
    "video": ["youtube", "youtube.com", "m.youtube.com", "vimeo", "vimeo.com", "twitch", "twitch.tv",
              "dailymotion", "dailymotion.com"],
    "shopping": ["amazon", "amazon.com", "ebay", "ebay.com", "etsy", "etsy.com", "walmart",
                 "walmart.com", "shopping.google.com"],
}


class InputError(Exception):
    pass


def pct(part, whole):
    if not whole:
        return Decimal(0)
    return (Decimal(100) * Decimal(part) / Decimal(whole)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)


def read_csv(path, required):
    rows = []
    try:
        with open(path, newline="", encoding="utf-8-sig") as handle:
            reader = csv.DictReader(handle)
            header = [h.strip() for h in (reader.fieldnames or [])]
            for r in reader:
                if None in r:
                    raise InputError("%s line %d has more cells than the header; quote any value that holds a comma" % (
                        path, reader.line_num))
                rows.append({(k or "").strip(): (v or "").strip() for k, v in r.items()})
    except OSError as exc:
        raise InputError("cannot read %s: %s" % (path, exc))
    except (UnicodeDecodeError, csv.Error):
        raise InputError("%s is not a UTF-8 CSV file; save a workbook or UTF-16 export as CSV UTF-8" % path)
    missing = [c for c in required if c not in header]
    if missing:
        raise InputError("%s lacks columns: %s" % (path, ", ".join(missing)))
    return header, rows


def to_int(text, where):
    if text == "":
        return 0
    try:
        value = Decimal(text.replace(",", ""))
    except Exception:
        raise InputError("%s: %r is not a number" % (where, text))
    if value != value.to_integral_value() or value < 0:
        raise InputError("%s: %r is not a whole number of 0 or more" % (where, text))
    return int(value)


def load_categories(path):
    header, rows = read_csv(path, [])
    lower = dict((h.lower(), h) for h in header)
    src = lower.get("source")
    cat = next((lower[h] for h in lower if "category" in h), None)
    if not src or not cat:
        raise InputError("%s needs a source column and a category column" % path)
    cats = {}
    for r in rows:
        value = r[cat].upper()
        for name in ("search", "social", "video", "shopping"):
            if name.upper() in value:
                cats[r[src].lower()] = name
    if not cats:
        raise InputError("%s has no search, social, video or shopping categories" % path)
    return cats


def load_registry(path):
    """Approved pairs: {(source, medium): channel, or "" when the file names no channel for the pair}."""
    header, rows = read_csv(path, [])
    src = "source" if "source" in header else "utm_source"
    med = "medium" if "medium" in header else "utm_medium"
    if src not in header or med not in header:
        raise InputError("%s needs source and medium columns (or utm_source and utm_medium)" % path)
    registry = {}
    for r in rows:
        channel = r.get("expected_channel") or r.get("expected_custom_channel", "")
        default = r.get("expected_default_channel", "")
        if not channel and default in DEFAULT_CHANNELS:
            channel = default
        registry[(r[src], r[med])] = channel
    return registry


def seed_categories():
    cats = {}
    for name, sources in SEED_CATEGORIES.items():
        for s in sources:
            cats[s] = name
    return cats


def predict(source, medium, campaign, cats):
    s, m, c = source.lower(), medium.lower(), campaign.lower()
    cat = cats.get(s)
    paid = bool(PAID_MEDIUM.match(m))
    shopping_campaign = bool(SHOPPING_CAMPAIGN.match(c)) if c else False
    if s == "(direct)" and m in ("(not set)", "(none)"):
        return "Direct"
    if "cross-network" in c:
        return "Cross-network"
    if (cat == "shopping" or shopping_campaign) and paid:
        return "Paid Shopping"
    if cat == "search" and paid:
        return "Paid Search"
    if cat == "social" and paid:
        return "Paid Social"
    if cat == "video" and paid:
        return "Paid Video"
    if m in DISPLAY_MEDIA:
        return "Display"
    if paid:
        return "Paid Other"
    if cat == "shopping" or shopping_campaign:
        return "Organic Shopping"
    if cat == "social" or m in SOCIAL_MEDIA:
        return "Organic Social"
    if cat == "video" or "video" in m:
        return "Organic Video"
    if cat == "search" or m == "organic":
        return "Organic Search"
    if m == "ai-assistant":
        return "AI Assistant"
    if m in REFERRAL_MEDIA:
        return "Referral"
    if s in EMAIL_VALUES or m in EMAIL_VALUES:
        return "Email"
    if m == "affiliate":
        return "Affiliates"
    if m == "audio":
        return "Audio"
    if s == "sms" or m == "sms":
        return "SMS"
    if m.endswith("push") or "mobile" in m or "notification" in m or s == "firebase":
        return "Mobile Push Notifications"
    return "Unassigned"


def normalise(text):
    return re.sub(r"_+", "_", re.sub(r"[-\s]", "_", text.strip().lower()))


def is_not_set(source, medium):
    return source in ("(not set)", "") or medium in ("(not set)", "")


def is_automatic(source, medium):
    return (source == "(direct)" and medium == "(none)") or medium in AUTOMATIC_MEDIA


def has_macro(*values):
    return any(p.search(v) for v in values for p in MACRO_PATTERNS)


def label(row):
    text = "%s / %s" % (row["source"], row["medium"])
    return text + (" / %s" % row["campaign"] if row["campaign"] else "")


def analyse(args):
    _, rows = read_csv(args.sessions, ["source", "medium", "sessions"])
    if args.source_categories:
        cats, cat_origin = load_categories(args.source_categories), "Google's list (%s)" % args.source_categories
    else:
        cats, cat_origin = seed_categories(), "built-in seed list"
    registry = load_registry(args.registry) if args.registry else None
    wmap = None
    if args.warehouse_map:
        _, map_rows = read_csv(args.warehouse_map, ["source", "medium", "warehouse_channel"])
        wmap = {}
        for r in map_rows:
            wmap[(r["source"], r["medium"], r.get("campaign", ""))] = r["warehouse_channel"]

    data = []
    for number, r in enumerate(rows, start=2):
        where = "%s line %d" % (args.sessions, number)
        row = {"source": r["source"], "medium": r["medium"], "campaign": r.get("campaign", ""),
               "sessions": to_int(r["sessions"], where + " sessions"),
               "key_events": to_int(r.get("key_events", ""), where + " key_events"),
               "click_id_sessions": to_int(r.get("click_id_sessions", ""), where + " click_id_sessions")}
        row["channel"] = predict(row["source"], row["medium"], row["campaign"], cats)
        flags = []
        if row["source"].lower() == "google" and row["medium"].lower() == "cpc":
            flags.append("auto-tagged: verify network type")
        if has_macro(row["source"], row["medium"], row["campaign"]):
            flags.append("macro")
            row["macro"] = True
        if row["channel"] == "Unassigned":
            flags.append("unassigned")
        if row["click_id_sessions"] and row["channel"] not in PAID_CHANNELS:
            flags.append("click-ID sessions outside paid (%d)" % row["click_id_sessions"])
        if normalise(row["medium"]) in set(normalise(x) for x in OBJECTIVE_MEDIA):
            flags.append("objective in medium")
        if row["medium"].lower() in AI_LIKE_MEDIA:
            flags.append("AI-like medium misses the AI Assistant rule (medium ai-assistant)")
        if (not args.source_categories and row["source"].lower() not in cats
                and row["source"].lower() not in NOT_A_SITE and row["channel"] in AFTER_CATEGORY_RULES):
            flags.append("category unknown, load Google's list")
            row["category_unknown"] = True
        if registry is not None:
            exp = registry.get((row["source"], row["medium"]))
            if exp is None:
                exp = registry.get(("*", row["medium"]))
            row["registry_channel"] = exp or None
            if exp is None and not is_not_set(row["source"], row["medium"]) and not is_automatic(row["source"], row["medium"]):
                flags.append("unapproved")
                row["unapproved"] = True
        if wmap is not None:
            wh = wmap.get((row["source"], row["medium"], row["campaign"]))
            if wh is None:
                wh = wmap.get((row["source"], row["medium"], ""))
            row["warehouse_channel"] = wh
        row["flags"] = flags
        data.append(row)
    return data, cats, cat_origin, registry, wmap


def summarise(args, data, registry, wmap):
    total = sum(r["sessions"] for r in data)
    total_ke = sum(r["key_events"] for r in data)
    s = OrderedDict([("sessions", total), ("key_events", total_ke)])
    statuses = OrderedDict()

    # CHN-01 vocabulary.
    if registry is not None:
        unapproved = [r for r in data if r.get("unapproved")]
        pairs = OrderedDict()
        for r in unapproved:
            key = "%s / %s" % (r["source"], r["medium"])
            pairs[key] = pairs.get(key, 0) + r["sessions"]
        u_sessions = sum(pairs.values())
        big = [(k, v) for k, v in sorted(pairs.items(), key=lambda kv: -kv[1]) if pct(v, total) >= Decimal("0.5")]
        share = pct(u_sessions, total)
        if share > 5:
            st = "FAIL"
        elif share > 2 or big:
            st = "WARN"
        else:
            st = "PASS"
        s["chn01"] = OrderedDict([("unapproved_sessions", u_sessions), ("share", share),
                                  ("pairs_at_half_percent", big), ("pairs", pairs)])
    else:
        breaches = [r for r in data if not is_not_set(r["source"], r["medium"]) and
                    (r["source"] != r["source"].lower() or r["medium"] != r["medium"].lower()
                     or " " in r["source"] + r["medium"])]
        s["chn01"] = OrderedDict([("note", "no registry: checked the lowercase rule only"),
                                  ("rule_breach_sessions", sum(r["sessions"] for r in breaches))])
        st = "WARN"
    statuses["CHN-01"] = st

    # CHN-02 Unassigned share, overall and per paid campaign.
    un = [r for r in data if r["channel"] == "Unassigned"]
    un_s, un_k = sum(r["sessions"] for r in un), sum(r["key_events"] for r in un)
    paid_campaigns = OrderedDict()
    for r in data:
        if r["campaign"] and not r["campaign"].startswith("("):
            c = paid_campaigns.setdefault(r["campaign"], {"paid": False, "sessions": 0, "unassigned": 0})
            c["sessions"] += r["sessions"]
            c["paid"] = c["paid"] or r["channel"] in PAID_CHANNELS
            if r["channel"] == "Unassigned":
                c["unassigned"] += r["sessions"]
    bad_campaigns = [(k, pct(v["unassigned"], v["sessions"])) for k, v in paid_campaigns.items()
                     if v["paid"] and pct(v["unassigned"], v["sessions"]) > 5]
    share_s, share_k = pct(un_s, total), pct(un_k, total_ke)
    if share_s > 5 or share_k > 5 or bad_campaigns:
        st = "FAIL"
    elif share_s > 1 or share_k > 1:
        st = "WARN"
    else:
        st = "PASS"
    statuses["CHN-02"] = st
    s["chn02"] = OrderedDict([("unassigned_sessions", un_s), ("session_share", share_s),
                              ("unassigned_key_events", un_k), ("key_event_share", share_k),
                              ("paid_campaigns_over_5_percent", bad_campaigns)])

    # CHN-03 click-ID sessions in paid channels.
    cid_total = sum(r["click_id_sessions"] for r in data)
    cid_paid = sum(r["click_id_sessions"] for r in data if r["channel"] in PAID_CHANNELS)
    outside = [(label(r), r["channel"], r["click_id_sessions"]) for r in data
               if r["click_id_sessions"] and r["channel"] not in PAID_CHANNELS]
    if cid_total == 0:
        statuses["CHN-03"] = "NOT CHECKED"
        share = None
    else:
        share = pct(cid_paid, cid_total)
        statuses["CHN-03"] = "PASS" if share >= 98 else ("WARN" if share >= 95 else "FAIL")
    s["chn03"] = OrderedDict([("click_id_sessions", cid_total), ("in_paid", cid_paid),
                              ("share", share), ("outside_paid", outside)])

    # CHN-04 variants.
    groups = OrderedDict()
    for r in data:
        if is_not_set(r["source"], r["medium"]):
            continue
        g = groups.setdefault((normalise(r["source"]), normalise(r["medium"])), OrderedDict())
        pair = (r["source"], r["medium"])
        g[pair] = g.get(pair, 0) + r["sessions"]
    variants = []
    for gkey, spellings in groups.items():
        if len(spellings) < 2:
            continue
        canonical = None
        if registry is not None:
            canonical = next((p for p in spellings if p in registry), None)
        if canonical is None:
            canonical = max(spellings.items(), key=lambda kv: kv[1])[0]
        other = sum(v for p, v in spellings.items() if p != canonical)
        variants.append(OrderedDict([
            ("canonical", "%s / %s" % canonical),
            ("spellings", ["%s / %s" % p for p in spellings]),
            ("non_canonical_sessions", other), ("share", pct(other, total))]))
    statuses["CHN-04"] = "WARN" if any(v["share"] >= Decimal("0.1") for v in variants) else "PASS"
    s["chn04"] = variants

    # CHN-05 macros.
    macro = [r for r in data if r.get("macro")]
    m_s = sum(r["sessions"] for r in macro)
    m_share = pct(m_s, total)
    on_paid = [r for r in macro if r["channel"] in PAID_CHANNELS]
    if not macro:
        st = "PASS"
    elif m_share >= Decimal("0.1") or on_paid:
        st = "FAIL"
    else:
        st = "WARN"
    statuses["CHN-05"] = st
    s["chn05"] = OrderedDict([("sessions", m_s), ("share", m_share),
                              ("rows", [(label(r), r["channel"], r["sessions"]) for r in macro])])

    # CHN-07 warehouse logic.
    if wmap is not None:
        mapped = [r for r in data if r.get("warehouse_channel")]
        unmapped = [r for r in data if not r.get("warehouse_channel")]
        agree, disagreements, paid_to_organic, unknown_to_direct = 0, [], [], []
        for r in mapped:
            expected = r.get("registry_channel") or r["channel"]
            wh = r["warehouse_channel"]
            if wh == expected:
                agree += r["sessions"]
                continue
            disagreements.append((label(r), expected, wh, r["sessions"]))
            if expected in PAID_CHANNELS and wh in ORGANIC_OR_DIRECT:
                paid_to_organic.append("%s as %s" % (label(r), wh))
            if wh == "Direct" and (is_not_set(r["source"], r["medium"]) or expected == "Unassigned"):
                unknown_to_direct.append("%s as %s" % (label(r), wh))
        mapped_total = sum(r["sessions"] for r in mapped)
        share = pct(agree, mapped_total)
        if share < 95 or paid_to_organic or unknown_to_direct:
            st = "FAIL"
        elif share < 98:
            st = "WARN"
        else:
            st = "PASS"
        statuses["CHN-07"] = st
        s["chn07"] = OrderedDict([
            ("agree_sessions", agree), ("mapped_sessions", mapped_total), ("share", share),
            ("disagreements", disagreements), ("paid_mapped_to_organic_or_direct", paid_to_organic),
            ("unknown_mapped_to_direct", unknown_to_direct),
            ("unmapped_rows", [label(r) for r in unmapped])])
    return s, statuses


def num(n):
    return format(n, ",")


def print_report(args, data, cat_origin, s, statuses):
    out = ["Channel classification check: %s" % args.sessions,
           "Rules: GA4 default channel group, manual-traffic rules (checked 2026-11-01). Source categories: %s." % cat_origin,
           "Rows %d, sessions %s, key events %s, click-ID sessions %s." % (
               len(data), num(s["sessions"]), num(s["key_events"]), num(s["chn03"]["click_id_sessions"])),
           "", "Rows (pair: sessions, predicted channel, flags)"]
    for r in data:
        extra = "; " + "; ".join(r["flags"]) if r["flags"] else ""
        wh = (", warehouse %s" % r["warehouse_channel"]) if r.get("warehouse_channel") else ""
        out.append("- %s: %s, %s%s%s" % (label(r), num(r["sessions"]), r["channel"], wh, extra))
    shares = OrderedDict()
    for r in data:
        t = shares.setdefault(r["channel"], [0, 0])
        t[0] += r["sessions"]
        t[1] += r["key_events"]
    out += ["", "Shares by predicted channel"]
    for ch, (ses, ke) in sorted(shares.items(), key=lambda kv: -kv[1][0]):
        out.append("- %s: %s sessions (%s%%), %s key events (%s%%)" % (
            ch, num(ses), pct(ses, s["sessions"]), num(ke), pct(ke, s["key_events"])))
    unknown = [r for r in data if r.get("category_unknown")]
    if unknown:
        out.append("- %d rows (%s sessions) have a source outside the seed list; their channel can change "
                   "with Google's list (--source-categories)." % (len(unknown), num(sum(r["sessions"] for r in unknown))))
    out += ["", "Checks"]
    c1 = s["chn01"]
    if "share" in c1:
        big = ", ".join("%s %s (%s%%)" % (k, num(v), pct(v, s["sessions"])) for k, v in c1["pairs_at_half_percent"])
        out.append("- CHN-01 vocabulary: unapproved pairs %s sessions (%s%%)%s. Suggested: %s" % (
            num(c1["unapproved_sessions"]), c1["share"],
            ("; pairs at 0.5%% or more: %s" % big) if big else "", statuses["CHN-01"]))
    else:
        out.append("- CHN-01 vocabulary: %s; %s sessions break it. Suggested: %s" % (
            c1["note"], num(c1["rule_breach_sessions"]), statuses["CHN-01"]))
    c2 = s["chn02"]
    camp = "; paid campaigns over 5%% Unassigned: %s" % ", ".join(
        "%s (%s%%)" % x for x in c2["paid_campaigns_over_5_percent"]) if c2["paid_campaigns_over_5_percent"] else ""
    key_events = ("%s key events (%s%%)" % (num(c2["unassigned_key_events"]), c2["key_event_share"])
                  if s["key_events"] else "key events not in the input, so sessions only")
    out.append("- CHN-02 Unassigned: %s sessions (%s%%), %s%s. Suggested: %s" % (
        num(c2["unassigned_sessions"]), c2["session_share"], key_events, camp, statuses["CHN-02"]))
    c3 = s["chn03"]
    if c3["share"] is None:
        out.append("- CHN-03 click IDs: no click_id_sessions in the input. Suggested: NOT CHECKED")
    else:
        outside = "; outside paid: %s" % ", ".join("%s (%s) %s" % (a, b, num(c)) for a, b, c in c3["outside_paid"]) \
            if c3["outside_paid"] else ""
        out.append("- CHN-03 click IDs: %s click-ID sessions, %s in paid channels (%s%%)%s. Suggested: %s" % (
            num(c3["click_id_sessions"]), num(c3["in_paid"]), c3["share"], outside, statuses["CHN-03"]))
    if s["chn04"]:
        for v in s["chn04"]:
            out.append("- CHN-04 variants: %s spelled %d ways (%s); %s sessions in non-canonical spellings (%s%%). Suggested: %s" % (
                v["canonical"], len(v["spellings"]), ", ".join(v["spellings"]), num(v["non_canonical_sessions"]),
                v["share"], statuses["CHN-04"]))
    else:
        out.append("- CHN-04 variants: none. Suggested: %s" % statuses["CHN-04"])
    c5 = s["chn05"]
    rows5 = ", ".join("%s (%s)" % (a, b) for a, b, _ in c5["rows"])
    out.append("- CHN-05 macros: %s sessions (%s%%)%s. Suggested: %s" % (
        num(c5["sessions"]), c5["share"], (": " + rows5) if rows5 else "", statuses["CHN-05"]))
    if "chn07" in s:
        c7 = s["chn07"]
        dis = "; ".join("%s expected %s, warehouse %s (%s)" % (a, b, c, num(d)) for a, b, c, d in c7["disagreements"])
        out.append("- CHN-07 warehouse logic: agreement %s of %s sessions (%s%%)%s%s%s%s. Suggested: %s" % (
            num(c7["agree_sessions"]), num(c7["mapped_sessions"]), c7["share"],
            ("; disagreements: " + dis) if dis else "",
            ("; paid mapped to organic or direct: " + ", ".join(c7["paid_mapped_to_organic_or_direct"]))
            if c7["paid_mapped_to_organic_or_direct"] else "",
            ("; unknown mapped to Direct: " + ", ".join(c7["unknown_mapped_to_direct"]))
            if c7["unknown_mapped_to_direct"] else "",
            ("; rows without a warehouse mapping: " + ", ".join(c7["unmapped_rows"])) if c7["unmapped_rows"] else "",
            statuses["CHN-07"]))
    else:
        out.append("- CHN-07 warehouse logic: not checked (add --warehouse-map).")
    out += ["", "Suggested statuses: %s." % ", ".join("%s %s" % kv for kv in statuses.items()),
            "These come from the default thresholds. Record the final status in the findings register with this output as evidence.",
            "Nothing was changed."]
    print("\n".join(out))


def to_jsonable(value):
    if isinstance(value, Decimal):
        return str(value)
    if isinstance(value, tuple):
        return [to_jsonable(v) for v in value]
    if isinstance(value, dict):
        return OrderedDict((str(k), to_jsonable(v)) for k, v in value.items())
    if isinstance(value, list):
        return [to_jsonable(v) for v in value]
    return value


def main(argv=None):
    p = argparse.ArgumentParser(description="Predict GA4 default channels and check UTM quality.")
    p.add_argument("sessions", help="sessions by source, medium and campaign (Q15 output)")
    p.add_argument("--registry", help="approved source and medium pairs")
    p.add_argument("--warehouse-map", help="the warehouse's channel per source and medium")
    p.add_argument("--source-categories", help="Google's source category list")
    p.add_argument("--json", action="store_true")
    args = p.parse_args(argv)
    try:
        data, _, cat_origin, registry, wmap = analyse(args)
        if not data:
            raise InputError("%s has no rows" % args.sessions)
        if not sum(r["sessions"] for r in data):
            raise InputError("%s has no sessions, so there is nothing to classify" % args.sessions)
        s, statuses = summarise(args, data, registry, wmap)
    except InputError as exc:
        print("error: %s" % exc, file=sys.stderr)
        return 2
    if args.json:
        print(json.dumps(to_jsonable(OrderedDict([("rows", data), ("summary", s), ("suggested", statuses)])), indent=2))
    else:
        print_report(args, data, cat_origin, s, statuses)
    return 1 if "FAIL" in statuses.values() else 0


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