#!/usr/bin/env python3
"""Prove or rule out writer conflicts on critical CRM fields from a property or field history export.

Usage (from a private working folder; SKILL_DIR is this skill's folder):
  python3 -I "$SKILL_DIR/scripts/writer_history.py" <history.csv|json ...> --format generic|hubspot|salesforce|attio
      [--field hubspot_owner_id] [--object contacts] [--window-days 30] [--as-of YYYY-MM-DD]
      [--group value,time,source] [--list-sources sources.csv]
      [--writer-map writers.csv] [--config audit-config.json] [--json] [--out audit-work/history.json]

Formats:
  generic     CSV with object, record_id, field, changed_at, value, writer
  hubspot     HubSpot's Export property history file is wide: one row per record, then a repeating group
              per revision, newest first, under headers HubSpot does not document. Read the header row,
              then pass --field, --object and --group with the group's columns in order (value, time,
              source, or skip for a column to ignore). Export with Include Source Information ticked, or
              the file names no writer and the run stops. --list-sources writes each distinct source to a
              local file for the user to label; pass the filled file as --writer-map. An unmapped source
              counts as automated and shows as a short hash, so no address reaches the output.
              A long file (one change per row, such as an API read) is read by column alias instead;
              source type plus source label or ID become the writer. Workflows and apps often appear by
              numeric ID: map each to its inventory label with --writer-map.
  salesforce  field history query output: ParentId (or AccountId, OpportunityId...), Field, DataType,
              NewValue, CreatedDate, CreatedById; use --writer-map to turn user IDs into labels.
              Lookup fields such as Owner write two rows per change, one with record IDs (DataType
              EntityId) and one with names (DataType Text); the ID row is kept. Without a DataType
              column, rows that share record, field, CreatedDate and CreatedById count as one change.
  attio       JSON list of {record_id, object, field, data: [historic values]}, or one API response
              with --record-id and --field

A flip-flop record has value A, then value B from a different writer, then A again from a writer other
than B's, with all three changes inside the window. It is "between automated writers" when no person
made any of the three changes, and "reversing a person's change" otherwise. FLD-02 reads FAIL at 3 or
more flip-flop records, or when flip-flops between automated writers reach 1% of records changed; WARN at
1 or 2 records otherwise. Values are hashed with SHA-256 as they are read and are never printed or
written; equality is all the analysis needs. Human writers without a writer-map label are reported as
"user"; unmapped user IDs count as automated until the writer map names them.

Reads local files only and writes only the --out file. Exit code: 0 on valid input (the output is
evidence, not a verdict), 2 on bad input.
"""
import argparse
import csv
import datetime as dt
import hashlib
import json
import os
import re
import sys
from collections import Counter, defaultdict

HUMAN_TYPES = {"user", "human", "manual", "crm-ui", "ui", "workspace-member", "form"}
DEFAULTS = {"window_days": 30, "flip_flop_records_fail": 3, "flip_flop_rate_fail": 0.01, "flip_flop_records_warn": 1}
SF_ID = re.compile(r"^[A-Za-z0-9]{15}(?:[A-Za-z0-9]{3})?$")
HS_TIME = ("timestamp", "changed at", "date", "updated at", "change date")
HS_VALUE = ("value", "new value", "property value")
HS_CAPS = {"contacts": 45, "leads": None}  # revisions per property in Export property history; other objects 20


class BadInput(Exception):
    pass


def fail(msg):
    print("writer_history: " + msg, file=sys.stderr)
    sys.exit(2)


def slug(s):
    return re.sub(r"[^a-z0-9]+", "-", str(s or "").strip().lower()).strip("-")


def norm_writer(label, trusted=False):
    """type:name in lowercase. Human writers become "user" unless the label came from the writer map."""
    label = str(label or "").strip()
    if not label:
        return "unknown"
    if ":" in label:
        t, n = label.split(":", 1)
        t, n = slug(t), slug(n)
    else:
        t, n = slug(label), ""
    if t in HUMAN_TYPES and t != "form":
        return "user:" + n if (trusted and n) else "user"
    return t + (":" + n if n else "")


def is_human(label):
    return label.split(":", 1)[0] in HUMAN_TYPES


def hashed(value):
    return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode("utf-8")).hexdigest()


def parse_ts(s):
    s = str(s or "").strip()
    if not s:
        return None
    try:
        if re.match(r"^\d{4}-\d{2}-\d{2}$", s):
            return dt.datetime.strptime(s, "%Y-%m-%d").replace(tzinfo=dt.timezone.utc)
        s2 = s.replace("Z", "+00:00").replace(" ", "T", 1)
        s2 = re.sub(r"\.(\d{6})\d+", r".\1", s2)
        s2 = re.sub(r"([+-]\d{2})(\d{2})$", r"\1:\2", s2)
        d = dt.datetime.fromisoformat(s2)
        return d if d.tzinfo else d.replace(tzinfo=dt.timezone.utc)
    except ValueError:
        try:
            return dt.datetime.fromtimestamp(float(s) / (1000.0 if float(s) > 1e11 else 1.0), tz=dt.timezone.utc)
        except ValueError:
            return None


def norm_key(h):
    return re.sub(r"[^a-z0-9]+", "", str(h or "").replace("\ufeff", "").lower())


def pick(row, *names):
    keys = {norm_key(k): k for k in row}
    for n in names:
        if norm_key(n) in keys:
            return row[keys[norm_key(n)]]
    return None


def read_csv(path):
    try:
        with open(path, newline="", encoding="utf-8-sig") as f:
            return list(csv.DictReader(f))
    except (OSError, csv.Error, UnicodeDecodeError) as e:
        raise BadInput("cannot read %s: %s" % (path, e))


def read_hubspot(path, args):
    """Export property history is wide, so its groups become one long row per revision, oldest first. csv.reader
    keeps repeated headers, which a dict would collapse into the oldest group alone. A long file passes through."""
    try:
        with open(path, newline="", encoding="utf-8-sig") as f:
            rows = [r for r in csv.reader(f) if any(c.strip() for c in r)]
    except (OSError, csv.Error, UnicodeDecodeError) as e:
        raise BadInput("cannot read %s: %s" % (path, e))
    if not rows:
        return []
    header = [h.strip() for h in rows[0]]
    keys = [norm_key(h) for h in header]
    group = [g.strip().lower() for g in (getattr(args, "group", None) or "").split(",") if g.strip()]
    if not group:
        if len(set(keys)) == len(keys) and any(norm_key(n) in keys for n in HS_TIME) and any(norm_key(n) in keys for n in HS_VALUE):
            return [dict(zip(header, r)) for r in rows[1:]]
        raise BadInput("%s is not one change per row. HubSpot's Export property history is wide: read the header row, "
                       "then pass --field, --object and --group with one revision's columns in order, such as "
                       "--group value,time,source" % path)
    if "value" not in group or "time" not in group or set(group) - {"value", "time", "source", "skip"}:
        raise BadInput("--group lists value, time and source, plus skip for a column to ignore, in column order")
    if "source" not in group:
        raise BadInput("--group has no source column: re-export with Include Source Information ticked")
    if not getattr(args, "field", None):
        raise BadInput("a wide export covers one property: pass --field with its internal name")
    id_keys = [i for i, k in enumerate(keys) if k in ("recordid", "objectid", "hsobjectid", "id")]
    if not id_keys:
        raise BadInput("%s has no Record ID column (first columns: %s)" % (path, ", ".join(header[:6])))
    out, start = [], id_keys[0] + 1
    for r in rows[1:]:
        cells = r[start:]
        groups = [dict(zip(group, cells[i:i + len(group)])) for i in range(0, len(cells), len(group))]
        for g in reversed(groups):  # newest first in the file, so the oldest goes first here
            if (g.get("value") or "").strip() or (g.get("time") or "").strip():
                out.append({"record id": r[id_keys[0]], "property": args.field, "timestamp": g.get("time"),
                            "value": g.get("value"), "source text": (g.get("source") or "").strip()})
    return out


def load_input(path, fmt, args=None):
    if fmt == "hubspot":
        return read_hubspot(path, args)
    if fmt != "attio":
        return read_csv(path)
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except (OSError, ValueError) as e:
        raise BadInput("cannot read %s: %s" % (path, e))


def load_writer_map(path):
    out = {}
    for r in (read_csv(path) if path else []):
        raw = pick(r, "raw", "id", "writer", "user_id", "source")
        label = pick(r, "label", "writer_label", "role")
        if raw and label:
            out[str(raw).strip()] = str(label).strip()
    return out


def hubspot_writer(source_type, source_label):
    t = str(source_type or "").lower()
    label = str(source_label or "").strip()
    if "@" in label:  # never carry an email address into a writer label
        label = "redacted"
    if "ui" in t or "user" in t or "crm" == t or "mobile" in t:
        kind = "user"
    elif "workflow" in t or "automation" in t:
        kind = "workflow"
    elif "salesforce" in t:
        return "integration:salesforce"
    elif "integration" in t or "app" in t:
        kind = "integration"
    elif "import" in t:
        kind = "import"
    elif "form" in t:
        kind = "form"
    elif "api" in t:
        kind = "api"
    else:
        kind = slug(t) or "unknown"
    return kind if kind == "user" else kind + (":" + label if label else "")


# ---------- readers: parsed input -> (object, record_id, field, timestamp, value, raw writer) rows, notes ----------

def rows_generic(data, args):
    return [(pick(r, "object") or args.object or "record", pick(r, "record_id", "id"),
             pick(r, "field", "property"), pick(r, "changed_at", "timestamp", "date"),
             pick(r, "value", "new_value"), pick(r, "writer", "source")) for r in data], []


def source_label(text, wmap):
    """A wide export gives one source text per change. A mapped text keeps its map key. An unmapped one becomes a
    short hash, because the text can hold a person's address; it counts as automated until the map names it."""
    if not text:
        return "unknown"
    return text if text in wmap else "id:" + hashlib.sha256(text.lower().encode("utf-8")).hexdigest()[:8]


def rows_hubspot(data, args):
    out, notes = [], []
    wmap = getattr(args, "wmap", None) or {}
    per_record = Counter(r["record id"] for r in data if "source text" in r)
    if per_record:
        cap = HS_CAPS.get(args.object, 20)
        at_cap = [k for k, n in per_record.items() if cap and n >= cap]
        if at_cap:
            notes.append("%d records reached the export's cap of %d revisions, so older changes are missing: %s" % (
                len(at_cap), cap, ", ".join(sorted(at_cap)[:5])))
    for r in data:
        if "source text" in r:
            raw = source_label(r["source text"], wmap)
        else:
            raw = hubspot_writer(pick(r, "source type", "source", "change source"),
                                 pick(r, "source label", "source id", "source detail", "changed by", "source name"))
        out.append((pick(r, "object", "object type") or args.object or "record",
                    pick(r, "record id", "object id", "hs_object_id", "id"),
                    pick(r, "property", "property name", "internal name", "field") or args.field,
                    pick(r, *HS_TIME),
                    pick(r, *HS_VALUE), raw))
    return out, notes


def looks_like_sf_id(v):
    s = str(v or "").strip()
    return bool(SF_ID.match(s)) and any(ch.isdigit() for ch in s) and any(ch.isalpha() for ch in s)


def rows_salesforce(data, args):
    """One change per row, except lookup fields, which write an ID row and a name row per change."""
    recs = [{"rid": pick(r, "ParentId", "AccountId", "OpportunityId", "ContactId", "LeadId", "CaseId", "Id"),
             "field": pick(r, "Field") or args.field, "ts": pick(r, "CreatedDate"), "value": pick(r, "NewValue"),
             "by": str(pick(r, "CreatedById") or "unknown").strip(), "dtype": str(pick(r, "DataType") or "").strip().lower()}
            for r in data]
    notes = []
    if any(x["dtype"] for x in recs):
        with_ids = set((x["rid"], x["field"]) for x in recs if x["dtype"] == "entityid")
        kept = [x for x in recs if not (x["dtype"] == "text" and (x["rid"], x["field"]) in with_ids)]
        if len(kept) < len(recs):
            notes.append("lookup history: %d name rows (DataType Text) dropped beside their record ID rows (EntityId)"
                         % (len(recs) - len(kept)))
    else:
        groups = defaultdict(list)
        for x in recs:
            groups[(x["rid"], x["field"], x["ts"], x["by"])].append(x)
        kept = []
        for x in recs:
            g = groups[(x["rid"], x["field"], x["ts"], x["by"])]
            ids = [y for y in g if looks_like_sf_id(y["value"])]
            if x is (ids[0] if ids else g[0]):
                kept.append(x)
        if len(kept) < len(recs):
            notes.append("no DataType column: %d rows that share record, field, CreatedDate and CreatedById were merged "
                         "into one change each, keeping the value that looks like a record ID" % (len(recs) - len(kept)))
    return [(args.object or "record", x["rid"], x["field"], x["ts"], x["value"], "id:" + x["by"]) for x in kept], notes


def rows_attio(data, args):
    blocks = data if isinstance(data, list) else [{"record_id": args.record_id, "field": args.field,
                                                   "object": args.object, "data": (data or {}).get("data", [])}]
    out = []
    for b in blocks:
        for v in (b.get("data") or []) if isinstance(b, dict) else []:
            actor = v.get("created_by_actor") or {}
            raw = str(actor.get("type") or "unknown") + ":" + str(actor.get("id") or "")
            value = {k: x for k, x in v.items() if k not in ("active_from", "active_until", "created_by_actor", "attribute_type")}
            out.append((b.get("object") or args.object or "record", b.get("record_id"),
                        b.get("field") or b.get("attribute") or args.field, v.get("active_from"), value, raw))
    return out, []


READERS = {"generic": rows_generic, "hubspot": rows_hubspot, "salesforce": rows_salesforce, "attio": rows_attio}


# ---------- analysis ----------

def analyse(changes, window_start, window_end, thresholds):
    by_record = defaultdict(list)
    for c in changes:
        by_record[c["record_id"]].append(c)
    writer_counts = Counter()
    multi, flips, last_writer = 0, [], Counter()
    auto_flips, human_flips = [], []
    changed = 0
    for rid, rows in by_record.items():
        rows.sort(key=lambda c: (c["ts"], c["seq"]))
        inwin = [c for c in rows if window_start <= c["ts"].date() <= window_end]
        if not inwin:
            continue
        changed += 1
        for c in inwin:
            writer_counts[c["writer"]] += 1
        if len(set(c["writer"] for c in inwin)) >= 2:
            multi += 1
        last_writer[inwin[-1]["writer"]] += 1
        seq = []
        for c in inwin:  # collapse runs of the same value, keeping the first writer
            if seq and seq[-1]["vhash"] == c["vhash"]:
                continue
            seq.append(c)
        kind = None
        for i in range(len(seq) - 2):
            a, b, c = seq[i], seq[i + 1], seq[i + 2]
            if a["vhash"] == c["vhash"] != b["vhash"] and b["writer"] != a["writer"] and c["writer"] != b["writer"]:
                if not any(is_human(x["writer"]) for x in (a, b, c)):
                    kind = "automated"
                    break
                kind = "human"
        if kind:
            flips.append(rid)
            (auto_flips if kind == "automated" else human_flips).append(rid)
    n, na, nh = len(flips), len(auto_flips), len(human_flips)
    rate = (n / float(changed)) if changed else None
    auto_rate = (na / float(changed)) if changed else None
    fail_n, fail_rate = thresholds["flip_flop_records_fail"], thresholds["flip_flop_rate_fail"]
    if changed == 0:
        status, why = "NOT CHECKED", "no changes inside the window"
    elif n >= fail_n:
        status, why = "FAIL", "%d flip-flop records; fail at %d or more" % (n, fail_n)
    elif na >= 1 and auto_rate >= fail_rate:
        status = "FAIL"
        why = "flip-flops between automated writers on %d of %d records changed (%.1f%%); fail at %.0f%%" % (
            na, changed, 100 * auto_rate, 100 * fail_rate)
    elif n >= thresholds["flip_flop_records_warn"]:
        status = "WARN"
        why = "%d flip-flop record%s, %s; warn at 1 or 2" % (
            n, "" if n == 1 else "s", "an automated writer reversing a person's change" if not na else
            "between automated writers on under %.0f%% of records changed" % (100 * fail_rate))
    else:
        status, why = "PASS", "no flip-flops among the records changed (%d)" % changed
    total_last = sum(last_writer.values()) or 1
    return {
        "records_changed": changed, "changes": sum(writer_counts.values()),
        "changes_by_writer": dict(writer_counts.most_common()), "multi_writer_records": multi,
        "flip_flop_records": n, "automated_flip_flop_records": na, "human_reversal_records": nh,
        "flip_flop_rate": round(rate, 4) if rate is not None else None,
        "automated_flip_flop_rate": round(auto_rate, 4) if auto_rate is not None else None,
        "last_writer_share": {w: round(k / float(total_last), 3) for w, k in last_writer.most_common()},
        "sample_record_ids": flips[:10], "status": status, "status_reason": why,
    }


def run(datasets, args, wmap, thresholds, as_of, window, sources=()):
    """Parsed inputs to the JSON result. Raises BadInput when no row is usable."""
    window_start = as_of - dt.timedelta(days=window - 1)
    changes, notes, bad, seq = [], [], 0, 0
    unknown_type, raw_named, numeric = set(), set(), set()
    args.wmap = wmap
    for data in datasets:
        rows, reader_notes = READERS[args.format](data, args)
        notes.extend(reader_notes)
        for obj, rid, field, ts, value, raw in rows:
            seq += 1
            t = parse_ts(ts)
            if not rid or not field or t is None:
                bad += 1
                continue
            if args.field and str(field) != args.field:
                continue
            raw = str(raw or "unknown").strip()
            label = wmap.get(raw) or wmap.get(raw.split(":", 1)[-1])
            trusted = bool(label)
            if not label:
                if raw.startswith("id:"):
                    unknown_type.add(raw)
                elif raw.split(":", 1)[0] in ("api-token", "app"):
                    raw_named.add(raw)
                label = raw
            writer = norm_writer(label, trusted)
            if not trusted and writer.partition(":")[2].isdigit():
                numeric.add(writer)
            changes.append({"object": str(obj), "record_id": str(rid), "field": str(field), "ts": t, "blank": value is None,
                            "vhash": hashed(value), "writer": writer, "seq": seq})
    if not changes:
        raise BadInput("no usable rows (%d rows lacked a record ID, field or parseable time)" % bad)
    if all(c["writer"] == "unknown" for c in changes):  # every change would look like one writer: no flip-flop could show
        raise BadInput("no row names a writer: map the source column, or re-export HubSpot history with Include Source "
                       "Information ticked")
    if all(c["blank"] for c in changes):  # every value would look equal: no flip-flop could show
        raise BadInput("no row holds a value: map the value column")
    groups = defaultdict(list)
    for c in changes:
        groups[(c["object"], c["field"])].append(c)
    results = []
    for (obj, field), rows in sorted(groups.items()):
        r = analyse(rows, window_start, as_of, thresholds)
        r.update({"object": obj, "field": field, "rows_read": len(rows), "window_days": window})
        results.append(r)
    if bad:
        notes.append("rows skipped for lack of a record ID, field or parseable time: %d" % bad)
    if unknown_type:
        notes.append("writers not in the writer map (user IDs or hashed source text), so each may be a person or an "
                     "integration; they count as automated until --writer-map names them: %s" % ", ".join(sorted(unknown_type)))
    if raw_named:
        notes.append("app and token writers are shown by raw ID; add them to --writer-map to name them: %s"
                     % ", ".join(sorted(raw_named)))
    if numeric:
        notes.append("writers shown by numeric ID: %s; map each to its inventory label (type:name) with --writer-map "
                     "before reading FLD-02, or one workflow can count as two writers" % ", ".join(sorted(numeric)))
    return {"generated_by": "writer_history.py", "format": args.format, "window_days": window,
            "as_of": as_of.isoformat(), "window_start": window_start.isoformat(), "sources": list(sources),
            "fields": results, "notes": notes}


def main():
    p = argparse.ArgumentParser(description="Writer and flip-flop analysis for critical CRM fields.")
    p.add_argument("files", nargs="+")
    p.add_argument("--format", required=True, choices=sorted(READERS))
    p.add_argument("--field", help="only this field (or the field name when the file has none)")
    p.add_argument("--object", help="object name when the file has none")
    p.add_argument("--record-id", help="attio: record ID for a single API response")
    p.add_argument("--window-days", type=int, default=None)
    p.add_argument("--as-of", help="YYYY-MM-DD; the window is the N days ending on this date (default: today, UTC)")
    p.add_argument("--group", help="hubspot wide export: one revision's columns in order, such as value,time,source")
    p.add_argument("--list-sources", help="hubspot wide export: write raw,label,count here for the user to label, then stop")
    p.add_argument("--writer-map", help="CSV with raw,label columns, such as workflow:585051946,workflow:inbound-rotation")
    p.add_argument("--config", help="audit-config.json, for the FLD-02 thresholds")
    p.add_argument("--json", action="store_true", help="print JSON instead of text")
    p.add_argument("--out", help="also write the JSON result to this file")
    args = p.parse_args()

    th = dict(DEFAULTS)
    try:
        if args.config:
            try:
                with open(args.config, encoding="utf-8") as f:
                    th.update((json.load(f).get("FLD-02") or {}))
            except (OSError, ValueError) as e:
                raise BadInput("cannot read config %s: %s" % (args.config, e))
        window = args.window_days or int(th["window_days"])
        try:
            as_of = dt.datetime.strptime(args.as_of, "%Y-%m-%d").date() if args.as_of else dt.datetime.now(dt.timezone.utc).date()
        except ValueError:
            raise BadInput("--as-of must be YYYY-MM-DD, got %r" % args.as_of)
        wmap = load_writer_map(args.writer_map)
        datasets = []
        for path in args.files:
            if not os.path.isfile(path):
                raise BadInput("no such file: %s" % path)
            datasets.append(load_input(path, args.format, args))
        if args.list_sources:
            texts = Counter(r["source text"] for d in datasets for r in d if isinstance(r, dict) and r.get("source text"))
            if not texts:
                raise BadInput("--list-sources needs a wide HubSpot export with --group and a source column that is filled")
            with open(args.list_sources, "w", newline="", encoding="utf-8") as f:
                w = csv.writer(f, lineterminator="\n")
                w.writerow(["raw", "label", "count"])
                for text, n in sorted(texts.items(), key=lambda kv: (-kv[1], kv[0])):
                    w.writerow([text, wmap.get(text, ""), n])
            print("%d distinct sources written to %s; fill the label column with type:name labels and pass the file as "
                  "--writer-map. Nothing was changed." % (len(texts), os.path.basename(args.list_sources)))
            return
        out = run(datasets, args, wmap, th, as_of, window, [os.path.basename(f) for f in args.files])
    except BadInput as e:
        fail(str(e))
    if args.out:
        os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
        with open(args.out, "w", encoding="utf-8") as f:
            f.write(json.dumps(out, indent=1) + "\n")
    if args.json:
        print(json.dumps(out, indent=1))
        return
    for r in out["fields"]:
        print("writer_history: %s.%s, window %d days ending %s (from %s)" % (r["object"], r["field"], window, out["as_of"], out["window_start"]))
        print("  records changed: %d; changes: %d" % (r["records_changed"], r["changes"]))
        print("  changes by writer: " + ", ".join("%s %d" % kv for kv in r["changes_by_writer"].items()))
        print("  records with two or more writers: %d" % r["multi_writer_records"])
        if r["flip_flop_rate"] is not None:
            kinds = [k for k in ("%d between automated writers" % r["automated_flip_flop_records"]
                                 if r["automated_flip_flop_records"] else "",
                                 "%d reversing a person's change" % r["human_reversal_records"]
                                 if r["human_reversal_records"] else "") if k]
            print("  flip-flop records: %d (%.1f%% of records changed)%s; samples: %s" % (
                r["flip_flop_records"], 100 * r["flip_flop_rate"], (": " + ", ".join(kinds)) if kinds else "",
                ", ".join(r["sample_record_ids"]) or "none"))
        print("  last writer share: " + ", ".join("%s %.0f%%" % (w, 100 * s) for w, s in r["last_writer_share"].items()))
        print("  FLD-02 would read: %s (%s)" % (r["status"], r["status_reason"]))
    for n in out["notes"]:
        print("note: " + n)
    print("Values were hashed on read and are not printed. Nothing was changed.")


if __name__ == "__main__":
    main()