#!/usr/bin/env python3
"""Measure how well one system's keys reach another, printing counts only.

Usage (run from the audit folder, by the script's path in the skill folder, so
the --exceptions file is written into the audit folder):
  python3 <skill folder>/scripts/key_match.py --expected EXPECTED.csv
      --expected-key COLUMN --delivered DELIVERED.csv --delivered-key COLUMN
      [--normalize] [--exceptions OUT.csv] [--min-sample 200] [--pass 98]
      [--fail 95] [--dup-fail 0.5] [--json]

Example: paid billing accounts (expected) against CRM companies (delivered),
both keyed on account_id. Column names are matched case-insensitively.

Definitions:
  E  distinct usable expected keys       D  delivered rows
  K  delivered rows with a usable key    R  distinct usable delivered keys
  matched M = keys in both E and R; missing = E - M; unexpected = R - M
  coverage = M / E; excess duplicate rate = (K - R) / K; a zero denominator is n/a
  A key is unusable when blank, or when it has leading or trailing spaces and
  --normalize is off. --normalize trims and lowercases keys before comparing.

Checks: KEY01 coverage (98% or more PASS, 95 to 98% WARN, below 95% FAIL),
KEY02 excess duplicates (above 0.5% FAIL, any WARN), KEY03 unusable keys,
KEY04 email-like keys, KEY05 duplicates inside the expected file,
KEY06 small sample (under --min-sample distinct expected keys: counts only).
Rules: references/flows-identity-and-data-quality.md.

Privacy: never prints key values. --exceptions writes key_or_hash, issue,
count; keys containing '@' are written as SHA-256 hashes of the trimmed,
lowercased value. Keep that file inside the audit folder.

Exit code: 0 no FAIL, 1 at least one FAIL (the normal result of an audit
with findings), 2 bad input or usage.
Standard library only. Reads local files and writes only the --exceptions file.
"""

import argparse
import csv
import hashlib
import io
import json
import os
import sys
from collections import Counter


def fail_input(message):
    print(f"Input error: {message}", file=sys.stderr)
    sys.exit(2)


def read_column(path, column, label):
    """Raw values of one column, whitespace kept, so padded keys can be counted."""
    try:
        with open(path, "rb") as handle:
            raw = handle.read()
    except OSError as exc:
        fail_input(f"cannot read the {label} file {path}: {exc.strerror or exc}")
    if b"\x00" in raw:
        fail_input(f"the {label} file {path} is not a text CSV; save a workbook or UTF-16 export as CSV UTF-8")
    text = ""
    for encoding in ("utf-8-sig", "cp1252", "latin-1"):
        try:
            text = raw.decode(encoding)
            break
        except UnicodeDecodeError:
            continue
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    first = text.split("\n", 1)[0]
    counts = {d: first.count(d) for d in (",", ";", "\t")}
    delimiter = max(counts, key=lambda d: (counts[d], d == ","))
    reader = csv.reader(io.StringIO(text), delimiter=delimiter if counts[delimiter] else ",")
    try:
        header = [h.strip().lower() for h in next(reader)]
        if column.strip().lower() not in header:
            fail_input(f"the {label} file has no column '{column}'; it has: {', '.join(header)}")
        index = header.index(column.strip().lower())
        return [values[index] if index < len(values) else "" for values in reader if any(v.strip() for v in values)]
    except StopIteration:
        fail_input(f"the {label} file {path} is empty")
    except csv.Error as exc:
        fail_input(f"the {label} file {path} cannot be read as CSV: {exc}")


def classify(values, normalize):
    """Split raw values into usable keys and counts of blank and padded keys."""
    usable, blank, padded = [], 0, 0
    for value in values:
        if not value.strip():
            blank += 1
        elif value != value.strip() and not normalize:
            padded += 1
        else:
            usable.append(value.strip().lower() if normalize else value)
    return usable, blank, padded


def pct(numerator, denominator):
    return None if not denominator else numerator / denominator * 100


def show(value):
    return "n/a" if value is None else f"{value:.2f}%"


def plural(n, word):
    return f"{n} {word}" + ("" if n == 1 else "s")


def main():
    parser = argparse.ArgumentParser(description="Key coverage and duplicates between two systems, counts only. Read-only.")
    parser.add_argument("--expected", required=True)
    parser.add_argument("--expected-key", required=True)
    parser.add_argument("--delivered", required=True)
    parser.add_argument("--delivered-key", required=True)
    parser.add_argument("--normalize", action="store_true")
    parser.add_argument("--exceptions")
    parser.add_argument("--min-sample", type=int, default=200)
    parser.add_argument("--pass", dest="pass_line", type=float, default=98.0)
    parser.add_argument("--fail", dest="fail_line", type=float, default=95.0)
    parser.add_argument("--dup-fail", type=float, default=0.5)
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()
    if args.exceptions:
        for source in (args.expected, args.delivered):
            if os.path.exists(args.exceptions) and os.path.samefile(args.exceptions, source):
                fail_input("--exceptions must not point at an input file")

    expected_raw = read_column(args.expected, args.expected_key, "expected")
    delivered_raw = read_column(args.delivered, args.delivered_key, "delivered")
    expected, e_blank, e_padded = classify(expected_raw, args.normalize)
    delivered, d_blank, d_padded = classify(delivered_raw, args.normalize)
    e_set, d_counts = set(expected), Counter(delivered)
    d_set = set(d_counts)
    matched, missing, unexpected = e_set & d_set, e_set - d_set, d_set - e_set
    k_rows, r_keys = len(delivered), len(d_set)
    coverage = pct(len(matched), len(e_set))
    dup_rate = pct(k_rows - r_keys, k_rows)
    dup_keys = sum(1 for n in d_counts.values() if n > 1)
    manifest_dups = len(expected) - len(e_set)
    small = len(e_set) < args.min_sample

    findings = []

    def add(severity, check, name, message):
        findings.append({"severity": severity, "check": check, "name": name, "message": message})

    if small:
        add("NOT CHECKED", "KEY01", "coverage", f"{len(e_set)} distinct expected keys is below --min-sample {args.min_sample}; "
            f"counts only: {len(matched)} matched, {len(missing)} missing.")
        add("NOT CHECKED", "KEY02", "excess-duplicates", f"sample below {args.min_sample}; counts only: "
            f"{plural(k_rows - r_keys, 'excess row')} on {plural(dup_keys, 'key')}.")
    else:
        if coverage is None:
            add("NOT CHECKED", "KEY01", "coverage", "no usable expected keys.")
        else:
            severity = "PASS" if coverage >= args.pass_line else "WARN" if coverage >= args.fail_line else "FAIL"
            add(severity, "KEY01", "coverage", f"{show(coverage)} of expected keys reach the delivered file ({len(matched)} of {len(e_set)}; "
                f"PASS at {args.pass_line:g}%, FAIL below {args.fail_line:g}%); {len(missing)} missing.")
        if dup_rate is None:
            add("NOT CHECKED", "KEY02", "excess-duplicates", "no usable delivered keys.")
        else:
            severity = "FAIL" if dup_rate > args.dup_fail else "WARN" if dup_rate > 0 else "PASS"
            add(severity, "KEY02", "excess-duplicates", f"{plural(k_rows - r_keys, 'excess row')} on {plural(dup_keys, 'key')}, "
                f"{show(dup_rate)} of keyed delivered rows (FAIL above {args.dup_fail:g}%).")
    unusable = []
    if e_blank or e_padded:
        unusable.append(f"expected has {e_blank} blank and {e_padded} padded keys")
    if d_blank or d_padded:
        unusable.append(f"delivered has {d_blank} blank and {d_padded} padded keys")
    if unusable:
        add("WARN", "KEY03", "unusable-keys", "; ".join(unusable) + (". Padded keys match after --normalize." if (e_padded or d_padded) else "."))
    elif expected_raw or delivered_raw:
        add("PASS", "KEY03", "unusable-keys", "no blank or padded keys in either file.")
    else:
        add("NOT CHECKED", "KEY03", "unusable-keys", "neither file has a data row.")
    emailish = [label for label, keys in (("expected", expected), ("delivered", delivered))
                if keys and sum("@" in k for k in keys) > len(keys) / 2]
    if emailish:
        add("WARN", "KEY04", "email-like-keys", f"most {' and '.join(emailish)} keys contain '@'. Email is not a durable key; "
            "prefer an internal ID. Exceptions are written as SHA-256 hashes.")
    elif expected or delivered:
        add("PASS", "KEY04", "email-like-keys", "keys are not email addresses.")
    else:
        add("NOT CHECKED", "KEY04", "email-like-keys", "neither file has a usable key.")
    if manifest_dups:
        add("WARN", "KEY05", "manifest-duplicates", f"the expected file repeats {plural(manifest_dups, 'key')}; rates use distinct keys. "
            "Fix the manifest first.")
    elif expected:
        add("PASS", "KEY05", "manifest-duplicates", "no repeated keys in the expected file.")
    else:
        add("NOT CHECKED", "KEY05", "manifest-duplicates", "the expected file has no usable key.")
    if not small:
        add("PASS", "KEY06", "small-sample", f"{len(e_set)} distinct expected keys, at or above --min-sample {args.min_sample}, so rates are reported.")

    order = ["FAIL", "WARN", "PASS", "NOT CHECKED"]
    findings.sort(key=lambda f: (order.index(f["severity"]), f["check"]))
    counts = {s: sum(1 for f in findings if f["severity"] == s) for s in order}
    written = ""
    if args.exceptions:
        def safe(key):
            return hashlib.sha256(key.strip().lower().encode("utf-8")).hexdigest() if "@" in key else key
        with open(args.exceptions, "w", newline="", encoding="utf-8") as handle:
            writer = csv.writer(handle)
            writer.writerow(["key_or_hash", "issue", "count"])
            writer.writerows([safe(k), "missing", 1] for k in sorted(missing))
            writer.writerows([safe(k), "unexpected", 1] for k in sorted(unexpected))
            writer.writerows([safe(k), "duplicate", n] for k, n in sorted(d_counts.items()) if n > 1)
        written = f" Exceptions written to {os.path.basename(args.exceptions)}."
    summary = (f"{counts['FAIL']} FAIL, {counts['WARN']} WARN, {counts['PASS']} PASS, {counts['NOT CHECKED']} NOT CHECKED. "
               f"Nothing was changed.{written}")

    stats = {"expected_rows": len(expected_raw), "expected_distinct_usable": len(e_set), "delivered_rows": len(delivered_raw),
             "delivered_keyed_rows": k_rows, "delivered_distinct": r_keys, "matched": len(matched), "missing": len(missing),
             "unexpected": len(unexpected), "coverage_pct": None if coverage is None else round(coverage, 2),
             "excess_duplicate_rows": k_rows - r_keys, "excess_duplicate_pct": None if dup_rate is None else round(dup_rate, 2),
             "normalized": args.normalize}
    if args.json:
        print(json.dumps({"stats": stats, "findings": findings, "summary": {**counts, "line": summary}}, indent=2))
    else:
        out = ["Key match (read-only; counts only, key values are never printed)",
               f"  expected: {os.path.basename(args.expected)}, column {args.expected_key}: {len(expected_raw)} rows, {len(e_set)} distinct usable keys",
               f"  delivered: {os.path.basename(args.delivered)}, column {args.delivered_key}: {len(delivered_raw)} rows, {k_rows} with a usable key, {r_keys} distinct",
               f"  matched {len(matched)}; missing {len(missing)}; unexpected {len(unexpected)}"
               + (" (unexpected keys are often trials or churned accounts: explain them before calling them defects)" if unexpected else ""),
               f"  coverage {len(matched)} / {len(e_set)} = {show(coverage)}",
               f"  excess duplicates ({k_rows} - {r_keys}) / {k_rows} = {show(dup_rate)}",
               f"  keys {'trimmed and lowercased (--normalize)' if args.normalize else 'compared exactly'}", ""]
        out += [f"{f['severity']} {f['check']} {f['name']}: {f['message']}" for f in findings]
        out += ["", summary]
        print("\n".join(out))
    return 1 if counts["FAIL"] else 0


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