#!/usr/bin/env python3
"""Check each deletion against your retention register: runs missed, runs by the wrong job, tables with no rule,
and stretches with no record. Python 3 and DuckDB only. It reads files; it never connects to a platform.

    python3 retention_deletion_check.py --queries                      # the export statements for each platform
    python3 retention_deletion_check.py --dir exports --start 2026-10-01 --end 2026-10-31
    python3 retention_deletion_check.py --make-demo demo-clean         # generated sample exports (invented names)
    python3 retention_deletion_check.py --make-demo demo-planted --plant findings   # or --plant inputs

Folder --dir holds (names fixed; a platform you do not export simply has no file):
    register.csv              table_name, platform, rule_id, policy_reference, retention_days, cadence_days, job_type, job
    coverage.csv              platform, covered_from, covered_to   (one row per export you took; see --queries)
    snowflake_deletions.csv   query_id, query_text, database_name, schema_name, query_type, user_name, role_name,
                              query_tag, execution_status, start_time
    databricks_history.csv    table_name, version, timestamp, user_name, operation, job_id
    redshift_deletions.csv    query_id, username, database_name, query_label, query_type, status, start_time, query_text
    exceptions.csv            platform, table_name, reason   (optional: tables that need no rule; name* allowed)

Your own policy sets every period. Nothing here says a period is enough for any law or framework.
Exit code: 0 no findings, 1 findings, 2 an input check failed (the lists are INCOMPLETE).
"""
import argparse
import csv
import datetime as dt
import fnmatch
import hashlib
import pathlib
import re
import sys

import duckdb

PLATFORMS = ("snowflake", "databricks", "redshift")
FILES = {"register": "register.csv", "coverage": "coverage.csv", "snowflake": "snowflake_deletions.csv",
         "databricks": "databricks_history.csv", "redshift": "redshift_deletions.csv",
         "exceptions": "exceptions.csv"}
COLUMNS = {
    "register": ["table_name", "platform", "rule_id", "policy_reference", "retention_days", "cadence_days",
                 "job_type", "job"],
    "coverage": ["platform", "covered_from", "covered_to"],
    "snowflake": ["query_id", "query_text", "database_name", "schema_name", "query_type", "user_name", "role_name",
                  "query_tag", "execution_status", "start_time"],
    "databricks": ["table_name", "version", "timestamp", "user_name", "operation", "job_id"],
    "redshift": ["query_id", "username", "database_name", "query_label", "query_type", "status", "start_time",
                 "query_text"],
    "exceptions": ["platform", "table_name", "reason"],
}
# What each platform's register row may name as "the job" (the field of the deletion record it is compared with).
JOB_TYPES = {"snowflake": ("query_tag", "user", "role"), "databricks": ("job_id", "user"),
             "redshift": ("user", "label")}
# Longest span one export can reach back, from the vendor's own page (see the guide's Sources). Databricks keeps
# history per table (logRetentionDuration, 30 days by default), so its start is read from each table's own rows.
VENDOR_DAYS = {"snowflake": 365, "redshift": 7}
DELETE_OPS = {"databricks": {"delete", "truncate"}}
SF_TYPES = {"delete", "truncate_table"}
DATE = re.compile(r"\d{4}-\d{2}-\d{2}")
STAMP = re.compile(r"(\d{4}-\d{2}-\d{2})[ T](\d{2}:\d{2}:\d{2})(\.\d{1,9})?(Z|\+00:00)?")

QUERIES = """\
-- 1. Snowflake. Needs a role that can read SNOWFLAKE.ACCOUNT_USAGE (it keeps 365 days; latency up to 45 minutes).
--    Without it, INFORMATION_SCHEMA.QUERY_HISTORY_BY_USER keeps about 7 days for one user (same columns).
SELECT query_id, query_text, database_name, schema_name, query_type, user_name, role_name, query_tag,
       execution_status, TO_CHAR(CONVERT_TIMEZONE('UTC', start_time), 'YYYY-MM-DD HH24:MI:SS') AS start_time
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE query_type IN ('DELETE', 'TRUNCATE_TABLE')
  AND start_time >= DATEADD('day', -365, CURRENT_TIMESTAMP())
ORDER BY start_time, query_id;
-- Save as snowflake_deletions.csv. In coverage.csv: snowflake, <today minus 364 days> (the first full day), <today>.

-- 2. Databricks. A notebook, once per register table (DESCRIBE HISTORY cannot be a subquery). Keep ALL operations:
--    the earliest row tells the check where each table's history starts.
--    spark.conf.set("spark.sql.session.timeZone", "UTC")
--    for t in TABLES_TO_WATCH:   # every register table, plus any other table you want NO RULE to catch
--        spark.sql(f"DESCRIBE HISTORY {t}").selectExpr(f"'{t}' AS table_name", "version",
--            "date_format(timestamp, 'yyyy-MM-dd HH:mm:ss') AS timestamp", "userName AS user_name", "operation",
--            "job.jobId AS job_id")   # null when no job wrote the commit
--    Save the union as databricks_history.csv. In coverage.csv: databricks, (leave blank), <today>.

-- 3. Redshift. AWS documents seven days of log history for its STL views and states no period for SYS_QUERY_HISTORY,
--    so plan for about seven: run this at least weekly and add every export's rows to one file. One coverage.csv
--    row per export: redshift, <first day it reached>, <day you ran it>. Run it as a superuser (or a user with
--    SYSLOG ACCESS UNRESTRICTED): other users see only their own rows.
SELECT query_id, TRIM(username) AS username, TRIM(database_name) AS database_name, TRIM(query_label) AS query_label,
       TRIM(query_type) AS query_type, TRIM(status) AS status,
       TO_CHAR(start_time, 'YYYY-MM-DD HH24:MI:SS') AS start_time, TRIM(query_text) AS query_text
FROM sys_query_history
WHERE LOWER(query_text) LIKE '%delete%' OR LOWER(query_text) LIKE '%truncate%'   -- broad on purpose; the script picks out the deletions
ORDER BY start_time, query_id;
-- Save as redshift_deletions.csv (times in UTC).
"""


# ------------------------------------------------------------------ small helpers

def parse_date(text):
    t = (text or "").strip()
    if not DATE.fullmatch(t):
        return None
    try:
        return dt.date.fromisoformat(t)
    except ValueError:
        return None


def parse_stamp(text):
    m = STAMP.fullmatch((text or "").strip())
    if not m:
        return None
    try:
        return dt.datetime.strptime(m.group(1) + " " + m.group(2), "%Y-%m-%d %H:%M:%S")
    except ValueError:
        return None


IDENT = re.compile(r'\s*(?:"((?:[^"]|"")*)"|`([^`]*)`|([A-Za-z_][\w$]*))')


def parse_name(text, pos=0):
    """a."B".c at text[pos:] -> (['a','B','c'], end position), or (None, pos)."""
    parts = []
    while True:
        m = IDENT.match(text, pos)
        if not m:
            return (parts or None), pos
        parts.append((m.group(1) or "").replace('""', '"') if m.group(1) is not None
                     else (m.group(2) if m.group(2) is not None else m.group(3)))
        pos = m.end()
        dot = re.compile(r"\s*\.").match(text, pos)
        if not dot:
            return parts, pos
        pos = dot.end()


def strip_leading_comments(text):
    t = text.lstrip()
    while True:
        if t.startswith("--") or t.startswith("//"):
            nl = t.find("\n")
            t = "" if nl < 0 else t[nl + 1:].lstrip()
        elif t.startswith("/*"):
            end = t.find("*/")
            t = "" if end < 0 else t[end + 2:].lstrip()
        else:
            return t


TARGET = re.compile(r"(?:DELETE(?:\s+FROM)?|TRUNCATE(?:\s+TABLE)?(?:\s+IF\s+EXISTS)?)\s+(?:ONLY\s+)?", re.I)


def target_of(sql):
    """The table a statement that STARTS with DELETE [FROM] or TRUNCATE names, as name parts; else None. FROM is
    optional after DELETE on Redshift.
    A statement that only reads the table (INSERT ... SELECT, GRANT ... DELETE ON) never matches."""
    t = strip_leading_comments(sql or "")
    m = TARGET.match(t)
    if not m:
        return None
    parts, end = parse_name(t, m.end())
    rest = t[end:].lstrip()
    if parts and rest.startswith("("):
        # IDENTIFIER('DB.S.T') or TABLE('DB.S.T') writes the name as a string: read it. IDENTIFIER(:tbl),
        # IDENTIFIER($t) or IDENTIFIER(?) names the table only when the statement runs.
        lit = re.match(r"\(\s*'((?:[^']|'')*)'\s*\)", rest)
        if len(parts) == 1 and parts[0].lower() in ("identifier", "table") and lit:
            named, _ = parse_name(lit.group(1).replace("''", "'"))
            return named
        return RUN_TIME
    return parts


# A deletion whose table is named only when it runs, such as Snowflake's DELETE FROM IDENTIFIER(:tbl). Its text holds
# no table name, so no rule can be matched to it: it is an input problem, never a guess.
RUN_TIME = "run-time name"


def key_of(parts):
    return ".".join(p.lower() for p in parts)


def merge(intervals):
    out = []
    for a, b in sorted(intervals):
        if out and a <= out[-1][1] + dt.timedelta(days=1):
            out[-1] = (out[-1][0], max(out[-1][1], b))
        else:
            out.append((a, b))
    return out


def clip(intervals, lo, hi):
    return [(max(a, lo), min(b, hi)) for a, b in intervals if max(a, lo) <= min(b, hi)]


def complement(lo, hi, covered):
    out, cur = [], lo
    for a, b in covered:
        if a > cur:
            out.append((cur, a - dt.timedelta(days=1)))
        cur = max(cur, b + dt.timedelta(days=1))
    if cur <= hi:
        out.append((cur, hi))
    return out


def sha256(path):
    return hashlib.sha256(pathlib.Path(path).read_bytes()).hexdigest()


# ------------------------------------------------------------------ reading the files

def read_file(folder, kind, problems, inputs):
    """List of dicts (lower-case headers, blanks as ''), or None when the file is absent or unreadable."""
    path = folder / FILES[kind]
    if not path.is_file():
        return None
    try:
        con = duckdb.connect()
        cur = con.execute("SELECT * FROM read_csv(?, header=true, all_varchar=true, delim=',', quote='\"', "
                          "escape='\"', null_padding=true)", [str(path)])
        names = [d[0].strip().lower() for d in cur.description]
        rows = [{n: ("" if v is None else str(v)) for n, v in zip(names, r)} for r in cur.fetchall()]
    except Exception as e:  # a CSV that cannot be read fails the file check; it never crashes the run
        problems.append(f"{FILES[kind]}: cannot be read as a CSV ({str(e).splitlines()[0][:120]})")
        return None
    missing = [c for c in COLUMNS[kind] if c not in names]
    if missing:
        problems.append(f"{FILES[kind]}: no column named {', '.join(missing)}")
        return None
    inputs.append((FILES[kind], len(rows), sha256(path)))
    return rows


def load_register(rows, problems):
    """Valid rules, keyed by (platform, table key). A bad row is reported and left out, never guessed."""
    rules, seen, bad = {}, {}, 0
    for i, r in enumerate(rows, start=2):
        where = f"register.csv line {i}"
        v = {k: r[k].strip() for k in COLUMNS["register"]}
        issues = [f"{k} is blank" for k, x in v.items() if not x]
        plat = v["platform"].lower()
        if v["platform"] and plat not in PLATFORMS:
            issues.append(f"platform '{v['platform']}' is not one of {', '.join(PLATFORMS)}")
        for k in ("retention_days", "cadence_days"):
            if v[k] and not (re.fullmatch(r"\d+", v[k]) and int(v[k]) >= 1):
                issues.append(f"{k} '{v[k]}' is not a whole number of days, 1 or more")
        if plat in JOB_TYPES and v["job_type"] and v["job_type"].lower() not in JOB_TYPES[plat]:
            issues.append(f"job_type '{v['job_type']}' is not one of {', '.join(JOB_TYPES[plat])} for {plat}")
        parts, end = parse_name(v["table_name"]) if v["table_name"] else (None, 0)
        if v["table_name"] and (not parts or len(parts) != 3 or v["table_name"][end:].strip()):
            issues.append(f"table_name '{v['table_name']}' is not a three-part name (database.schema.table)")
        if not issues:
            key = (plat, key_of(parts))
            if key in seen:
                issues.append(f"table {v['table_name']} already has a rule on line {seen[key]}")
            else:
                seen[key] = i
                rules[key] = {**v, "platform": plat, "job_type": v["job_type"].lower(), "key": key[1],
                              "cadence": int(v["cadence_days"])}
        if issues:
            bad += 1
            problems.append(f"{where}: " + "; ".join(issues))
    return rules, bad


def load_coverage(rows, files_present, problems):
    """{platform: [(from, to)]}; databricks may leave covered_from blank."""
    cov, bad = {p: [] for p in PLATFORMS}, 0
    for i, r in enumerate(rows, start=2):
        plat = r["platform"].strip().lower()
        a, b = parse_date(r["covered_from"]), parse_date(r["covered_to"])
        issues = []
        if plat not in PLATFORMS:
            issues.append(f"platform '{r['platform']}' is not one of {', '.join(PLATFORMS)}")
        if b is None:
            issues.append(f"covered_to '{r['covered_to']}' is not written as YYYY-MM-DD")
        if a is None and not (plat == "databricks" and not r["covered_from"].strip()):
            issues.append(f"covered_from '{r['covered_from']}' is not written as YYYY-MM-DD")
        if a and b and a > b:
            issues.append("covered_from is after covered_to")
        cap = VENDOR_DAYS.get(plat)
        if a and b and cap and (b - a).days + 1 > cap:
            issues.append(f"claims {(b - a).days + 1} days from one export; this check allows {cap} for {plat} "
                          "(take exports more often and add one row per export)")
        if issues:
            bad += 1
            problems.append(f"coverage.csv line {i}: " + "; ".join(issues))
        elif b is not None:
            cov[plat].append((a or dt.date.min, b))
    for p in files_present:
        if not cov[p] and not any("coverage.csv" in x for x in problems):
            problems.append(f"{FILES[p]} is present but coverage.csv has no row for {p}: say which days it covers")
    return {p: merge(v) for p, v in cov.items()}, bad


def load_exceptions(rows, problems):
    out = []
    for i, r in enumerate(rows, start=2):
        plat, name, why = r["platform"].strip().lower(), r["table_name"].strip().lower(), r["reason"].strip()
        if plat not in PLATFORMS or not name or not why:
            problems.append(f"exceptions.csv line {i}: needs a platform ({', '.join(PLATFORMS)}), a table name and a "
                            "reason")
        else:
            out.append((plat, name))
    return out


# ------------------------------------------------------------------ turning exports into deletion events

def events_from(platform, rows, rules, problems, window, args):
    """-> (events, stats). An event is one successful deletion: platform, table key, who, when, id."""
    lastparts = {k[1].split(".")[-1] for k in rules if k[0] == platform}
    stats = {"rows": len(rows), "not_deletion": 0, "not_successful": 0, "bad_time": 0, "outside_period": 0}
    events, unread = [], {"time": 0, "status": 0, "name": 0, "run_time": 0}

    def touches(text):
        low = (text or "").lower()
        return any(p in low for p in lastparts)

    for r in rows:
        if platform == "databricks":
            parts, _ = parse_name(r["table_name"].strip()) if r["table_name"].strip() else (None, 0)
            ts = parse_stamp(r["timestamp"])
            if ts is None or not parts or len(parts) != 3:
                unread["time" if ts is None else "name"] += 1
                continue
            if r["operation"].strip().lower() not in DELETE_OPS["databricks"]:
                stats["not_deletion"] += 1
                continue
            actor = {"user": r["user_name"].strip(), "job_id": r["job_id"].strip()}
            events.append((platform, key_of(parts), actor, ts, f"v{r['version'].strip()}"))
            continue
        text = r["query_text"]
        if platform == "snowflake":
            if r["query_type"].strip().lower() not in SF_TYPES:
                stats["not_deletion"] += 1
                continue
            status_ok = r["execution_status"].strip().lower() == "success"
            status_known = bool(r["execution_status"].strip())
            actor = {"user": r["user_name"].strip(), "role": r["role_name"].strip(), "query_tag": r["query_tag"].strip()}
            db, sch = r["database_name"].strip(), r["schema_name"].strip()
        else:
            parts0 = target_of(text)
            if parts0 is None:
                stats["not_deletion"] += 1
                continue
            status_ok = r["status"].strip().lower() == "success"
            status_known = bool(r["status"].strip())
            actor = {"user": r["username"].strip(), "label": r["query_label"].strip()}
            db, sch = r["database_name"].strip(), args.redshift_default_schema
        ts = parse_stamp(r["start_time"])
        parts = target_of(text)
        if ts is None:
            stats["bad_time"] += 1
            if touches(text):
                unread["time"] += 1
            continue
        if not status_known:
            if touches(text):
                unread["status"] += 1
            continue
        if not status_ok:
            stats["not_successful"] += 1      # a refused or failed attempt is not a deletion
            continue
        if parts == RUN_TIME:
            if window[0] <= ts.date() <= window[1]:
                unread["run_time"] += 1    # only a deletion in the period can hide a run or a rogue deletion
            else:
                stats["outside_period"] += 1
            continue
        if parts is not None and len(parts) == 2 and db:
            parts = [db] + parts
        elif parts is not None and len(parts) == 1 and db and sch:
            parts = [db, sch] + parts
        if parts is None or len(parts) != 3:
            if touches(text):
                unread["name"] += 1
            continue
        events.append((platform, key_of(parts), actor, ts, r["query_id"].strip()))
    lo, hi = window
    kept = []
    for e in events:
        if lo <= e[3].date() <= hi:
            kept.append(e)
        else:
            stats["outside_period"] += 1
    fname = FILES[platform]
    if unread["time"]:
        problems.append(f"{fname}: {unread['time']} deletion row(s) with a start time not written as "
                        "YYYY-MM-DD HH:MM:SS (UTC), left out")
    if unread["status"]:
        problems.append(f"{fname}: {unread['status']} row(s) touching a register table have no status, left out")
    if unread["run_time"]:
        problems.append(f"{fname}: {unread['run_time']} deletion row(s) with a table named only when the statement runs "
                        "(IDENTIFIER(...)), so the table is unknown, left out")
    if unread["name"]:
        problems.append(f"{fname}: {unread['name']} deletion row(s) touching a register table name no table this "
                        "check can resolve to database.schema.table, left out")
    return kept, stats


def actor_matches(rule, actor):
    jt, job = rule["job_type"], rule["job"].strip().lower()
    return actor.get(jt, "").strip().lower() == job


def describe_job(rule):
    return {"query_tag": "query tag", "user": "user", "role": "role", "job_id": "job id",
            "label": "query label"}[rule["job_type"]] + f" {rule['job']}"


def describe_actor(platform, actor):
    shown = lambda v: v if v else "(blank)"
    keys = {"snowflake": ("user", "role", "query_tag"), "databricks": ("user", "job_id"),
            "redshift": ("user", "label")}[platform]
    return ", ".join(f"{k.replace('_', ' ')} {shown(actor.get(k, ''))}" for k in keys)


# ------------------------------------------------------------------ the four lists

def evaluate(rules, coverage, events_by_platform, db_info, exceptions, window):
    lo, hi = window
    day = dt.timedelta(days=1)
    missed, notby, norule, norec = [], [], [], []
    by_table = {}
    for ev in events_by_platform.values():
        for e in ev:
            by_table.setdefault((e[0], e[1]), []).append(e)
    for key in sorted(rules):
        rule = rules[key]
        plat = rule["platform"]
        base = {"platform": plat, "table": rule["table_name"], "rule_id": rule["rule_id"],
                "policy": rule["policy_reference"]}
        if plat not in events_by_platform:
            norec.append({**base, "detail": f"NOT CHECKED: no usable {FILES[plat]} was read, so there is no deletion "
                                            "record to read (this is not 'nothing was deleted')",
                          "from": lo, "to": hi, "events": 0})
            continue
        cov = list(coverage.get(plat, []))
        start = lo
        if plat == "databricks":
            seen = db_info.get(key[1])
            if not seen:
                norec.append({**base, "detail": "no history row at all for this table in the export: the table was "
                                                "not exported, or its history is gone", "from": lo, "to": hi,
                              "events": 0})
                continue
            first, has_v0 = seen
            if first > lo:
                if has_v0:
                    start = first          # the table did not exist before its first version
                else:
                    norec.append({**base, "detail": f"the table's history in the export starts {first}; earlier "
                                                    "history has expired or was not exported",
                                  "from": lo, "to": first - day, "events": 0})
                    start = first
        covered = merge(clip(cov, start, hi))
        for a, b in complement(start, hi, covered):
            norec.append({**base, "detail": "no deletion record covers these days (no export reached them)",
                          "from": a, "to": b, "events": 0})
        evs = sorted(by_table.get(key, []), key=lambda e: (e[3], e[4]))
        named = [e for e in evs if actor_matches(rule, e[2])]
        for a, b in covered:
            days = sorted({e[3].date() for e in named if a <= e[3].date() <= b})
            pts = [a] + days + [b]
            for x, y in zip(pts, pts[1:]):
                if (y - x).days > rule["cadence"]:
                    missed.append({**base, "detail": f"{(y - x).days} days from {x} to {y} with no deletion by the "
                                                     f"named job ({describe_job(rule)}) in between; the rule runs "
                                                     f"every {rule['cadence']}", "from": x, "to": y, "events": None})
        groups = {}
        for e in evs:
            if not actor_matches(rule, e[2]):
                groups.setdefault(describe_actor(plat, e[2]), []).append(e)
        for who in sorted(groups):
            g = groups[who]
            notby.append({**base, "detail": f"deletion not by the named job ({describe_job(rule)}): {who}",
                          "from": min(e[3] for e in g).date(), "to": max(e[3] for e in g).date(), "events": len(g)})
    ruled = set(rules)
    for tk in sorted(by_table):
        if tk in ruled:
            continue
        if any(p == tk[0] and fnmatch.fnmatchcase(tk[1], pat) for p, pat in exceptions):
            continue
        g = by_table[tk]
        users = sorted({e[2].get("user", "") for e in g})
        norule.append({"platform": tk[0], "table": tk[1], "rule_id": "", "policy": "",
                       "detail": f"deletion on a table the register does not list, by {len(users)} user(s)",
                       "from": min(e[3] for e in g).date(), "to": max(e[3] for e in g).date(), "events": len(g)})
    return missed, notby, norule, norec


# ------------------------------------------------------------------ reporting

def print_list(title, meaning, items):
    print(f"\n{title}  ({len(items)})")
    print(f"  {meaning}")
    if not items:
        print("  none")
    for it in items:
        who = f"{it['rule_id']} | " if it["rule_id"] else ""
        print(f"  {who}{it['platform']} | {it['table']}" + (f" | policy {it['policy']}" if it["policy"] else ""))
        span = f"{it['from']}" if it["from"] == it["to"] else f"{it['from']} to {it['to']}"
        count = "" if it["events"] is None else f"; {it['events']} deletion(s)"
        print(f"      {it['detail']}  [{span}{count}]")


def write_evidence(out_root, lists, inputs, fingerprint, args, result):
    out_root = pathlib.Path(out_root)
    stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    folder, n = out_root / f"run-{stamp}", 1
    while folder.exists():
        n += 1
        folder = out_root / f"run-{stamp}-{n}"
    folder.mkdir(parents=True)
    with open(folder / "findings.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["list", "platform", "table_name", "rule_id", "policy_reference", "detail", "from_date",
                    "to_date", "deletions"])
        for name, items in lists:
            for it in items:
                w.writerow([name, it["platform"], it["table"], it["rule_id"], it["policy"], it["detail"],
                            it["from"], it["to"], it["events"]])
    with open(folder / "inputs.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["file", "rows", "sha256"])
        w.writerows(inputs)
        w.writerow(["deletion events used", "", fingerprint])
    with open(folder / "run.txt", "w", encoding="utf-8") as f:
        f.write(f"period {args.start} to {args.end}; result {result}\n")
    return folder


def run_check(args):
    folder = pathlib.Path(args.dir)
    if not folder.is_dir():
        print(f"No folder {folder}. Export the records (--queries prints the statements), or try --make-demo <folder>.")
        return 2
    start, end = parse_date(args.start), parse_date(args.end)
    if not start or not end or start > end:
        print("--start and --end must be dates written YYYY-MM-DD, start not after end.")
        return 2
    problems, inputs = [], []
    reg_rows = read_file(folder, "register", problems, inputs)
    if reg_rows is None:
        if not any("register.csv" in p for p in problems):
            problems.append("register.csv is missing: there is nothing to check against")
        reg_rows = []
    rules, bad_rules = load_register(reg_rows, problems)
    data = {p: read_file(folder, p, problems, inputs) for p in PLATFORMS}
    present = [p for p in PLATFORMS if data[p] is not None]
    cov_rows = read_file(folder, "coverage", problems, inputs)
    if cov_rows is None and present and not any("coverage.csv" in p for p in problems):
        problems.append("coverage.csv is missing: say which days each export covers")
    coverage, _ = load_coverage(cov_rows or [], present if cov_rows is not None else [], problems)
    exc_rows = read_file(folder, "exceptions", problems, inputs)
    exceptions = load_exceptions(exc_rows or [], problems)

    events, stats = {}, {}
    for p in present:
        events[p], stats[p] = events_from(p, data[p], rules, problems, (start, end), args)
    db_info = {}
    if data["databricks"] is not None:
        for r in data["databricks"]:
            parts, _ = parse_name(r["table_name"].strip()) if r["table_name"].strip() else (None, 0)
            ts = parse_stamp(r["timestamp"])
            if not parts or len(parts) != 3 or ts is None:
                continue
            k = key_of(parts)
            first, v0 = db_info.get(k, (ts.date(), False))
            db_info[k] = (min(first, ts.date()), v0 or r["version"].strip() == "0")

    print(f"Retention register deletion check: period {start} to {end}")
    print("\nINPUT CHECKS")
    print(f"  [{'FAIL' if problems else 'ok'}] {len(problems)} problem(s) in the inputs")
    for p in problems:
        print(f"      - {p}")
    print("\nINPUTS (SHA-256 of each file, so a re-run shows whether the export changed)")
    for name, n, h in inputs:
        print(f"  {name}: {n} row(s), sha256 {h}")
    print(f"  register: {len(rules)} rule(s) read, {bad_rules} row(s) left out")
    print("\nPOPULATION (what each export held)")
    fp_lines = []
    for p in PLATFORMS:
        if p not in events:
            print(f"  {p}: not exported (NOT CHECKED)")
            continue
        s = stats[p]
        on_reg = sum(1 for e in events[p] if (p, e[1]) in rules)
        print(f"  {p}: {s['rows']} row(s); {s['not_deletion']} not a deletion, {s['not_successful']} refused or "
              f"failed (not counted), {s['bad_time']} bad time, {s['outside_period']} outside the period; "
              f"{len(events[p])} deletion(s) in the period, {on_reg} on register tables")
        fp_lines += [f"{e[0]}|{e[1]}|{sorted(e[2].items())}|{e[3]}|{e[4]}" for e in events[p]]
    fingerprint = hashlib.sha256("\n".join(sorted(fp_lines)).encode("utf-8")).hexdigest()
    print(f"  deletion events used: sha256 {fingerprint}")

    missed, notby, norule, norec = evaluate(rules, coverage, events, db_info, exceptions, (start, end))
    lists = [("MISSED RUN", missed), ("NOT BY THE NAMED JOB", notby), ("NO RULE", norule), ("NO RECORD", norec)]
    print_list("1. MISSED RUN", "A rule whose named job left a stretch longer than its cadence with no deletion.",
               missed)
    print_list("2. NOT BY THE NAMED JOB", "A deletion on a register table by someone or something else.", notby)
    print_list("3. NO RULE", "A deletion on a table the register does not list.", norule)
    print_list("4. NO RECORD", "Days or tables the exports cannot speak for. Never read as 'nothing was deleted'.",
               norec)
    total = sum(len(i) for _, i in lists)
    result = "INCOMPLETE" if problems else ("FINDINGS" if total else "CLEAN")
    print(f"\nSUMMARY: {len(missed)} missed run, {len(notby)} not by the named job, {len(norule)} no rule, "
          f"{len(norec)} no record")
    note = {"INCOMPLETE": "an input check failed: the lists above may be wrong or short",
            "FINDINGS": "each finding is for a person to explain; it is evidence, not a verdict",
            "CLEAN": "no finding in the records exported; read NO RECORD above before relying on it"}[result]
    print(f"RESULT: {result}: {note}")
    ev = write_evidence(args.out, lists, inputs, fingerprint, args, result)
    print(f"Evidence written to {ev}")
    return 2 if problems else (1 if total else 0)


# ------------------------------------------------------------------ generated sample exports (invented)

def make_demo(folder, plant):
    """Writes generated exports for 2026-10-01 to 2026-10-31. Every name is invented; example.com addresses only.
    plant='findings' plants one of each list; plant='inputs' plants one fault for each input check."""
    folder = pathlib.Path(folder)
    folder.mkdir(parents=True, exist_ok=True)
    d = lambda day: dt.date(2026, 10, day)
    stamp = lambda day, h=2: f"2026-10-{day:02d} {h:02d}:00:05"
    reg = [
        ["FINANCE_DB.RETAIL.ORDER_ARCHIVE", "snowflake", "R-101", "DS-4.1", "730", "7", "query_tag", "retention-purge"],
        ["FINANCE_DB.RETAIL.CALL_RECORDINGS", "snowflake", "R-102", "DS-4.2", "365", "15", "query_tag",
         "retention-purge"],
        ["main.sales.web_sessions", "databricks", "R-201", "DS-4.3", "90", "7", "job_id", "1001"],
        ["main.sales.support_tickets", "databricks", "R-202", "DS-4.4", "1095", "14", "job_id", "1002"],
        ["analytics.public.clickstream", "redshift", "R-301", "DS-4.5", "400", "7", "user", "svc_retention"],
        ["analytics.public.audit_extracts", "redshift", "R-302", "DS-4.6", "2555", "7", "user", "svc_retention"],
    ]
    sf, dbx, rs = [], [], []

    def snow(qid, table, day, user="SVC_PURGE", tag="retention-purge", status="SUCCESS", verb="DELETE FROM"):
        sf.append([qid, f"{verb} {table} WHERE created_at < DATEADD('day', -730, CURRENT_DATE())",
                   "FINANCE_DB", "RETAIL", "TRUNCATE_TABLE" if verb.startswith("TRUNC") else "DELETE", user,
                   "PURGE_ROLE", tag, status, stamp(day)])

    for day in range(1, 32):
        snow(f"sf-a-{day}", "FINANCE_DB.RETAIL.ORDER_ARCHIVE", day)
    for day in (1, 16, 31):
        snow(f"sf-c-{day}", "retail.call_recordings", day)
    snow("sf-x-1", "FINANCE_DB.RETAIL.ORDER_ARCHIVE", 12, status="FAILED_WITH_ERROR", user="TEMP_USER", tag="")

    def dbrow(table, version, day, op, user, job):
        dbx.append([table, str(version), stamp(day, 3), user, op, job])

    for t in ("main.sales.web_sessions", "main.sales.support_tickets"):
        dbx.append([t, "0", "2026-09-15 08:00:00", "svc-loader", "CREATE TABLE", ""])
    v = 1
    for day in range(1, 32):
        dbrow("main.sales.web_sessions", v, day, "DELETE", "svc-purge", "1001")
        v += 1
    v = 1
    for day in (2, 9, 16, 23, 30):
        dbrow("main.sales.support_tickets", v, day, "DELETE", "svc-purge", "1002")
        v += 1
        dbrow("main.sales.support_tickets", v, day, "WRITE", "svc-loader", "")
        v += 1

    def red(qid, table, day, user="svc_retention", status="success", label="purge"):
        rs.append([qid, user, "analytics", label, "OTHER", status, stamp(day, 4), f"DELETE FROM {table}"])

    for day in range(1, 32):
        red(f"rs-a-{day}", "public.clickstream", day)
    for day in (3, 10, 17, 24, 31):
        red(f"rs-b-{day}", "analytics.public.audit_extracts", day)
    cov = [["snowflake", "2026-10-01", "2026-10-31"], ["databricks", "", "2026-10-31"],
           ["redshift", "2026-10-01", "2026-10-07"], ["redshift", "2026-10-08", "2026-10-14"],
           ["redshift", "2026-10-15", "2026-10-21"], ["redshift", "2026-10-22", "2026-10-28"],
           ["redshift", "2026-10-29", "2026-10-31"]]
    exc = [["snowflake", "finance_db.scratch.*", "scratch schema, rebuilt daily by analysts"]]

    if plant == "findings":
        sf[:] = [r for r in sf if r[0] != "sf-c-16"]                       # MISSED RUN: R-102's 16 Oct run never ran
        for r in dbx:                                                     # NOT BY THE NAMED JOB: 16 Oct by hand
            if r[0].endswith("support_tickets") and r[4] == "DELETE" and r[2].startswith("2026-10-16"):
                r[3], r[5] = "analyst1@example.com", ""
        rs.append(["rs-n-12", "analyst2", "analytics", "", "OTHER", "success", stamp(12, 9),
                   "DELETE FROM analytics.public.tmp_exports WHERE id > 0"])   # NO RULE
        keep = []                                                          # NO RECORD (a): no export for 15-21 Oct
        for r in rs:
            if not "2026-10-15" <= r[6][:10] <= "2026-10-21":
                keep.append(r)
        rs[:] = keep
        cov = [c for c in cov if c[1] != "2026-10-15"]
        dbx[:] = [r for r in dbx if not (r[0].endswith("web_sessions") and r[1] != "0" and r[2][:10] < "2026-10-08")]
        dbx[:] = [r for r in dbx if not (r[0].endswith("web_sessions") and r[1] == "0")]   # NO RECORD (b): history
    if plant == "inputs":
        reg.append(["", "snowflake", "", "DS-9", "30", "7", "user", "svc"])                  # blank cells
        reg.append(["a.b.c", "oracle", "R-901", "DS-9", "30", "7", "user", "svc"])           # unknown platform
        reg.append(["a.b.d", "snowflake", "R-902", "DS-9", "30", "weekly", "user", "svc"])   # cadence not a number
        reg.append(["main.sales.x", "databricks", "R-903", "DS-9", "30", "7", "query_tag", "t"])  # wrong job_type
        reg.append(["orders", "snowflake", "R-904", "DS-9", "30", "7", "user", "svc"])       # not three parts
        reg.append(reg[0][:2] + ["R-905"] + reg[0][3:])                                      # same table twice
        sf.append(["sf-bad-1", "DELETE FROM FINANCE_DB.RETAIL.ORDER_ARCHIVE", "FINANCE_DB", "RETAIL", "DELETE",
                   "SVC_PURGE", "PURGE_ROLE", "retention-purge", "SUCCESS", "10/05/2026 02:00"])  # time not strict
        sf.append(["sf-bad-2", "DELETE FROM FINANCE_DB.RETAIL.ORDER_ARCHIVE", "FINANCE_DB", "RETAIL", "DELETE",
                   "SVC_PURGE", "PURGE_ROLE", "retention-purge", "", stamp(5)])                  # no status
        sf.append(["sf-bad-3", "DELETE FROM (SELECT * FROM order_archive)", "", "", "DELETE", "SVC_PURGE",
                   "PURGE_ROLE", "retention-purge", "SUCCESS", stamp(6)])                       # name unresolvable
        sf.append(["sf-bad-4", "DELETE FROM IDENTIFIER(:tbl) WHERE created_at < :cutoff", "FINANCE_DB", "RETAIL",
                   "DELETE", "SVC_PURGE", "PURGE_ROLE", "retention-purge", "SUCCESS", stamp(7)])  # named at run time
        cov.append(["redshift", "2026-10-01", "2026-10-31"])                                # 31 days from one export
        cov.append(["snowflake", "2026-10-31", "2026-10-01"])                               # from after to
        cov.append(["snowflake", "01/10/2026", "2026-10-31"])                               # date not strict
        exc.append(["redshift", "analytics.public.tmp_exports", ""])                        # exception, no reason

    def write(name, header, rows):
        with open(folder / name, "w", newline="", encoding="utf-8") as f:
            w = csv.writer(f)
            w.writerow(header)
            w.writerows(rows)

    write(FILES["register"], COLUMNS["register"], reg)
    write(FILES["coverage"], COLUMNS["coverage"], cov)
    write(FILES["snowflake"], COLUMNS["snowflake"], sf)
    write(FILES["databricks"], [c if c != "timestamp" or plant != "inputs" else "timestmp"
                                for c in COLUMNS["databricks"]], dbx)       # inputs: a misspelled header
    write(FILES["redshift"], COLUMNS["redshift"], rs)
    write(FILES["exceptions"], COLUMNS["exceptions"], exc)
    print(f"Wrote generated sample exports (invented names) to {folder}")
    print("Run: python3 retention_deletion_check.py --dir %s --start 2026-10-01 --end 2026-10-31" % folder)
    return 0


def main():
    ap = argparse.ArgumentParser(description="Check each deletion against a retention register.")
    ap.add_argument("--dir", default="retention-exports", help="folder holding register.csv and the exports")
    ap.add_argument("--start", help="first day of the period, YYYY-MM-DD")
    ap.add_argument("--end", help="last day of the period, YYYY-MM-DD")
    ap.add_argument("--out", default="evidence-out", help="each run writes its own run-<UTC time> folder here")
    ap.add_argument("--redshift-default-schema", default="public",
                    help="schema for a Redshift name written with no schema (your search_path)")
    ap.add_argument("--queries", action="store_true", help="print the export statements and stop")
    ap.add_argument("--make-demo", metavar="FOLDER", help="write generated sample exports and stop")
    ap.add_argument("--plant", choices=["findings", "inputs"], help="with --make-demo: plant faults")
    args = ap.parse_args()
    if args.queries:
        print(QUERIES)
        return 0
    if args.make_demo:
        return make_demo(args.make_demo, args.plant)
    if args.plant:
        print("--plant only works with --make-demo")
        return 2
    if not args.start or not args.end:
        print("Say the period: --start YYYY-MM-DD --end YYYY-MM-DD (your policy sets it).")
        return 2
    return run_check(args)


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