#!/usr/bin/env python3
"""Rebuild who could read or change each financial table on every day of a period, from your own exports, in DuckDB.

    python3 financial_table_access.py --demo                  # clean generated records
    python3 financial_table_access.py --demo --plant-error    # generated records with a short grant, a missing day and late records
    python3 financial_table_access.py --demo --plant-input    # generated records with planted input faults
    python3 financial_table_access.py --demo --plant-header   # generated records with one column renamed in members.csv
    python3 financial_table_access.py                         # your exports: set DATA_DIR and the settings below first
    python3 financial_table_access.py --data <folder>         # your exports in <folder> (instead of editing DATA_DIR)
    python3 financial_table_access.py --strict                # exit 1 if any day is NOT COVERED or the input is incomplete
    add  --out <folder>  to write every file under that folder (default: ./access-out)

It reads up to six CSV files (see COLUMN_MAP): the financial tables you name, the grants each platform keeps (as dated
records, or as one export a day), who is in each role or group, and which login is which person. For every financial
table it prints:
  1. The period's access population: each person, role or group that could read or change the table on any day of the
     period, how (direct, or through which role or group), the first and last day, and the number of days. The rows have
     a row count and a SHA-256.
  2. The changes in the period: every grant and revoke on those tables, and every person added to or removed from a role
     or group that holds access, with who made it where the platform says.
  3. NOT COVERED: the days of the period for which a platform's records cannot rebuild the population (history shorter
     than the period, a day with no export, a platform not exported). It is never read as "nobody had access".
Input checks run first. If one fails, nothing is computed and the run says INCOMPLETE.
The script reads files; it never connects to a platform and never writes to one.

The demo records are GENERATED (fixed, no random numbers): not a company's data, and not a run on Snowflake, Databricks
or Redshift. Their column names come from the platforms' own pages:
  Snowflake   https://docs.snowflake.com/en/sql-reference/account-usage/grants_to_roles
              https://docs.snowflake.com/en/sql-reference/account-usage/grants_to_users
  Databricks  https://docs.databricks.com/aws/en/sql/language-manual/information-schema/table_privileges
  Redshift    https://docs.aws.amazon.com/redshift/latest/dg/r_SVV_RELATION_PRIVILEGES.html
"""
import csv
import datetime as dt
import hashlib
import pathlib
import re
import sys

import duckdb

# ---- CHANGE THESE FOR YOUR OWN DATA ---------------------------------------------------------
PERIOD_START = "2026-07-01"   # first day of the audit period (YYYY-MM-DD)
PERIOD_END = "2026-09-30"     # last day of the audit period (YYYY-MM-DD)
AS_OF = "2026-10-05"          # the day you took the exports (YYYY-MM-DD)
DATA_DIR = None               # folder holding the files named in FILE_NAMES
PLATFORMS = ["snowflake", "databricks", "redshift"]   # the platforms you exported
# How each platform's records come: "history" (dated grants, from Snowflake ACCOUNT_USAGE) or "snapshots" (one export a day).
FORM = {"snowflake": "history", "databricks": "snapshots", "redshift": "snapshots"}
# For a "history" platform: how many days back the platform keeps its records (Snowflake ACCOUNT_USAGE: 365).
HISTORY_DAYS = {"snowflake": 365}
# A holder that means every user of the platform. It is shown as "everyone", not expanded into names.
PUBLIC_HOLDERS = {"snowflake": ["PUBLIC"], "databricks": ["account users"], "redshift": ["public"]}
FILE_NAMES = {"tables": "financial_tables.csv", "grants": "grants.csv", "members": "members.csv",
              "grant_snapshots": "grant_snapshots.csv", "member_snapshots": "member_snapshots.csv", "logins": "logins.csv"}
# For each file: the script's column name (left, do not change) = the column name in your file (right).
# Names are matched without regard to case or surrounding spaces.
COLUMN_MAP = {
    "tables": {"platform": "platform", "table_catalog": "table_catalog", "table_schema": "table_schema",
               "table_name": "table_name"},
    "grants": {"platform": "platform", "holder": "holder", "holder_kind": "holder_kind", "privilege": "privilege",
               "table_catalog": "table_catalog", "table_schema": "table_schema", "table_name": "table_name",
               "start_date": "start_date", "end_date": "end_date", "granted_by": "granted_by"},
    "members": {"platform": "platform", "holder": "holder", "member": "member", "start_date": "start_date",
                "end_date": "end_date", "granted_by": "granted_by"},
    "grant_snapshots": {"platform": "platform", "snapshot_date": "snapshot_date", "holder": "holder",
                        "holder_kind": "holder_kind", "privilege": "privilege", "table_catalog": "table_catalog",
                        "table_schema": "table_schema", "table_name": "table_name"},
    "member_snapshots": {"platform": "platform", "snapshot_date": "snapshot_date", "holder": "holder", "member": "member"},
    "logins": {"platform": "platform", "login": "login", "email": "email"},
}
# Optional columns: read when the file has them. member_kind (USER for a user; anything else is a role, group or share)
# keeps a user apart from a role of the same name, such as Snowflake's per-user roles: a user is never walked as a role.
OPTIONAL_COLUMNS = {"members": {"member_kind": "member_kind"}, "member_snapshots": {"member_kind": "member_kind"}}
# What each privilege lets a holder do. Names are compared in upper case with _ read as a space.
READ_PRIVILEGES = ["SELECT"]
CHANGE_PRIVILEGES = ["INSERT", "UPDATE", "DELETE", "TRUNCATE", "MODIFY", "ALTER", "DROP", "REBUILD"]
BOTH_PRIVILEGES = ["OWNERSHIP", "ALL PRIVILEGES", "ALL"]
# Privileges that do not read or change the data. A privilege on a financial table that is in no list fails an input check.
IGNORED_PRIVILEGES = ["REFERENCES", "USAGE", "BROWSE", "APPLY TAG", "APPLY MASKING POLICY", "APPLY ROW ACCESS POLICY",
                      "MANAGE", "READ METADATA", "EVOLVE SCHEMA",
                      # Unity Catalog privileges on a catalog or schema that do not read or change a table's rows
                      "USE CATALOG", "USE SCHEMA", "CREATE TABLE", "CREATE SCHEMA", "CREATE FUNCTION", "CREATE VOLUME",
                      "CREATE MODEL", "CREATE MATERIALIZED VIEW", "EXECUTE", "READ VOLUME", "WRITE VOLUME", "REFRESH",
                      "EXTERNAL USE SCHEMA"]
# ----------------------------------------------------------------------------------------------

KNOWN_PLATFORMS = ["snowflake", "databricks", "redshift"]
KNOWN_FORMS = ["history", "snapshots"]
KNOWN_KINDS = ["user", "role", "group", "public", "service_account"]
# In member_snapshots.csv, a row whose member is this marker says "this role was exported on this day" and names nobody:
# Snowflake's SHOW GRANTS OF ROLE returns no rows for a role with no members, so the export adds one marker row per role.
MEMBER_MARKER = "-"
OPEN_END = "2999-12-31"
RUN_AT = dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")


def dcol(col):
    """A date read strictly: YYYY-MM-DD and a real calendar day; anything else is NULL and counted by an input check."""
    return (f"CASE WHEN regexp_full_match(trim({col}), '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]') "
            f"THEN try_strptime(trim({col}), '%Y-%m-%d')::DATE END")


def blank(col):
    return f"nullif(trim({col}), '')"


def norm(col):
    return f"lower(trim({col}))"


def pnorm(col):
    return f"upper(replace(trim({col}), '_', ' '))"


def quote_list(items):
    return ", ".join("'" + str(i).replace("'", "''") + "'" for i in items)


def cell(v):
    """A CSV cell a spreadsheet will not run: text starting with = + - @, a tab or a carriage return gets a ' first."""
    if isinstance(v, (dt.date, dt.datetime)):
        return v.strftime("%Y-%m-%d")
    return "'" + v if isinstance(v, str) and v[:1] in ("=", "+", "-", "@", "\t", "\r") else v


def days(a, b):
    d, out = dt.date.fromisoformat(a), []
    while d <= dt.date.fromisoformat(b):
        out.append(d.isoformat())
        d += dt.timedelta(days=1)
    return out


# ---------------------------------------------------------------------------------------------
# Demo records: GENERATED, in the column shapes the three platforms document.
# ---------------------------------------------------------------------------------------------
def generate(folder, plant_error, plant_input, plant_header):
    global PERIOD_START, PERIOD_END, AS_OF, PLATFORMS, FORM, HISTORY_DAYS, PUBLIC_HOLDERS
    PERIOD_START, PERIOD_END, AS_OF = "2026-07-01", "2026-09-30", "2026-10-05"
    PLATFORMS = ["snowflake", "databricks", "redshift"]
    FORM = {"snowflake": "history", "databricks": "snapshots", "redshift": "snapshots"}
    HISTORY_DAYS = {"snowflake": 365}
    PUBLIC_HOLDERS = {"snowflake": ["PUBLIC"], "databricks": ["account users"], "redshift": ["public"]}
    folder.mkdir(parents=True, exist_ok=True)
    em = lambda n: f"{n}.novak@example.com" if n in ("ada", "chloe", "elena", "gwen", "iris", "luca") else f"{n}.reyes@example.com"
    people = ["ada", "ben", "chloe", "dev", "elena", "farid", "gwen", "hugo", "iris", "luca"]
    tables = [["snowflake", "FIN_DB", "LEDGER", "GL_BALANCES"], ["snowflake", "FIN_DB", "LEDGER", "JOURNAL_ENTRIES"],
              ["snowflake", "FIN_DB", "AP", "AP_INVOICES"],
              ["databricks", "main", "finance", "gl_balances"], ["databricks", "main", "finance", "revenue_recognition"],
              ["redshift", "dev", "finance", "ledger_entries"]]
    # Snowflake: dated records in the shape of GRANTS_TO_ROLES and GRANTS_TO_USERS (dates cut from the timestamps)
    grants = []   # platform, holder, holder_kind, privilege, catalog, schema, table, start, end, granted_by
    for t in ("GL_BALANCES", "JOURNAL_ENTRIES"):
        grants.append(["snowflake", "FIN_READER", "role", "SELECT", "FIN_DB", "LEDGER", t, "2025-02-03", "", "SECURITYADMIN"])
    for t, s in (("GL_BALANCES", "LEDGER"), ("JOURNAL_ENTRIES", "LEDGER"), ("AP_INVOICES", "AP")):
        grants.append(["snowflake", "FIN_OWNER", "role", "OWNERSHIP", "FIN_DB", s, t, "2024-01-10", "", "SYSADMIN"])
    grants.append(["snowflake", "FIN_POSTER", "role", "INSERT", "FIN_DB", "LEDGER", "JOURNAL_ENTRIES", "2025-02-03", "", "SECURITYADMIN"])
    grants.append(["snowflake", "FIN_POSTER", "role", "UPDATE", "FIN_DB", "LEDGER", "JOURNAL_ENTRIES", "2025-02-03", "", "SECURITYADMIN"])
    grants.append(["snowflake", "AP_ANALYST", "role", "SELECT", "FIN_DB", "AP", "AP_INVOICES", "2025-03-04", "", "SECURITYADMIN"])
    grants.append(["snowflake", "FIN_READER", "role", "SELECT", "FIN_DB", "LEDGER", "GL_BALANCES", "2024-02-01", "2025-02-02", "SECURITYADMIN"])  # an older grant, replaced
    grants.append(["snowflake", "FIN_DB_OLD_ROLE", "role", "SELECT", "FIN_DB", "LEDGER", "GL_BALANCES", "2024-03-01", "2026-06-20", "SECURITYADMIN"])  # revoked before the period
    grants.append(["snowflake", "ANALYST_SANDBOX", "role", "SELECT", "FIN_DB", "SANDBOX", "SCRATCH", "2025-01-01", "", "SYSADMIN"])  # not a financial table
    grants.append(["snowflake", "REPORTING", "role", "SELECT", "FIN_DB", "AP", "AP_INVOICES", "2026-08-03", "2026-08-31", "SECURITYADMIN"])  # a month, granted and revoked
    members = []  # platform, holder, member, start, end, granted_by
    for u, s, e in (("ADA", "2024-01-10", ""), ("BEN", "2025-02-03", ""), ("CHLOE", "2025-06-16", "2026-08-20"),
                    ("ELENA", "2025-09-01", "")):
        members.append(["snowflake", "FIN_OWNER" if u == "ADA" else "FIN_READER", u, s, e, "SECURITYADMIN"])
    members += [["snowflake", "FIN_READER", "REPORTING", "2025-02-03", "", "SECURITYADMIN"],
                ["snowflake", "REPORTING", "DEV", "2025-04-07", "", "SECURITYADMIN"],
                ["snowflake", "REPORTING", "IRIS", "2026-07-22", "", "SECURITYADMIN"],     # joined the role inside the period
                ["snowflake", "FIN_POSTER", "FARID", "2025-02-03", "", "SECURITYADMIN"],
                ["snowflake", "AP_ANALYST", "LUCA", "2025-03-04", "2026-07-10", "SECURITYADMIN"],  # left the role inside the period
                ["snowflake", "FIN_OWNER", "SYSADMIN", "2024-01-10", "", "SYSADMIN"],
                ["snowflake", "SYSADMIN", "HUGO", "2024-01-10", "", "ACCOUNTADMIN"],
                ["snowflake", "FIN_DB_OLD_ROLE", "GWEN", "2024-03-01", "", "SECURITYADMIN"]]
    logins = [["snowflake", u.upper(), em(u)] for u in people] + [["databricks", em(u), em(u)] for u in people] \
        + [["redshift", u, em(u)] for u in people]
    # Databricks and Redshift: one export a day
    snap_days = days("2026-07-01", "2026-09-30")
    gsnap, msnap = [], []   # platform, day, holder, kind, privilege, catalog, schema, table  /  platform, day, holder, member
    def on(day, a, b):
        return a <= day <= b
    for d in snap_days:
        for t in ("gl_balances", "revenue_recognition"):
            gsnap.append(["databricks", d, "finance-stewards" if t == "revenue_recognition" else "ada.novak@example.com", "", "OWNERSHIP",
                          "main", "finance", t])
        gsnap.append(["databricks", d, "finance-readers", "", "SELECT", "main", "finance", ""])     # on the schema: inherited
        gsnap.append(["databricks", d, "sp-etl-nightly", "", "MODIFY", "main", "finance", "gl_balances"])
        gsnap.append(["databricks", d, "elena.novak@example.com", "", "SELECT", "main", "finance", "revenue_recognition"])
        for m in ("ben.reyes@example.com", "farid.reyes@example.com"):
            msnap.append(["databricks", d, "finance-readers", m])
        if on(d, "2026-07-01", "2026-08-05"):
            msnap.append(["databricks", d, "finance-readers", "hugo.reyes@example.com"])     # left the group 2026-08-06
        for m in ("ada.novak@example.com", "iris.novak@example.com"):
            msnap.append(["databricks", d, "finance-stewards", m])
        gsnap.append(["redshift", d, "ben", "user", "OWNERSHIP", "dev", "finance", "ledger_entries"])
        gsnap.append(["redshift", d, "auditors_ro", "group", "SELECT", "dev", "finance", "ledger_entries"])
        gsnap.append(["redshift", d, "etl_loader", "user", "INSERT", "dev", "finance", "ledger_entries"])
        gsnap.append(["redshift", d, "ada", "user", "SELECT", "dev", "finance", "ledger_entries"])
        gsnap.append(["redshift", d, "analyst_sandbox", "user", "SELECT", "dev", "sandbox", "scratch"])
        for m in ("iris", "luca"):
            msnap.append(["redshift", d, "auditors_ro", m])
    if plant_error:
        if any(r[2] == "GWEN" and r[1] == "FIN_POSTER" for r in members) or any(r[2] == "account users" for r in gsnap):
            sys.exit("A planted value already occurs in the clean records: stopping.")
        members.append(["snowflake", "FIN_POSTER", "GWEN", "2026-08-12", "2026-08-20", "SECURITYADMIN"])   # nine days
        for d in days("2026-09-10", "2026-09-14"):                                                            # five days
            gsnap.append(["databricks", d, "account users", "", "SELECT", "main", "finance", "gl_balances"])
        before = len(gsnap) + len(msnap)
        gsnap[:] = [r for r in gsnap if not (r[0] == "redshift" and (r[1] == "2026-08-14" or r[1] < "2026-07-10"))]
        msnap[:] = [r for r in msnap if not (r[0] == "redshift" and (r[1] == "2026-08-14" or r[1] < "2026-07-10"))]
        if before - len(gsnap) - len(msnap) != 70:     # ten days of seven Redshift rows each
            sys.exit("Could not plant a fault: the Redshift days were not removed.")
    if plant_input:
        grants[0][7] = "08/12/26"                                              # not a date
        members.append(["snowflake", "FIN_POSTER", "DEV", "2026-09-01", "2026-08-01", "SECURITYADMIN"])    # ends before it starts
        grants.append(["oracle", "x", "role", "SELECT", "x", "y", "z", "2026-01-01", "", ""])             # a platform not in PLATFORMS
        grants.append(["snowflake", "FIN_READER", "role", "FLY", "FIN_DB", "LEDGER", "GL_BALANCES", "2026-01-01", "", ""])  # unknown privilege
        grants.append(["snowflake", "", "role", "SELECT", "FIN_DB", "LEDGER", "GL_BALANCES", "2026-01-01", "", ""])         # blank holder
        grants.append(["snowflake", "FIN_READER", "role", "SELECT", "FIN_DB", "", "GL_BALANCES", "2026-01-01", "", ""])      # a table with no schema
        tables.append(list(tables[0]))                                          # the same financial table twice
        tables.append(["snowflake", "FIN_DB", "LEDGER", "NO_SUCH_TABLE"])       # no grant record mentions it
        logins.append(["snowflake", "ADA", "ada.other@example.com"])           # one login, two people
        gsnap.append(["databricks", "2026-10-09", "ada.novak@example.com", "", "SELECT", "main", "finance", "gl_balances"])   # after AS_OF
        gsnap.append(["databricks", "2026-07-01", "ada.novak@example.com", "bot", "SELECT", "main", "finance", "gl_balances"])  # a kind not known
    tabs = {"tables": tables, "grants": grants, "members": members, "grant_snapshots": gsnap,
            "member_snapshots": msnap, "logins": logins}
    for name, rows in tabs.items():
        cols = list(COLUMN_MAP[name])
        with open(folder / FILE_NAMES[name], "w", newline="", encoding="utf-8") as f:
            w = csv.writer(f)
            w.writerow(cols)
            w.writerows(rows)
    if plant_header:
        p = folder / FILE_NAMES["members"]
        p.write_text(p.read_text(encoding="utf-8").replace("platform,holder,member", "platform,role,member", 1), encoding="utf-8")


# ---------------------------------------------------------------------------------------------
# ENGINE
# ---------------------------------------------------------------------------------------------
def main():
    args = sys.argv[1:]
    global DATA_DIR
    out = pathlib.Path(args[args.index("--out") + 1]) if "--out" in args else pathlib.Path("access-out")
    if "--data" in args:
        DATA_DIR = args[args.index("--data") + 1]
    plant_error, plant_input, plant_header = "--plant-error" in args, "--plant-input" in args, "--plant-header" in args
    suffix = "-planted-error" if plant_error else "-planted-input" if plant_input else "-planted-header" if plant_header else ""
    if "--demo" in args:
        gen = out / f"generated{suffix}"
        generate(gen, plant_error, plant_input, plant_header)
        DATA_DIR = str(gen)
        print(f"Generated records written to {gen}/ (not a company's data).")
    if DATA_DIR is None:
        print("Set DATA_DIR at the top of the script, pass --data <folder> (or run with --demo).")
        return 2
    folder = pathlib.Path(DATA_DIR)
    out.mkdir(parents=True, exist_ok=True)
    problems_found = []

    def check(name, bad, detail=""):
        print(f"  [{'FAIL' if bad else 'ok'}] {name}" + (f": {detail}" if bad and detail else ""))
        if bad:
            problems_found.append(name)

    con = duckdb.connect()
    print(f"\nRun at {RUN_AT}. Exports taken {AS_OF}. Period {PERIOD_START} to {PERIOD_END}.")
    print("\nInput checks")
    def real_date(v):
        try:
            return isinstance(v, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", v) is not None and bool(dt.date.fromisoformat(v))
        except ValueError:
            return False
    bad = [n for n, v in (("PERIOD_START", PERIOD_START), ("PERIOD_END", PERIOD_END), ("AS_OF", AS_OF)) if not real_date(v)]
    check("PERIOD_START, PERIOD_END and AS_OF are real dates written YYYY-MM-DD", bad, ", ".join(bad) + " not a date")
    check("the period ends on or after the day it starts", not bad and PERIOD_END < PERIOD_START,
          f"{PERIOD_END} is before {PERIOD_START}")
    bad_plat = not isinstance(PLATFORMS, list) or not PLATFORMS or [p for p in PLATFORMS if p not in KNOWN_PLATFORMS]
    check("PLATFORMS lists the platforms you exported, from snowflake, databricks, redshift", bad_plat, f"PLATFORMS is {PLATFORMS!r}")
    if bad or bad_plat:
        return 2
    bad_form = [p for p in PLATFORMS if FORM.get(p) not in KNOWN_FORMS]
    check("FORM names history or snapshots for every platform in PLATFORMS", bad_form, "no usable FORM for " + ", ".join(bad_form))
    bad_days = [p for p in PLATFORMS if FORM.get(p) == "history" and not (isinstance(HISTORY_DAYS.get(p), int) and HISTORY_DAYS[p] > 0)]
    check("HISTORY_DAYS gives a number of days for every history platform", bad_days, "no HISTORY_DAYS for " + ", ".join(bad_days))
    if bad_form or bad_days:
        return 2
    need = {"tables": True, "logins": True,
            "grants": "history" in FORM.values(), "members": "history" in [FORM[p] for p in PLATFORMS],
            "grant_snapshots": "snapshots" in [FORM[p] for p in PLATFORMS],
            "member_snapshots": "snapshots" in [FORM[p] for p in PLATFORMS]}
    need["grants"] = "history" in [FORM[p] for p in PLATFORMS]

    # --- load every file; a missing file or column is a failure, the table is then empty
    bad_files = []
    for name, cmap in COLUMN_MAP.items():
        path = folder / FILE_NAMES[name]
        have = []
        if path.is_file():
            try:
                have = [d[0] for d in con.execute("SELECT * FROM read_csv(?, header=true, all_varchar=true) LIMIT 0",
                                                  [str(path)]).description]
            except duckdb.Error as e:
                bad_files.append(f"{FILE_NAMES[name]} cannot be read ({str(e).splitlines()[0][:80]})")
        lowered = {h.strip().lower(): h for h in have}
        missing = [theirs for theirs in cmap.values() if theirs.strip().lower() not in lowered]
        if not path.is_file():
            if need[name]:
                bad_files.append(f"{FILE_NAMES[name]} is missing")
        elif have and missing:
            bad_files.append(f"{FILE_NAMES[name]} has no column {', '.join(missing)}")
        opt = OPTIONAL_COLUMNS.get(name, {})
        if path.is_file() and have and not missing:
            sel = ", ".join([f'"{lowered[theirs.strip().lower()]}" AS "{mine}"' for mine, theirs in cmap.items()] +
                            [(f'"{lowered[theirs.strip().lower()]}"' if theirs.strip().lower() in lowered else "''") + f' AS "{mine}"'
                             for mine, theirs in opt.items()])
            con.execute(f"CREATE TABLE t_{name} AS SELECT {sel} FROM read_csv(?, header=true, all_varchar=true)", [str(path)])
        else:
            con.execute(f"CREATE TABLE t_{name} ({', '.join(chr(34) + c + chr(34) + ' VARCHAR' for c in [*cmap, *opt])})")
    check("every file you need is there with every column the script reads", bad_files, "; ".join(bad_files))
    counts = {n: con.execute(f"SELECT count(*) FROM t_{n}").fetchone()[0] for n in COLUMN_MAP}
    for n, c in counts.items():
        if need[n] or c:
            print(f"      read {c} rows from {FILE_NAMES[n]}")
    q = lambda sql: con.execute(sql).fetchone()[0]
    if bad_files:
        print(f"\nINCOMPLETE: {len(problems_found)} input check(s) failed, so nothing is computed. Fix the input.")
        return 2

    con.execute("CREATE TEMP TABLE t_form AS SELECT * FROM (VALUES " + ", ".join(
        f"('{p}', '{FORM[p]}')" for p in PLATFORMS) + ") v(platform, form)")
    asof, ps, pe = f"DATE '{AS_OF}'", f"DATE '{PERIOD_START}'", f"DATE '{PERIOD_END}'"
    dated = {"grants": ["start_date", "end_date"], "members": ["start_date", "end_date"],
             "grant_snapshots": ["snapshot_date"], "member_snapshots": ["snapshot_date"]}
    n = sum(q(f"SELECT count(*) FROM t_{t} WHERE ({dcol(c)}) IS NULL"
              + (f" AND {blank(c)} IS NOT NULL" if c == "end_date" else ""))
            for t, cols in dated.items() for c in cols)
    check("every start date, end date and snapshot date is a real date written YYYY-MM-DD (an end date may be empty)", n,
          f"{n} value(s) empty where a date is needed, or not written YYYY-MM-DD")
    n = sum(q(f"SELECT count(*) FROM t_{t} WHERE ({dcol('end_date')}) < ({dcol('start_date')})") for t in ("grants", "members"))
    check("no record ends before it starts", n, f"{n} record(s) with an end date before the start date")
    n = sum(q(f"SELECT count(*) FROM t_{t} WHERE ({dcol('snapshot_date')}) > {asof}") for t in ("grant_snapshots", "member_snapshots")) \
        + sum(q(f"SELECT count(*) FROM t_{t} WHERE ({dcol('start_date')}) > {asof}") for t in ("grants", "members"))
    check("no snapshot date or start date is after AS_OF", n, f"{n} record(s) dated after {AS_OF}")
    odd = []
    for tbl in COLUMN_MAP:
        k = q(f"SELECT count(*) FROM t_{tbl} WHERE {norm('platform')} NOT IN ({quote_list(PLATFORMS)})")
        if k:
            odd.append(f"{FILE_NAMES[tbl]}: {k}")
    check("every platform value is one of PLATFORMS", odd, "unlisted platforms: " + ", ".join(odd))
    n = sum(q(f"SELECT count(*) FROM t_{t} WHERE {blank('holder_kind')} IS NOT NULL AND {norm('holder_kind')} NOT IN ({quote_list(KNOWN_KINDS)})")
            for t in ("grants", "grant_snapshots"))
    check("every holder_kind is empty or one of user, role, group, public, service_account", n, f"{n} row(s) with another kind")
    key_cols = {"tables": ["platform", "table_catalog", "table_schema", "table_name"],
                "grants": ["platform", "holder", "privilege", "table_catalog"],
                "members": ["platform", "holder", "member"],
                "grant_snapshots": ["platform", "holder", "privilege", "table_catalog"],
                "member_snapshots": ["platform", "holder", "member"], "logins": ["platform", "login", "email"]}
    n = sum(q(f"SELECT count(*) FROM t_{t} WHERE " + " OR ".join(f"{blank(c)} IS NULL" for c in cols)) for t, cols in key_cols.items())
    check("no key column is blank (platform, holder, member, privilege, catalog, table, login, email)", n, f"{n} row(s) with a blank key column")
    n = sum(q(f"SELECT count(*) FROM t_{t} WHERE {blank('table_schema')} IS NULL AND {blank('table_name')} IS NOT NULL")
            for t in ("grants", "grant_snapshots"))
    check("a grant that names a table also names its schema", n, f"{n} row(s) with a table and no schema")
    tk = "lower(trim(platform)), lower(trim(table_catalog)), lower(trim(table_schema)), lower(trim(table_name))"
    n = q(f"SELECT count(*) - count(DISTINCT ({tk})) FROM t_tables")
    check("every financial table is listed once", n, f"{n} row(s) repeat a table")
    check("at least one financial table is listed", counts["tables"] == 0, f"{FILE_NAMES['tables']} has no rows: you set the financial tables")
    n = q(f"SELECT count(*) - count(DISTINCT ({norm('platform')}, {norm('login')})) FROM t_logins")
    check("each login is listed once per platform", n, f"{n} login row(s) repeat a login")

    # --- the grants, normalized: one row per holder, privilege and object, with the days it was held
    levels = ([(p, "READ") for p in READ_PRIVILEGES] + [(p, "CHANGE") for p in CHANGE_PRIVILEGES]
              + [(p, "BOTH") for p in BOTH_PRIVILEGES] + [(p, "NONE") for p in IGNORED_PRIVILEGES])
    con.execute("CREATE TEMP TABLE lvl AS SELECT * FROM (VALUES " + ", ".join(
        f"('{p.upper().replace('_', ' ')}', '{lv}')" for p, lv in levels) + ") v(priv, level)")
    con.execute(f"""CREATE TEMP TABLE kindmap AS SELECT * FROM (VALUES ('user','login'),('service_account','login'),
        ('role','role or group'),('group','role or group'),('public','everyone')) v(k, kind)""")
    obj = lambda: (f"{norm('table_catalog')} AS cat, coalesce({norm('table_schema')}, '') AS sch, "
                   f"coalesce({norm('table_name')}, '') AS tbl")
    # history platforms: dated records
    con.execute(f"""CREATE TEMP TABLE gh AS
        SELECT {norm('g.platform')} AS platform, {norm('g.holder')} AS holder, trim(g.holder) AS shown,
               {norm('g.holder_kind')} AS kind, {pnorm('g.privilege')} AS priv,
               {norm('g.table_catalog')} AS cat, coalesce({blank('g.table_schema')}, '') AS sch0,
               coalesce({blank('g.table_name')}, '') AS tbl0, ({dcol('g.start_date')}) AS s, ({dcol('g.end_date')}) AS e,
               coalesce({blank('g.granted_by')}, '') AS granted_by
        FROM t_grants g JOIN t_form f ON f.platform = {norm('g.platform')} AND f.form = 'history'
        WHERE ({dcol('g.start_date')}) IS NOT NULL""")
    con.execute("""CREATE TEMP TABLE gh2 AS SELECT platform, holder, shown, kind, priv, cat, lower(sch0) AS sch, lower(tbl0) AS tbl,
        s AS start_d, coalesce(e, DATE '2999-12-31') AS end_d, true AS start_is_change,
        CASE WHEN e IS NULL THEN NULL ELSE e END AS end_event, granted_by, 'dated records' AS basis FROM gh""")
    # snapshot platforms: one row per day becomes the run of days the row was seen
    con.execute(f"""CREATE TEMP TABLE gs AS
        SELECT {norm('g.platform')} AS platform, {norm('g.holder')} AS holder, trim(g.holder) AS shown,
               {norm('g.holder_kind')} AS kind, {pnorm('g.privilege')} AS priv, {norm('g.table_catalog')} AS cat,
               coalesce({norm('g.table_schema')}, '') AS sch, coalesce({norm('g.table_name')}, '') AS tbl,
               ({dcol('g.snapshot_date')}) AS d
        FROM t_grant_snapshots g JOIN t_form f ON f.platform = {norm('g.platform')} AND f.form = 'snapshots'
        WHERE ({dcol('g.snapshot_date')}) IS NOT NULL AND ({dcol('g.snapshot_date')}) <= {asof}""")
    con.execute("""CREATE TEMP TABLE sd AS SELECT platform, d, dense_rank() OVER (PARTITION BY platform ORDER BY d) AS rk
                   FROM (SELECT DISTINCT platform, d FROM gs)""")
    # the export unit each snapshot row came from: one SHOW GRANTS per Snowflake table, one export per Databricks
    # catalog or Redshift database; a day counts for a table only when the table's own unit was exported that day
    unit = lambda a: (f"CASE WHEN {a}platform = 'snowflake' THEN {a}cat || '.' || {a}sch || '.' || {a}tbl "
                      f"ELSE {a}cat END")
    con.execute(f"""CREATE TEMP TABLE ud AS SELECT platform, u, d, dense_rank() OVER (PARTITION BY platform, u ORDER BY d) AS rk
                   FROM (SELECT DISTINCT platform, {unit('')} AS u, d FROM gs)""")
    con.execute("""CREATE TEMP TABLE gs2 AS
        WITH x AS (SELECT DISTINCT g.platform, g.holder, coalesce(g.kind, '') AS kind, g.priv, g.cat, g.sch, g.tbl, g.d, s.rk, s.u
                   FROM gs g
                   JOIN ud s ON s.platform = g.platform AND s.u = UNIT_G AND s.d = g.d),
             y AS (SELECT *, rk - row_number() OVER (PARTITION BY platform, holder, kind, priv, cat, sch, tbl ORDER BY rk) AS grp
                   FROM x),
             z AS (SELECT platform, u, holder, kind, priv, cat, sch, tbl, min(d) AS start_d, max(d) AS end_d, max(rk) AS maxrk
                   FROM y GROUP BY platform, u, holder, kind, priv, cat, sch, tbl, grp)
        SELECT z.platform, z.holder, z.holder AS shown, z.kind, z.priv, z.cat, z.sch, z.tbl, z.start_d, z.end_d,
               z.start_d > (SELECT min(d) FROM ud WHERE ud.platform = z.platform AND ud.u = z.u) AS start_is_change,
               (SELECT min(d) FROM ud WHERE ud.platform = z.platform AND ud.u = z.u AND ud.rk = z.maxrk + 1) AS end_event,
               '' AS granted_by, 'daily exports' AS basis FROM z""".replace("UNIT_G", unit("g.")))
    con.execute("""CREATE TEMP TABLE shown AS
        SELECT platform, holder, min(shown) AS shown, coalesce(max(nullif(kind, '')), '') AS kind
        FROM (SELECT platform, holder, shown, kind FROM gh UNION ALL SELECT platform, holder, shown, kind FROM gs)
        GROUP BY platform, holder""")
    con.execute("""CREATE TEMP TABLE priv AS
        SELECT row_number() OVER () AS id, p.platform, p.holder, coalesce(lower(p.kind), '') AS hkind, p.priv, l.level, p.cat,
               p.sch, p.tbl, p.start_d, p.end_d, p.start_is_change, p.end_event, p.granted_by, p.basis
        FROM (SELECT * FROM gh2 UNION ALL SELECT * FROM gs2) p LEFT JOIN lvl l ON l.priv = p.priv""")
    # --- who is in each role or group
    con.execute(f"""CREATE TEMP TABLE eh AS
        SELECT {norm('m.platform')} AS platform, {norm('m.holder')} AS holder, trim(m.holder) AS holder_shown,
               {norm('m.member')} AS member, trim(m.member) AS member_shown, ({dcol('m.start_date')}) AS start_d,
               coalesce(({dcol('m.end_date')}), DATE '2999-12-31') AS end_d, true AS start_is_change,
               ({dcol('m.end_date')}) AS end_event, coalesce({blank('m.granted_by')}, '') AS granted_by, 'dated records' AS basis,
               coalesce(lower(trim(m.member_kind)), '') AS member_kind
        FROM t_members m JOIN t_form f ON f.platform = {norm('m.platform')} AND f.form = 'history'
        WHERE ({dcol('m.start_date')}) IS NOT NULL""")
    con.execute(f"""CREATE TEMP TABLE ms AS
        SELECT {norm('m.platform')} AS platform, {norm('m.holder')} AS holder, trim(m.holder) AS holder_shown,
               {norm('m.member')} AS member, trim(m.member) AS member_shown, ({dcol('m.snapshot_date')}) AS d,
               coalesce(lower(trim(m.member_kind)), '') AS member_kind
        FROM t_member_snapshots m JOIN t_form f ON f.platform = {norm('m.platform')} AND f.form = 'snapshots'
        WHERE ({dcol('m.snapshot_date')}) IS NOT NULL AND ({dcol('m.snapshot_date')}) <= {asof}""")
    # member exports: one SHOW GRANTS OF ROLE per Snowflake role, one platform-wide export on Databricks and Redshift,
    # so a Snowflake role's runs are counted on that role's own export days
    munit = lambda a: f"CASE WHEN {a}platform = 'snowflake' THEN {a}holder ELSE '' END"
    con.execute(f"""CREATE TEMP TABLE md AS SELECT platform, u, d, dense_rank() OVER (PARTITION BY platform, u ORDER BY d) AS rk
                   FROM (SELECT DISTINCT platform, {munit('')} AS u, d FROM ms)""")
    con.execute("""CREATE TEMP TABLE es AS
        WITH x AS (SELECT DISTINCT m.platform, m.holder, m.member, m.member_kind, m.d, s.rk, s.u FROM ms m
                   JOIN md s ON s.platform = m.platform AND s.u = MUNIT_M AND s.d = m.d WHERE m.member <> 'MARKER'),
             y AS (SELECT *, rk - row_number() OVER (PARTITION BY platform, holder, member, member_kind ORDER BY rk) AS grp
                   FROM x),
             z AS (SELECT platform, u, holder, member, member_kind, min(d) AS start_d, max(d) AS end_d, max(rk) AS maxrk
                   FROM y GROUP BY platform, u, holder, member, member_kind, grp)
        SELECT z.platform, z.holder, z.holder AS holder_shown, z.member, z.member AS member_shown, z.start_d, z.end_d,
               z.start_d > (SELECT min(d) FROM md WHERE md.platform = z.platform AND md.u = z.u) AS start_is_change,
               (SELECT min(d) FROM md WHERE md.platform = z.platform AND md.u = z.u AND md.rk = z.maxrk + 1) AS end_event,
               '' AS granted_by, 'daily exports' AS basis, z.member_kind FROM z""".replace("MUNIT_M", munit("m.")).replace("MARKER", MEMBER_MARKER))
    con.execute("CREATE TEMP TABLE edge AS SELECT * FROM eh UNION ALL SELECT * FROM es")
    con.execute(f"""CREATE TEMP TABLE fin AS
        SELECT row_number() OVER (ORDER BY {tk}) AS fno, {norm('platform')} AS platform, {norm('table_catalog')} AS cat,
               {norm('table_schema')} AS sch, {norm('table_name')} AS tbl,
               trim(table_catalog) || '.' || trim(table_schema) || '.' || trim(table_name) AS shown_name
        FROM (SELECT DISTINCT * FROM t_tables)""")
    # grants that reach a financial table: on the table, or on its schema or catalog (Unity Catalog inherits downward)
    con.execute("""CREATE TEMP TABLE pm AS
        SELECT p.id AS pid, f.fno, p.platform, p.holder, p.hkind, p.priv, p.level, p.start_d, p.end_d, p.start_is_change,
               p.end_event, p.granted_by, p.basis
        FROM priv p JOIN fin f ON f.platform = p.platform AND f.cat = p.cat
          AND (p.sch = '' OR p.sch = f.sch) AND (p.tbl = '' OR (p.sch <> '' AND p.tbl = f.tbl))""")
    n = q("SELECT count(*) FROM fin f WHERE NOT EXISTS (SELECT 1 FROM pm WHERE pm.fno = f.fno)")
    check("every financial table appears in at least one grant record", n,
          f"{n} financial table(s) no grant record names: a stale list, a wrong catalog or schema, or an export that cannot see them")
    unknown = q("SELECT count(*) FROM pm WHERE level IS NULL")
    check("every privilege on a financial table is one the script knows (READ, CHANGE, BOTH or IGNORED lists)", unknown,
          f"{unknown} grant(s) on a financial table with a privilege in none of the four lists at the top of the script")
    if problems_found:
        print(f"\nINCOMPLETE: {len(problems_found)} input check(s) failed, so nothing is computed. Fix the input.")
        return 2

    # --- which days each platform's records can rebuild
    con.execute(f"""CREATE TEMP TABLE pdays AS SELECT unnest(generate_series({ps}, {pe}, INTERVAL 1 DAY))::DATE AS day""")
    plist = ", ".join(f"('{p}')" for p in PLATFORMS)
    con.execute(f"CREATE TEMP TABLE plat AS SELECT * FROM (VALUES {plist}) v(platform)")
    con.execute("""CREATE TEMP TABLE has_members AS SELECT DISTINCT platform FROM ms""")
    floor = {p: (dt.date.fromisoformat(AS_OF) - dt.timedelta(days=HISTORY_DAYS[p] - 1)).isoformat()
             for p in PLATFORMS if FORM[p] == "history"}
    con.execute("CREATE TEMP TABLE floors AS SELECT * FROM (VALUES " + ", ".join(
        f"('{p}', DATE '{d}')" for p, d in floor.items()) + ") v(platform, floor_d)" if floor else
        "CREATE TEMP TABLE floors(platform VARCHAR, floor_d DATE)")
    pub0 = ", ".join(f"('{pp.lower()}', '{hh.lower()}')" for pp, hs in PUBLIC_HOLDERS.items() for hh in hs) or "('', '')"
    con.execute(f"""CREATE TEMP TABLE daystate AS
        SELECT p.platform, d.day, t.form,
          CASE WHEN d.day > {asof} THEN 'after the day the exports were taken'
               WHEN t.form = 'history' AND d.day < fl.floor_d THEN 'before the platform''s records start (' || strftime(fl.floor_d, '%Y-%m-%d') || ')'
               WHEN t.form = 'history' AND NOT EXISTS (SELECT 1 FROM gh WHERE gh.platform = p.platform) THEN 'no grant records exported'
               WHEN t.form = 'snapshots' AND NOT EXISTS (SELECT 1 FROM sd WHERE sd.platform = p.platform) THEN 'no snapshots exported'
               WHEN t.form = 'snapshots' AND d.day < (SELECT min(s2.d) FROM sd s2 WHERE s2.platform = p.platform)
                    THEN 'before the first daily export (' || strftime((SELECT min(s2.d) FROM sd s2 WHERE s2.platform = p.platform), '%Y-%m-%d') || ')'
               WHEN t.form = 'snapshots' AND NOT EXISTS (SELECT 1 FROM sd WHERE sd.platform = p.platform AND sd.d = d.day)
                    THEN 'no export for this day'
               WHEN t.form = 'snapshots' AND NOT EXISTS (SELECT 1 FROM has_members h WHERE h.platform = p.platform)
                    AND EXISTS (SELECT 1 FROM pm JOIN gs ON gs.platform = pm.platform AND gs.holder = pm.holder
                                WHERE pm.platform = p.platform AND pm.level <> 'NONE'
                                AND (gs.kind IN ('role', 'group') OR (coalesce(gs.kind, '') = '' AND NOT EXISTS (
                                     SELECT 1 FROM t_logins l WHERE lower(trim(l.platform)) = gs.platform
                                     AND lower(trim(l.login)) = gs.holder)))
                                AND pm.start_d <= {pe} AND pm.end_d >= {ps}
                                AND (gs.platform, gs.holder) NOT IN (SELECT * FROM (VALUES {pub0})))
                    THEN 'no membership export (the grants name roles or groups, or holders with no holder_kind that are not in logins.csv)'
               WHEN t.form = 'snapshots' AND EXISTS (SELECT 1 FROM has_members h WHERE h.platform = p.platform)
                    AND NOT EXISTS (SELECT 1 FROM ms WHERE ms.platform = p.platform AND ms.d = d.day)
                    THEN 'no membership export for this day'
               ELSE NULL END AS reason
        FROM plat p CROSS JOIN pdays d JOIN t_form t ON t.platform = p.platform
        LEFT JOIN floors fl ON fl.platform = p.platform""")
    # --- who could reach each financial table, on which days, through which roles
    # via = the roles between the person and the grant, nearest the person first; empty for the holder itself
    con.execute("""CREATE TEMP TABLE reach AS
        WITH RECURSIVE r(pid, fno, platform, node, node_shown, s, e, depth, via, is_user, node_kind) AS (
            SELECT pm.pid, pm.fno, pm.platform, pm.holder, coalesce(h.shown, pm.holder), pm.start_d, pm.end_d, 0, '',
                   pm.hkind IN ('user', 'service_account'), pm.hkind
            FROM pm LEFT JOIN shown h ON h.platform = pm.platform AND h.holder = pm.holder
            WHERE pm.level IS NOT NULL AND pm.level <> 'NONE'
            UNION ALL
            SELECT r.pid, r.fno, r.platform, ed.member, ed.member_shown, greatest(r.s, ed.start_d), least(r.e, ed.end_d),
                   r.depth + 1, CASE WHEN r.depth = 0 THEN r.node_shown ELSE r.node_shown || ' > ' || r.via END,
                   ed.member_kind = 'user', ed.member_kind
            FROM r JOIN edge ed ON ed.platform = r.platform AND ed.holder = r.node
            WHERE greatest(r.s, ed.start_d) <= least(r.e, ed.end_d) AND r.depth < 10 AND NOT r.is_user)
        SELECT * FROM r""")

    # a day can also be NOT COVERED for one table only: its own export unit missed that day (snapshots), or the
    # history holds no grant on the table before a day, which is what a dropped or replaced table looks like
    # (GRANTS_TO_ROLES "does not contain grants on dropped objects"; CREATE OR REPLACE makes a new object)
    # roles and groups that hold privileges (by the export's holder_kind or because members name them), less "everyone"
    con.execute(f"""CREATE TEMP TABLE rg AS SELECT DISTINCT platform, holder FROM (
        SELECT platform, holder FROM gh WHERE kind IN ('role', 'group') UNION ALL
        SELECT platform, holder FROM gs WHERE kind IN ('role', 'group') UNION ALL SELECT platform, holder FROM edge)
        WHERE (platform, holder) NOT IN (SELECT * FROM (VALUES {pub0}))""")
    # Snowflake nodes that need their own member export: the roles above, every role that was exported (so a nested
    # role is checked too) and every share reached through a database role
    con.execute(f"""CREATE TEMP TABLE rl AS SELECT DISTINCT platform, holder FROM (
        SELECT platform, holder FROM rg UNION ALL SELECT platform, holder FROM ms
        UNION ALL SELECT platform, node FROM reach WHERE node LIKE 'share %')
        WHERE platform = 'snowflake' AND (platform, holder) NOT IN (SELECT * FROM (VALUES {pub0}))""")
    # Redshift: a role (holder_kind role) is not a group, so the pg_group member export never lists its people
    con.execute(f"""CREATE TEMP TABLE rmiss AS
        SELECT DISTINCT r.fno, d.day, r.node, r.platform FROM reach r
        JOIN (SELECT DISTINCT day FROM daystate) d ON d.day BETWEEN r.s AND r.e
        WHERE NOT r.is_user
          AND ((r.platform = 'snowflake' AND EXISTS (SELECT 1 FROM rl WHERE rl.platform = r.platform AND rl.holder = r.node))
               OR (r.platform = 'redshift' AND r.node_kind = 'role' AND (r.platform, r.node) NOT IN (SELECT * FROM (VALUES {pub0}))))
          AND NOT EXISTS (SELECT 1 FROM ms WHERE ms.platform = r.platform AND ms.holder = r.node AND ms.d = d.day)""")
    con.execute(f"""CREATE TEMP TABLE tday AS
        SELECT f.platform, f.fno, ds.day,
          CASE WHEN ds.reason IS NOT NULL THEN NULL
               WHEN ds.form = 'snapshots' AND ds.day < (SELECT min(ud.d) FROM ud WHERE ud.platform = f.platform AND ud.u = {unit('f.')})
                    THEN 'before the first export of this ' || CASE f.platform WHEN 'snowflake' THEN 'table' WHEN 'redshift' THEN 'database' ELSE 'catalog' END
               WHEN ds.form = 'snapshots' AND NOT EXISTS (SELECT 1 FROM ud WHERE ud.platform = f.platform AND ud.u = {unit('f.')} AND ud.d = ds.day)
                    THEN 'no export of this ' || CASE f.platform WHEN 'snowflake' THEN 'table' WHEN 'redshift' THEN 'database' ELSE 'catalog' END || ' for this day'
               WHEN f.platform IN ('snowflake', 'redshift') AND ds.form = 'snapshots'
                    AND EXISTS (SELECT 1 FROM rmiss WHERE rmiss.fno = f.fno AND rmiss.day = ds.day)
                    THEN 'no member rows for ' || (SELECT string_agg(rmiss.node, ', ' ORDER BY rmiss.node) FROM rmiss
                         WHERE rmiss.fno = f.fno AND rmiss.day = ds.day)
                         || CASE f.platform WHEN 'redshift' THEN ' on this day: a Redshift role''s members are not in the group export; add them to member_snapshots.csv'
                                 ELSE ' on this day: its member export (with its marker row) was missed' END
               WHEN ds.form = 'history' AND ds.day < (SELECT min(pm.start_d) FROM pm WHERE pm.fno = f.fno)
                    THEN 'before the earliest grant on this table (' || strftime((SELECT min(pm.start_d) FROM pm WHERE pm.fno = f.fno), '%Y-%m-%d')
                         || '): grants on a dropped or replaced table are not in the history'
               ELSE NULL END AS reason
        FROM fin f JOIN daystate ds ON ds.platform = f.platform""")
    con.execute("""CREATE TEMP TABLE cov AS SELECT t.platform, t.fno, t.day FROM tday t
                   JOIN daystate ds ON ds.platform = t.platform AND ds.day = t.day
                   WHERE ds.reason IS NULL AND t.reason IS NULL""")
    con.execute(f"""CREATE TEMP TABLE acc AS SELECT * FROM (VALUES ('READ','read'),('CHANGE','change'),('BOTH','read'),('BOTH','change')) v(level, access)""")
    con.execute(f"""CREATE TEMP TABLE dayrows AS
        SELECT DISTINCT r.platform, r.fno, r.node, r.is_user, r.node_kind, a.access, r.via, pm.priv,
               unnest(generate_series(greatest(r.s, {ps}), least(r.e, {pe}), INTERVAL 1 DAY))::DATE AS day
        FROM reach r JOIN pm ON pm.pid = r.pid AND pm.fno = r.fno JOIN acc a ON a.level = pm.level
        WHERE greatest(r.s, {ps}) <= least(r.e, {pe})""")
    con.execute("""CREATE TEMP TABLE covdays AS SELECT d.* FROM dayrows d JOIN cov c ON c.platform = d.platform AND c.fno = d.fno
                   AND c.day = d.day""")
    pub_sql = ", ".join(f"('{p.lower()}', '{h.lower()}')" for p, hs in PUBLIC_HOLDERS.items() for h in hs) or "('', '')"
    con.execute(f"CREATE TEMP TABLE pubs AS SELECT * FROM (VALUES {pub_sql}) v(platform, holder)")
    con.execute("""CREATE TEMP TABLE edge_holders AS SELECT DISTINCT platform, holder FROM edge""")
    con.execute("""CREATE TEMP TABLE names AS
        SELECT platform, key, min(shown) AS shown FROM (
          SELECT platform, holder AS key, shown FROM shown UNION ALL SELECT platform, holder, holder_shown FROM edge
          UNION ALL SELECT platform, member, member_shown FROM edge
          UNION ALL SELECT lower(trim(platform)), lower(trim(login)), trim(login) FROM t_logins) GROUP BY platform, key""")
    con.execute("""CREATE TEMP TABLE kinds AS
        SELECT n.platform, n.key,
               CASE WHEN EXISTS (SELECT 1 FROM pubs WHERE pubs.platform = n.platform AND pubs.holder = n.key) THEN 'everyone'
                    WHEN n.key LIKE 'share %' THEN 'role or group'
                    WHEN sh.kind <> '' AND EXISTS (SELECT 1 FROM kindmap km WHERE km.k = lower(sh.kind))
                         THEN (SELECT km.kind FROM kindmap km WHERE km.k = lower(sh.kind))
                    WHEN EXISTS (SELECT 1 FROM edge_holders eh WHERE eh.platform = n.platform AND eh.holder = n.key) THEN 'role or group'
                    ELSE 'login' END AS kind
        FROM names n LEFT JOIN shown sh ON sh.platform = n.platform AND sh.holder = n.key""")
    con.execute("""CREATE TEMP TABLE popraw AS
        SELECT f.platform, f.shown_name AS tname, d.node, nm.shown AS principal,
               CASE WHEN d.is_user THEN 'login' WHEN d.node_kind IN ('role', 'group') AND kd.kind <> 'everyone'
                    THEN 'role or group' ELSE kd.kind END AS kind, d.access, d.day, d.via, d.priv
        FROM covdays d JOIN fin f ON f.fno = d.fno JOIN names nm ON nm.platform = d.platform AND nm.key = d.node
        JOIN kinds kd ON kd.platform = d.platform AND kd.key = d.node""")
    con.execute("""CREATE TEMP TABLE logins AS
        SELECT lower(trim(platform)) AS platform, lower(trim(login)) AS login, lower(trim(email)) AS email FROM t_logins""")
    con.execute("""CREATE TEMP TABLE pop AS
        SELECT p.platform, p.tname AS "table", p.principal, p.kind, coalesce(l.email, '') AS person, p.access,
               strftime(min(p.day), '%Y-%m-%d') AS first_day, strftime(max(p.day), '%Y-%m-%d') AS last_day,
               count(DISTINCT p.day) AS days,
               string_agg(DISTINCT CASE WHEN p.via = '' THEN 'direct' ELSE 'via ' || p.via END,
                          '; ' ORDER BY CASE WHEN p.via = '' THEN 'direct' ELSE 'via ' || p.via END) AS how,
               string_agg(DISTINCT p.priv, ', ' ORDER BY p.priv) AS privileges
        FROM popraw p LEFT JOIN logins l ON p.kind = 'login' AND l.platform = p.platform AND l.login = p.node
        GROUP BY p.platform, p.tname, p.principal, p.kind, coalesce(l.email, ''), p.access""")

    # --- the changes in the period
    con.execute(f"""CREATE TEMP TABLE chg AS
        SELECT d AS change_date, what AS change, f.platform, f.shown_name AS "table", holder_shown AS holder, detail, granted_by, basis FROM (
          SELECT DISTINCT pm.fno, pm.start_d AS d, 'GRANT' AS what, nm.shown AS holder_shown, pm.priv AS detail, pm.granted_by, pm.basis
          FROM pm JOIN names nm ON nm.platform = pm.platform AND nm.key = pm.holder
          WHERE pm.start_is_change AND pm.start_d BETWEEN {ps} AND {pe} AND pm.level <> 'NONE'
          UNION ALL
          SELECT DISTINCT pm.fno, pm.end_event, 'REVOKE', nm.shown, pm.priv, pm.granted_by, pm.basis
          FROM pm JOIN names nm ON nm.platform = pm.platform AND nm.key = pm.holder
          WHERE pm.end_event BETWEEN {ps} AND {pe} AND pm.level <> 'NONE'
          UNION ALL
          SELECT DISTINCT r.fno, e.start_d, 'MEMBER ADDED', e.holder_shown, e.member_shown, e.granted_by, e.basis
          FROM edge e JOIN reach r ON r.platform = e.platform AND r.node = e.holder AND NOT r.is_user AND e.start_d BETWEEN r.s AND r.e
          WHERE e.start_is_change AND e.start_d BETWEEN {ps} AND {pe}
          UNION ALL
          SELECT DISTINCT r.fno, e.end_event, 'MEMBER REMOVED', e.holder_shown, e.member_shown, e.granted_by, e.basis
          FROM edge e JOIN reach r ON r.platform = e.platform AND r.node = e.holder AND NOT r.is_user
            -- the member's last day held falls in the role's access: a dated record's end_event is its last day,
            -- a daily export's is the first export without the member
            AND (CASE WHEN e.basis = 'daily exports' THEN e.end_event - 1 ELSE e.end_event END) BETWEEN r.s AND r.e
          WHERE e.end_event BETWEEN {ps} AND {pe}) x JOIN fin f ON f.fno = x.fno""")

    # --- NOT COVERED, as ranges of days with one reason
    con.execute("""CREATE TEMP TABLE nc AS
        WITH a AS (SELECT platform, '' AS scope, day, reason FROM daystate WHERE reason IS NOT NULL
                   UNION ALL SELECT t.platform, f.shown_name, t.day, t.reason FROM tday t JOIN fin f ON f.fno = t.fno
                   WHERE t.reason IS NOT NULL),
             x AS (SELECT *, row_number() OVER (PARTITION BY platform, scope, reason ORDER BY day) AS rn FROM a),
             y AS (SELECT platform, scope, reason, day, day - CAST(rn AS INTEGER) AS grp FROM x)
        SELECT platform, scope, strftime(min(day), '%Y-%m-%d') AS from_day, strftime(max(day), '%Y-%m-%d') AS to_day,
               count(*) AS n_days, reason FROM y GROUP BY platform, scope, reason, grp""")

    # ---- print
    pop_rows = con.execute("""SELECT platform, "table", principal, kind, person, access, first_day, last_day, days, how, privileges
                              FROM pop ORDER BY platform, "table", principal, kind, access, how""").fetchall()
    sha = hashlib.sha256("\n".join("|".join(str(c) for c in r) for r in pop_rows).encode("utf-8")).hexdigest()
    n_tab = q("SELECT count(*) FROM fin")
    period_days = len(days(PERIOD_START, PERIOD_END))
    print(f"\nPopulation: {len(pop_rows)} rows for {n_tab} financial tables, period {PERIOD_START} to {PERIOD_END} "
          f"({period_days} days), exports taken {AS_OF}, SHA-256 {sha[:16]}...")
    print("\nPlatforms")
    for p in PLATFORMS:
        c = q(f"SELECT count(*) FROM daystate WHERE platform = '{p}' AND reason IS NULL")
        tag = "NOT CHECKED" if c == 0 else "checked"
        print(f"  {p:<11} {tag:<11} {FORM[p]}: {c} of {period_days} days covered")
    print("\n1. Who could read or change each financial table (principals with a person's email where logins.csv gives one)")
    wide = max([len(r[1]) for r in pop_rows] + [5])
    wp = max([len(r[2]) for r in pop_rows] + [9])
    print(f"  {'table':<{wide}} {'principal':<{wp}} {'kind':<13} {'access':<6} {'first day':<10} {'last day':<10} {'days':>4}  how")
    for plat, t, who, kind, person, access, fd, ld, nd, how, privs in pop_rows:
        print(f"  {t:<{wide}} {who:<{wp}} {kind:<13} {access:<6} {fd:<10} {ld:<10} {nd:>4}  {how} ({privs})")
    if not pop_rows:
        print("  none")
    ppl = q("SELECT count(DISTINCT person) FROM pop WHERE person <> ''")
    unm = con.execute("""SELECT platform, principal, count(DISTINCT "table") FROM pop WHERE kind = 'login' AND person = ''
                         GROUP BY 1, 2 ORDER BY 1, 2""").fetchall()
    nrg = q("SELECT count(DISTINCT principal) FROM pop WHERE kind = 'role or group'")
    print(f"\n{ppl} people matched by email; {len(unm)} login(s) with no email in logins.csv; {nrg} role(s) or group(s) listed.")
    for plat, who, nt in unm:
        blank_kind = con.execute("SELECT count(*) FROM shown WHERE platform = ? AND lower(shown) = lower(?) AND kind = ''",
                                 [plat, who]).fetchone()[0]
        looks_like_login = "@" in who or re.fullmatch(r"[0-9a-fA-F-]{36}", who)   # an email, or an application id
        never_holder = (plat == "snowflake" and not con.execute(
            "SELECT count(*) FROM shown WHERE platform = ? AND lower(shown) = lower(?)", [plat, who]).fetchone()[0]
            and not con.execute("SELECT count(*) FROM ms WHERE platform = ? AND holder = lower(?)", [plat, who]).fetchone()[0])
        hint = ("; no holder_kind and no members exported: if this is a group, export its members"
                if blank_kind and not looks_like_login else
                "; reached as a member and never exported as a role: if it is a role, export its members"
                if never_holder and not looks_like_login else "")
        print(f"  unmatched login  {plat:<11} {who} ({nt} financial table(s){hint})")
    empty = con.execute("""SELECT DISTINCT p.platform, p.principal FROM pop p JOIN kinds k ON k.platform = p.platform
                           AND k.kind = 'role or group' AND p.kind = 'role or group'
                           WHERE NOT EXISTS (SELECT 1 FROM edge_holders eh WHERE eh.platform = p.platform
                                             AND eh.holder = lower(p.principal)) ORDER BY 1, 2""").fetchall()
    for plat, who in empty:
        exported = con.execute("SELECT count(*) FROM ms WHERE platform = ? AND holder = lower(?)", [plat, who]).fetchone()[0]
        if exported:   # only its marker rows: the export ran and found nobody
            print(f"  exported with no members: {plat} {who}")
        else:
            print(f"  no members exported for {plat} {who}: its people are not in the list above")

    print("\n2. Changes in the period (grants, revokes and people added to or removed from a role or group that holds access)")
    chg = con.execute("""SELECT strftime(change_date, '%Y-%m-%d'), change, platform, "table", holder, detail, granted_by, basis
                         FROM chg ORDER BY 1, change, platform, "table", holder, detail""").fetchall()
    for d, c, plat, t, h, det, gb, basis in chg:
        print(f"  {d} {c:<14} {plat:<11} {t}  {h}  {det}  by {gb or 'not recorded'}  ({basis})")
    if not chg:
        print("  none")

    print("\n3. NOT COVERED (days these exports cannot rebuild; never read as nobody had access)")
    ncr = con.execute("SELECT platform, scope, from_day, to_day, n_days, reason FROM nc ORDER BY platform, scope, from_day").fetchall()
    for plat, scope, a, b, nd, why in ncr:
        print(f"  {plat:<11} {(scope + ' ') if scope else ''}{a} to {b} ({nd} day(s)): {why}")
    if not ncr:
        print("  none")
    nnc = sum(r[4] for r in ncr if not r[1])
    ntc = sum(r[4] for r in ncr if r[1])
    print(f"\n{n_tab} financial tables, {len(pop_rows)} access rows, {len(chg)} changes, {nnc} platform-day(s) NOT COVERED"
          + (f", {ntc} table-day(s) NOT COVERED." if ntc else "."))

    # --- the evidence files
    def write(name, header, rows):
        path = out / name
        with open(path, "w", newline="", encoding="utf-8") as f:
            w = csv.writer(f)
            w.writerow(header)
            w.writerows([[cell(v) for v in r] for r in rows])
        return path
    p1 = write(f"access_population{suffix}.csv", ["platform", "table", "principal", "kind", "person", "access", "first_day",
                                                   "last_day", "days", "how", "privileges"], pop_rows)
    write(f"access_changes{suffix}.csv", ["change_date", "change", "platform", "table", "holder", "detail", "granted_by", "basis"], chg)
    write(f"access_not_covered{suffix}.csv", ["platform", "table", "from_day", "to_day", "days", "reason"], ncr)
    print(f"Evidence written to {out}/: {p1.name} ({len(pop_rows)} rows), access_changes{suffix}.csv, access_not_covered{suffix}.csv.")
    return 1 if ("--strict" in args and (nnc or ntc)) else 0


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