#!/usr/bin/env python3 """Dry-run a routing policy on recent signal episodes and check invariants I1 to I12. Usage: python3 route_dry_run.py --policy POLICY.json --records EPISODES.csv [--capacity ae=22,sdr=40] [--accept-validation valid] [--reply-window 30] [--out routed.csv] [--json] POLICY.json uses the format in references/routing-and-outbound.md (start from assets/routing-policy-template.json). EPISODES.csv has one row per signal delivery, repeat deliveries included, each with its episode_id and source_event_key, in the columns of assets/example-acme-episodes.csv. Owner columns use the same identifiers as the policy pools and fallback_owner (names or CRM owner IDs), or owner comparisons see different labels. Row numbers in the output are spreadsheet rows: the header is row 1. Values are checked before routing (exit 2 otherwise): signal_class is direct_request, first_party, inferred or public_change, and the boolean columns hold true or false. owner_available and covering_rep describe the account owner. Optional opportunity_owner_available and opportunity_covering_rep (both or neither) do the same for the opportunity owner: when present, rules with owner_from opportunity_owner and invariant I3 use them instead. What it does: - marks later rows that repeat a source_event_key as replays (_replay = true) - evaluates the rules in order; the first rule whose conditions all hold wins - resolves one owner per row: owner_from a column (cover_from instead when owner_available is false), round robin over a pool in row order, or the policy's fallback_owner for the fallback route - checks invariants I1 to I12 from references/rules.md on the result; the policy cannot switch them off. The fallback route counts as assigning an owner, so it cannot take an item from an owner who can act. - warns when one episode_id gets more than one task, nurture, research or sequence route from separate deliveries (business dedupe, W03) --capacity takes pool or owner names and the number of owned items allowed in the period the records cover (weekly capacity x weeks covered). Standard library only. Reads the two files, prints a report and writes only --out. Nothing is sent to a CRM, sequencer or any other system. Exit status: 0 no FAIL, 1 at least one FAIL, 2 bad input or policy. """ import argparse import csv import json import sys from collections import Counter, OrderedDict ROUTES = ( "replay", "no_action", "service", "cs_task", "opp_owner_task", "hold", "owner_task", "attach_existing", "pool_task", "sequence", "research", "nurture", "fallback", ) OPS = ("eq", "ne", "in", "not_in", "empty", "not_empty", "gt", "gte", "lt", "lte") TASK_ROUTES = {"owner_task", "pool_task", "opp_owner_task", "cs_task", "fallback"} HUMAN_ROUTES = TASK_ROUTES | {"service"} NEEDS_OWNER = TASK_ROUTES OWNER_CHECKED = {"owner_task", "pool_task", "sequence", "attach_existing"} PROSPECTING = {"sequence", "pool_task", "nurture", "research"} ACTION_ROUTES = TASK_ROUTES | PROSPECTING EXPIRED_BLOCKED = TASK_ROUTES | {"nurture"} UNDATED_BLOCKED = {"owner_task", "pool_task", "fallback", "nurture"} UNMATCHED_SAFE = {"no_action", "hold", "fallback", "service"} REQUIRED = ( "episode_id", "source_event_key", "signal_class", "identity_status", "suppressed", "customer", "open_opportunity", "opportunity_owner", "account_owner", "owner_available", "covering_rep", "active_sequence", "last_human_reply_days", "email_validation", "direct_request", "corroborated", "evidence_age_days", "expiry_days", "message_version_approved", ) SIGNAL_CLASSES = ("direct_request", "first_party", "inferred", "public_change") BOOLEANS = ( "suppressed", "customer", "open_opportunity", "owner_available", "active_sequence", "direct_request", "corroborated", "message_version_approved", ) OPP_COLUMNS = ("opportunity_owner_available", "opportunity_covering_rep") INVARIANTS = OrderedDict([ ("I1", "replay-and-episodes"), ("I2", "suppressed"), ("I3", "open-opportunity"), ("I4", "existing-owner"), ("I5", "customer"), ("I6", "coverage"), ("I7", "sequence-eligibility"), ("I8", "inferred-signal"), ("I9", "capacity"), ("I10", "direct-request"), ("I11", "identity"), ("I12", "evidence-age"), ]) SEVERITY_ORDER = {"FAIL": 0, "WARN": 1} class InputError(Exception): pass def text(value): return str(value).strip() if value is not None else "" def low(value): if isinstance(value, bool): return "true" if value else "false" return text(value).lower() def is_true(value): return low(value) == "true" def is_false(value): return low(value) == "false" def number(value): if isinstance(value, bool): return None try: return float(text(value)) except ValueError: return None def load_records(path): try: with open(path, newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) columns = [text(c) for c in (reader.fieldnames or [])] rows = [{text(k): text(v) for k, v in raw.items() if k is not None} for raw in reader] except (OSError, UnicodeDecodeError, csv.Error) as exc: raise InputError("cannot read %s as a UTF-8 CSV: %s" % (path, exc)) missing = [c for c in REQUIRED if c not in columns] if missing: raise InputError("%s is missing columns: %s" % (path, ", ".join(missing))) if not rows: raise InputError("%s has no rows" % path) present = [c for c in OPP_COLUMNS if c in columns] if len(present) == 1: raise InputError("%s has %s without %s; give both opportunity owner columns or neither" % (path, present[0], [c for c in OPP_COLUMNS if c not in present][0])) # A mislabelled class or boolean would slip past I2 to I12, so stop instead of guessing. for line, row in enumerate(rows, 2): if low(row.get("signal_class")) not in SIGNAL_CLASSES: raise InputError("row %d: signal_class %r is not one of %s" % (line, row.get("signal_class"), ", ".join(SIGNAL_CLASSES))) for column in BOOLEANS: if low(row.get(column)) not in ("true", "false"): raise InputError("row %d: %s must be true or false, got %r" % (line, column, row.get(column))) if present and low(row.get(OPP_COLUMNS[0])) not in ("true", "false", ""): raise InputError("row %d: %s must be true, false or blank, got %r" % (line, OPP_COLUMNS[0], row.get(OPP_COLUMNS[0]))) return columns, rows def availability(row, owner_column, cover_column): """Columns saying whether the owner in owner_column is away and who covers them.""" if owner_column == "opportunity_owner" and OPP_COLUMNS[0] in row: return OPP_COLUMNS return "owner_available", cover_column def check_condition(where, cond, known): if not isinstance(cond, dict) or not cond.get("field") or cond.get("op") not in OPS: raise InputError("%s: each condition needs field and op (%s)" % (where, ", ".join(OPS))) if cond["field"] not in known: raise InputError("%s: field %s is not a column in the records" % (where, cond["field"])) if cond["op"] in ("empty", "not_empty"): return if "value_field" in cond: if cond["value_field"] not in known: raise InputError("%s: value_field %s is not a column in the records" % (where, cond["value_field"])) elif "value" not in cond: raise InputError("%s: the condition on %s needs value or value_field" % (where, cond["field"])) def load_policy(path, columns): try: with open(path, encoding="utf-8-sig") as handle: policy = json.load(handle) except (OSError, ValueError) as exc: raise InputError("cannot read policy %s: %s" % (path, exc)) if not isinstance(policy, dict) or not isinstance(policy.get("rules"), list) or not policy["rules"]: raise InputError("the policy needs a non-empty rules list") pools = policy.get("pools") or {} if not isinstance(pools, dict): raise InputError("pools must map a pool name to a list of owners") for name, members in pools.items(): if not isinstance(members, list) or not members or not all(text(m) for m in members): raise InputError("pool %s needs a list of owner names" % name) known = set(columns) | {"_replay"} seen = set() for index, rule in enumerate(policy["rules"], 1): if not isinstance(rule, dict) or not text(rule.get("id")): raise InputError("rule %d needs an id" % index) where = "rule %s" % text(rule["id"]) if text(rule["id"]) in seen: raise InputError("duplicate %s" % where) seen.add(text(rule["id"])) if rule.get("route") not in ROUTES: raise InputError("%s: route must be one of %s" % (where, ", ".join(ROUTES))) for group in ("when", "when_any"): conditions = rule.get(group, []) if not isinstance(conditions, list): raise InputError("%s: %s must be a list" % (where, group)) for cond in conditions: check_condition(where, cond, known) for key in ("owner_from", "cover_from"): if rule.get(key) and rule[key] not in columns: raise InputError("%s: %s column %s is not in the records" % (where, key, rule[key])) if rule.get("pool") and rule["pool"] not in pools: raise InputError("%s: pool %s is not defined" % (where, rule["pool"])) if rule.get("method", "round_robin") != "round_robin": raise InputError("%s: only method round_robin can be dry-run" % where) return policy def holds(row, cond): left = text(row.get(cond["field"])) op = cond["op"] if op == "empty": return left == "" if op == "not_empty": return left != "" right = row.get(cond["value_field"]) if "value_field" in cond else cond.get("value") if op in ("in", "not_in"): options = right if isinstance(right, list) else [right] found = low(left) in {low(o) for o in options} return found if op == "in" else not found a, b = number(left), number(right) if op in ("gt", "gte", "lt", "lte"): if a is None or b is None: return False return {"gt": a > b, "gte": a >= b, "lt": a < b, "lte": a <= b}[op] same = (a == b) if (a is not None and b is not None) else (low(left) == low(right)) return same if op == "eq" else not same def matches(row, rule): if not all(holds(row, c) for c in rule.get("when", [])): return False any_of = rule.get("when_any", []) return not any_of or any(holds(row, c) for c in any_of) def route_rows(policy, rows): seen = set() counters = Counter() pools = policy.get("pools") or {} results = [] for index, row in enumerate(rows): key = row.get("source_event_key", "") row["_replay"] = "true" if key and key in seen else "false" if key: seen.add(key) result = {"row": index + 2, "rule_id": "", "route": "unrouted", "pool": "", "owner": "", "sequence": "", "reason": "no rule matched"} rule = next((r for r in policy["rules"] if matches(row, r)), None) if rule: result.update(rule_id=text(rule["id"]), route=rule["route"], reason=text(rule.get("reason")), sequence=text(rule.get("sequence"))) if rule.get("owner_from"): owner = row.get(rule["owner_from"], "") if rule.get("cover_from"): available, cover = availability(row, rule["owner_from"], rule["cover_from"]) if is_false(row.get(available)) and row.get(cover): owner = row[cover] result["owner"] = owner elif rule.get("pool"): members = pools[rule["pool"]] result["pool"] = rule["pool"] result["owner"] = text(members[counters[rule["pool"]] % len(members)]) counters[rule["pool"]] += 1 elif rule["route"] == "fallback": result["owner"] = text(policy.get("fallback_owner")) results.append(result) return results def expected_owner(row, column): available, cover = availability(row, column, "covering_rep") if is_false(row.get(available)) and row.get(cover): return row[cover] return row.get(column, "") def available_owner(row, column): """Who can act for the owner in column now: the owner, the covering rep while the owner is away, or blank when the owner is blank or away with no cover (then the fallback is right).""" owner = row.get(column, "") if not owner: return "" available, cover = availability(row, column, "covering_rep") if is_false(row.get(available)): return row.get(cover, "") return owner def sequence_problems(row, accepted, reply_window): problems = [] if low(row.get("email_validation")) not in accepted: problems.append("email_validation is %s" % (row.get("email_validation") or "blank")) if not is_true(row.get("message_version_approved")): problems.append("message version not approved") age, expiry = number(row.get("evidence_age_days")), number(row.get("expiry_days")) if age is None or expiry is None: problems.append("evidence age or expiry unknown") elif age > expiry: problems.append("evidence is %g days old, expiry %g" % (age, expiry)) if is_true(row.get("active_sequence")): problems.append("contact is in an active sequence") reply = number(row.get("last_human_reply_days")) if reply is not None and reply <= reply_window: problems.append("human reply %g days ago" % reply) if low(row.get("identity_status")) != "matched": problems.append("identity is %s" % (row.get("identity_status") or "blank")) return problems def check(rows, results, policy, args): findings = [] stats = Counter() accepted = {low(v) for v in args.accept_validation.split(",") if v.strip()} def add(severity, inv, res, message): findings.append({"severity": severity, "check": inv, "name": INVARIANTS[inv], "row": res["row"] if res else None, "message": message}) no_key, no_episode = [], [] by_episode = OrderedDict() for row, res in zip(rows, results): route, owner = res["route"], res["owner"] where = "episode %s (row %d)" % (row.get("episode_id") or "?", res["row"]) tag = " (rule %s)" % (res["rule_id"] or "none") if row["_replay"] == "true": stats["replays"] += 1 if route != "replay": add("FAIL", "I1", res, "%s repeats event key %s but routed to %s%s; only the first delivery may act" % (where, row["source_event_key"], route, tag)) continue if not row.get("source_event_key"): no_key.append(str(res["row"])) if route == "replay": add("FAIL", "I1", res, "%s is the first delivery of its event key but routed to replay%s; the signal is lost" % (where, tag)) continue if route == "unrouted": add("FAIL", "I6", res, "%s matched no rule; add a monitored fallback rule at the end" % where) continue stats["actions"] += 1 direct = is_true(row.get("direct_request")) matched = low(row.get("identity_status")) == "matched" if route in ACTION_ROUTES: if row.get("episode_id"): by_episode.setdefault(row["episode_id"], []).append(res) else: no_episode.append(str(res["row"])) if is_true(row.get("suppressed")): stats["suppressed"] += 1 if route in ("sequence", "nurture"): add("FAIL", "I2", res, "%s is suppressed but routed to %s%s" % (where, route, tag)) elif route in TASK_ROUTES and not direct: add("FAIL", "I2", res, "%s is suppressed and not a direct request but routed to %s%s" % (where, route, tag)) if is_true(row.get("open_opportunity")): stats["open_opp"] += 1 want = expected_owner(row, "opportunity_owner") if route in PROSPECTING: add("FAIL", "I3", res, "%s has an open opportunity but routed to %s%s; expected opp_owner_task to %s" % (where, route, tag, want or "the opportunity owner")) elif route in ("owner_task", "opp_owner_task") and owner != want: add("FAIL", "I3", res, "%s assigned %s but the opportunity owner is %s%s" % (where, owner or "nobody", want or "blank", tag)) elif route == "fallback": now = available_owner(row, "opportunity_owner") if now and owner != now: add("FAIL", "I3", res, "%s has an open opportunity but went to the fallback owner %s%s; expected " "opp_owner_task to %s" % (where, owner or "nobody", tag, now)) account_owner = row.get("account_owner", "") if account_owner and route in OWNER_CHECKED and owner: stats["owned"] += 1 want = expected_owner(row, "account_owner") if owner != want: cover = " (covering rep %s)" % want if want != account_owner else "" add("FAIL", "I4", res, "%s assigned %s but account owner is %s%s%s" % (where, owner, account_owner, cover, tag)) elif account_owner and route == "sequence": stats["owned"] += 1 add("FAIL", "I4", res, "%s has account owner %s but routed to a sequence that assigns nobody%s; a sequence " "is for unowned new business, or it names that owner" % (where, account_owner, tag)) elif account_owner and route == "fallback" and matched: stats["owned"] += 1 now = available_owner(row, "account_owner") if now and owner != now: cover = " (covering rep %s)" % now if now != account_owner else "" add("FAIL", "I4", res, "%s went to the fallback owner %s but account owner is %s%s%s" % (where, owner or "nobody", account_owner, cover, tag)) if is_true(row.get("customer")): stats["customer"] += 1 now = available_owner(row, "account_owner") if route in PROSPECTING: add("FAIL", "I5", res, "%s is a customer but routed to %s%s; expected cs_task" % (where, route, tag)) elif route == "fallback" and now and owner != now: add("FAIL", "I5", res, "%s is a customer but went to the fallback while %s can act%s; expected cs_task" % (where, now, tag)) if route in NEEDS_OWNER and not owner: add("FAIL", "I6", res, "%s routed to %s with no owner%s" % (where, route, tag)) if route == "sequence": stats["sequence"] += 1 problems = sequence_problems(row, accepted, args.reply_window) if problems: add("FAIL", "I7", res, "%s routed to sequence%s but %s" % (where, tag, "; ".join(problems))) if low(row.get("signal_class")) == "inferred": stats["inferred_seq"] += 1 if not is_true(row.get("corroborated")): add("FAIL", "I8", res, "%s is an inferred signal routed to sequence without corroboration%s" % (where, tag)) if direct: stats["direct"] += 1 if route not in HUMAN_ROUTES: add("FAIL", "I10", res, "%s is a direct request but routed to %s%s; it needs a human route" % (where, route, tag)) if not matched: stats["unmatched"] += 1 allowed = UNMATCHED_SAFE | (HUMAN_ROUTES if direct else set()) if route not in allowed: add("FAIL", "I11", res, "%s has identity %s but routed to %s%s; hold it or send it to a human" % (where, row.get("identity_status") or "blank", route, tag)) # Direct requests never expire. I7 already checks evidence age on sequence rows. if not direct and route != "sequence": age, expiry = number(row.get("evidence_age_days")), number(row.get("expiry_days")) if age is None or expiry is None: stats["undated"] += 1 if route in UNDATED_BLOCKED: gap = "no evidence age" if age is None else "no expiry_days" add("FAIL", "I12", res, "%s has %s but routed to %s%s; undated evidence goes to research" % (where, gap, route, tag)) elif age > expiry: stats["expired"] += 1 if route in EXPIRED_BLOCKED: add("FAIL", "I12", res, "%s has evidence %g days old, past its %g-day expiry, but routed to %s%s; " "expired evidence creates no work" % (where, age, expiry, route, tag)) # Business dedupe: the same news from a second source attaches to the open item. for episode, items in by_episode.items(): if len(items) > 1: add("WARN", "I1", items[0], "episode %s created %d actions from separate deliveries (rows %s; routes %s); " "attach later evidence to the open item instead" % (episode, len(items), ", ".join(str(r["row"]) for r in items), ", ".join(r["route"] for r in items))) if no_key: add("WARN", "I1", None, "%d rows have no source_event_key (rows %s); repeat deliveries cannot be detected" % (len(no_key), ", ".join(no_key))) if no_episode: add("WARN", "I1", None, "%d action rows have no episode_id (rows %s); the same news from two sources " "cannot be grouped" % (len(no_episode), ", ".join(no_episode))) passes = [] if args.capacity: pools = policy.get("pools") or {} owned = Counter(res["owner"] for row, res in zip(rows, results) if res["owner"] and row["_replay"] != "true" and res["route"] not in ("replay", "unrouted")) for name, limit in args.capacity.items(): members = [text(m) for m in pools.get(name, [name])] count = sum(owned[m] for m in members) kind = "pool" if name in pools else "owner" if count > limit: add("WARN", "I9", None, "%s %s holds %d items, above capacity %d for this period; raise the band " "threshold or let overflow expire" % (kind, name, count, limit)) else: passes.append(("PASS", "I9", "%s %s holds %d items, within capacity %d" % (kind, name, count, limit))) flagged = {f["check"] for f in findings} summary = { "I1": "%d later deliveries of an event key routed to replay; no episode created a second action" % stats["replays"], "I2": "%d suppressed rows; none sequenced or nurtured; tasks, the fallback included, only for direct requests" % stats["suppressed"], "I3": "%d open-opportunity rows kept with the opportunity owner or covering rep, the fallback only when " "neither can act; none in a sequence, pool task, nurture or research" % stats["open_opp"], "I4": "%d owned rows kept with the account owner or covering rep, the fallback only when neither can act" % stats["owned"], "I5": "%d customer rows; none in a sequence, pool task, nurture or research, or in the fallback while the " "account owner can act" % stats["customer"], "I6": "every row has one route and every task route has an owner", "I7": "%d sequence rows meet every enrollment condition" % stats["sequence"], "I8": "inferred rows sequenced: %d, all corroborated" % stats["inferred_seq"], "I10": "%d direct requests reached a human route" % stats["direct"], "I11": "%d rows without a matched identity held or sent to a human" % stats["unmatched"], "I12": "expired rows: %d, none given a task or nurture; undated rows: %d, none given an owner, pool or " "fallback task or nurture (I7 covers sequences)" % (stats["expired"], stats["undated"]), } for inv, message in summary.items(): if inv not in flagged: passes.append(("PASS", inv, message)) if not args.capacity: passes.append(("INFO", "I9", "not checked; pass --capacity with pool or owner limits for this period")) passes.sort(key=lambda p: int(p[1][1:])) findings.sort(key=lambda f: (SEVERITY_ORDER[f["severity"]], int(f["check"][1:]), f["row"] or 0)) return findings, passes def parse_capacity(value): caps = OrderedDict() for part in (value or "").split(","): if not part.strip(): continue name, sep, amount = part.partition("=") if not sep or not name.strip() or not amount.strip().isdigit(): raise InputError("--capacity takes name=number pairs, for example ae=22,sdr=40") caps[name.strip()] = int(amount) return caps def write_out(path, rows, results): columns = ["row", "episode_id", "source_event_key", "signal_class", "signal_type", "account_id", "contact_id", "replay", "rule_id", "route", "pool", "owner", "sequence", "reason"] with open(path, "w", newline="", encoding="utf-8") as handle: writer = csv.writer(handle) writer.writerow(columns) for row, res in zip(rows, results): writer.writerow([res["row"], row.get("episode_id", ""), row.get("source_event_key", ""), row.get("signal_class", ""), row.get("signal_type", ""), row.get("account_id", ""), row.get("contact_id", ""), row["_replay"], res["rule_id"], res["route"], res["pool"], res["owner"], res["sequence"], res["reason"]]) def main(): parser = argparse.ArgumentParser(description="Dry-run a routing policy and check invariants I1 to I12.") parser.add_argument("--policy", required=True) parser.add_argument("--records", required=True) parser.add_argument("--capacity", default="") parser.add_argument("--accept-validation", default="valid") parser.add_argument("--reply-window", type=float, default=30) parser.add_argument("--out") parser.add_argument("--json", action="store_true") args = parser.parse_args() try: args.capacity = parse_capacity(args.capacity) columns, rows = load_records(args.records) policy = load_policy(args.policy, columns) except InputError as exc: print("error: %s" % exc, file=sys.stderr) return 2 results = route_rows(policy, rows) findings, passes = check(rows, results, policy, args) if args.out: try: write_out(args.out, rows, results) except OSError as exc: print("error: cannot write %s: %s" % (args.out, exc), file=sys.stderr) return 2 routes = Counter(res["route"] for res in results) pool_tasks = Counter(res["pool"] for res in results if res["route"] == "pool_task") hits = Counter(res["rule_id"] for res in results if res["rule_id"]) rule_ids = [text(r["id"]) for r in policy["rules"]] never = [rid for rid in rule_ids if not hits[rid]] episodes = len({r.get("episode_id") for r in rows}) repeats = sum(1 for r in rows if r["_replay"] == "true") replay = routes["replay"] unrouted = routes["unrouted"] routed = len(rows) - replay - unrouted fails = sum(1 for f in findings if f["severity"] == "FAIL") warns = sum(1 for f in findings if f["severity"] == "WARN") failed = sorted({f["check"] for f in findings if f["severity"] == "FAIL"}, key=lambda c: int(c[1:])) if args.json: print(json.dumps({ "policy": {"policy_id": policy.get("policy_id", ""), "version": text(policy.get("version"))}, "rows": len(rows), "unique_episodes": episodes, "repeat_deliveries": repeats, "routes": {r: routes[r] for r in list(ROUTES) + ["unrouted"] if routes[r]}, "pool_tasks": dict(pool_tasks), "rule_hits": {rid: hits[rid] for rid in rule_ids}, "never_hit": never, "findings": findings, "passes": [{"severity": s, "check": c, "message": m} for s, c, m in passes], "reconciliation": {"rows": len(rows), "replay": replay, "routed": routed, "unrouted": unrouted}, "fail": fails, "warn": warns, }, indent=2)) return 1 if fails else 0 print("Routing dry run: policy %s version %s on %s" % (policy.get("policy_id", "?"), text(policy.get("version")) or "?", args.records)) print("Rows: %d (%d episodes, %d repeat deliveries)" % (len(rows), episodes, repeats)) print("") print("Routes:") for route in list(ROUTES) + ["unrouted"]: if routes[route]: extra = "" if route == "pool_task": extra = " (%s)" % ", ".join("%s %d" % (p, n) for p, n in sorted(pool_tasks.items())) print(" %-16s %3d%s" % (route, routes[route], extra)) print("") print("Rule hits:") for rid in rule_ids: print(" %-6s %3d" % (rid, hits[rid])) for rid in never: print("INFO rule %s never matched a row; prove it with a test row before relying on it" % rid) print("") print("Invariants:") for f in findings: print("%s %s %s: %s" % (f["severity"], f["check"], f["name"], f["message"])) for severity, inv, message in passes: print("%s %s %s: %s" % (severity, inv, INVARIANTS[inv], message)) print("") print("Reconciliation: %d rows = %d replay + %d routed + %d unrouted" % (len(rows), replay, routed, unrouted)) tail = " Failed invariants: %s." % ", ".join(failed) if failed else "" print("%d FAIL, %d WARN.%s Nothing was changed." % (fails, warns, tail)) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main())