#!/usr/bin/env python3
"""Profile a raw analytics export into the observed inventory CSV, keeping values local.

Usage (run from a scratch folder; SKILL is the installed skill folder):
  python3 "$SKILL/scripts/profile_export.py" FILE [FILE ...]
      --format mixpanel|amplitude|segment|posthog|generic --tool NAME
      [--event-key PATH --props-key PATH --time-key PATH --source-key PATH]
      [--since YYYY-MM-DD --until YYYY-MM-DD] [--values-for prop,prop]
      [--occurrence-key PATH] [--identity-keys PATH,PATH] [--max-records N] > observed.csv

Reads JSON Lines, a JSON array, .gz files and .zip archives whose members are
JSON Lines or gzip-compressed JSON Lines (Amplitude's Export API returns a ZIP
of JSON files). Dotted paths reach nested fields, for example properties.$lib.
Presets set the paths for each vendor's raw export. When a path is missing on
the first record, the script prints that record's top-level keys (never values)
to stderr, so check paths from that note and never by opening the export.

Version tail rule: run the version split as a separate profiler run, for example
--source-key version_name into versions-amplitude.csv, and read it by hand.
Never pass it to audit_plan.py drift --observed, which refuses version-like
sources.

Records before --since update first_seen and last_seen only, so a 90-day export
with --since at the window start shows events that stopped (rows 0, last_seen
set). Records after --until are ignored. Transport duplicates use each vendor's
own key: Mixpanel raw exports keep repeats of event, distinct_id, time and
$insert_id, while Amplitude already drops a repeat of device_id and insert_id
within 7 days at ingestion.

Writes the columns of assets/observed-inventory-template.csv to stdout and a
summary to stderr. It never writes raw values. --values-for lists the top 50
values of the named properties after replacing emails, phone numbers (also
inside text), URLs with sensitive query keys and text over 64 characters with
tokens; when a property has more than 50 distinct values, values seen fewer
than 5 times fold into <rare>=n. --identity-keys writes ID classes (uuid,
numeric, numeric_over_2_53, email_like, short_under_5, placeholder, other, null)
per event and as a '*' total, never the IDs.

Large exports: every distinct occurrence and transport key is held in memory as
a hash, so try one day first on a laptop. Give all of a tool's files to one run:
drift stops on a row it has already read, so separate runs for separate weeks
cannot be passed together. If the export does not fit, use a warehouse route, or
--max-records and label the inventory partial. Standard library only, Python 3.8
or later. Exit codes: 0 success, 2 bad input.
"""
import argparse
import csv
import gzip
import io
import json
import re
import sys
import zipfile
from collections import Counter, defaultdict
from datetime import date, datetime, timezone

PRESETS = {
    "mixpanel": {"event": "event", "props": "properties", "time": "properties.time",
                 "source": ["properties.mp_lib", "properties.$os"],
                 "transport": ["event", "properties.distinct_id", "properties.time", "properties.$insert_id"]},
    "amplitude": {"event": "event_type", "props": "event_properties", "time": "event_time",
                  "source": ["platform"], "transport": ["device_id", "$insert_id"]},
    "segment": {"event": "event", "props": "properties", "time": "timestamp",
                "source": ["context.library.name"], "transport": ["messageId"]},
    "posthog": {"event": "event", "props": "properties", "time": "timestamp",
                "source": ["properties.$lib"], "transport": ["uuid"]},
    "generic": {"event": None, "props": None, "time": None, "source": [], "transport": []},
}
HEADER = ["tool", "source", "event", "property", "value_type", "rows", "event_rows", "first_seen", "last_seen",
          "values", "pii_hits", "sentinel_hits", "empty_hits", "duplicate_keys", "transport_duplicates", "max_params"]
NUMERIC_RE = re.compile(r"^-?\d+(\.\d+)?$")
ISO_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}")
EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[A-Za-z]{2,}")
# International numbers with a leading +, or a national number written with separators.
# Bare digit runs are left alone, because numeric IDs look the same.
PHONE_RE = re.compile(r"^(\+\d[\d\s().-]{8,18}\d|\(?\d{3}\)?[\s.-]\d{3}[\s.-]\d{4})$")
# An international number inside free text, such as a note or an error message.
PHONE_IN_TEXT_RE = re.compile(r"(?<![\w+])\+\d[\d\s().-]{8,18}\d(?!\d)")
URL_QUERY_RE = re.compile(r"[?&](email|token|password|code|reset|key)=", re.I)
UUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
SENTINELS = {"undefined", "null", "NULL", "NaN", "nan", "[object Object]", "None", "nil"}
# Compared lowercased, before the UUID test. Includes the IDs Amplitude answers with a 400 error and the
# all-zero device ID that merges every opted-out device into one profile.
PLACEHOLDER_IDS = {"null", "undefined", "0", "guest", "anonymous", "", "none", "nil", "unknown", "-1", "n/a", "na",
                   "{}", "nan", "[object object]", "lmy47d", "00000000-0000-0000-0000-000000000000"}
MISSING = object()
TOP_VALUES = 50
RARE_BELOW = 5  # with more than TOP_VALUES distinct values, values seen fewer times fold into <rare>
TOKENS = {"<email>", "<phone>", "<url_query>", "<long_text>", "<array>", "<object>"}


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


def get_path(record, path):
    current = record
    for part in path.split("."):
        if isinstance(current, dict) and part in current:
            current = current[part]
        else:
            return MISSING
    return current


def to_date(value):
    if value is MISSING or value is None or isinstance(value, bool):
        return None
    if isinstance(value, (int, float)) or (isinstance(value, str) and re.fullmatch(r"\d{9,17}(\.\d+)?", value.strip())):
        seconds = float(value)
        if seconds > 1e14:
            seconds /= 1e6  # microseconds
        elif seconds > 1e11:
            seconds /= 1e3  # milliseconds
        try:
            return datetime.fromtimestamp(seconds, tz=timezone.utc).date()
        except (OverflowError, OSError, ValueError):
            return None
    if isinstance(value, str):
        m = re.match(r"(\d{4})-(\d{2})-(\d{2})", value.strip())
        if m:
            try:
                return date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
            except ValueError:
                return None
    return None


def value_type(value):
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "boolean"
    if isinstance(value, int):
        return "integer"
    if isinstance(value, float):
        return "integer" if value.is_integer() else "number"
    if isinstance(value, list):
        return "array"
    if isinstance(value, dict):
        return "object"
    text = str(value).strip()
    if NUMERIC_RE.match(text):
        return "numeric_string"
    if ISO_DATE_RE.match(text):
        return "timestamp"
    return "string"


def looks_personal(text):
    return bool(EMAIL_RE.search(text) or PHONE_RE.match(text.strip()) or PHONE_IN_TEXT_RE.search(text)
                or URL_QUERY_RE.search(text))


def value_label(value):
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, list):
        return "<array>"
    if isinstance(value, dict):
        return "<object>"
    text = repr(value) if isinstance(value, float) else str(value)
    if EMAIL_RE.search(text):
        return "<email>"
    if PHONE_RE.match(text.strip()) or PHONE_IN_TEXT_RE.search(text):
        return "<phone>"
    if URL_QUERY_RE.search(text):
        return "<url_query>"
    if len(text) > 64:
        return "<long_text>"
    text = text.replace("|", "_").replace("=", "_").replace("\n", " ").replace("\r", " ")
    if text[:1] in ("+", "-", "@"):
        text = "'" + text  # keeps spreadsheets from running it as a formula
    return text


def id_class(value):
    if value is MISSING or value is None:
        return "null"
    if isinstance(value, bool):
        return "other"
    if isinstance(value, (int, float)):
        if isinstance(value, float) and not value.is_integer():
            return "other"
        if abs(value) > 2 ** 53:
            return "numeric_over_2_53"
        text = str(int(value))
    else:
        text = str(value).strip()
    if text.lower() in PLACEHOLDER_IDS:
        return "placeholder"
    if EMAIL_RE.search(text):
        return "email_like"
    if UUID_RE.fullmatch(text):
        return "uuid"
    if len(text) < 5:
        return "short_under_5"
    if text.isdigit():
        return "numeric"
    return "other"


def values_text(counter):
    """value=count pairs, most common first; rare values fold into <rare>, the rest of the list into <other>."""
    items, rare = list(counter.items()), 0
    if len(items) > TOP_VALUES:
        kept = []
        for value, count in items:
            if count < RARE_BELOW and value not in TOKENS:
                rare += count
            else:
                kept.append((value, count))
        items = kept
    items.sort(key=lambda x: -x[1])
    parts = [f"{v}={c}" for v, c in items[:TOP_VALUES]]
    rest = sum(c for _, c in items[TOP_VALUES:])
    if rest:
        parts.append(f"<other>={rest}")
    if rare:
        parts.append(f"<rare>={rare}")
    return "|".join(parts)


def iter_text(handle, label, stats):
    char = handle.read(1)
    while char and char.isspace():
        char = handle.read(1)
    if not char:
        return
    if char == "[":
        try:
            data = json.loads(char + handle.read())
        except ValueError as exc:
            fail(f"{label} is not valid JSON: {exc}")
        for record in data:
            yield record
        return
    line = char + handle.readline()
    while line:
        if line.strip():
            try:
                yield json.loads(line)
            except ValueError:
                stats["malformed lines"] += 1
        line = handle.readline()


def iter_records(path, stats):
    lower = path.lower()
    try:
        if lower.endswith(".zip"):
            with zipfile.ZipFile(path) as archive:
                for info in archive.infolist():
                    if info.is_dir():
                        continue
                    with archive.open(info) as raw:
                        head = raw.read(2)
                    with archive.open(info) as raw:
                        stream = gzip.GzipFile(fileobj=raw) if head == b"\x1f\x8b" else raw
                        for record in iter_text(io.TextIOWrapper(stream, encoding="utf-8"), f"{path}:{info.filename}", stats):
                            yield record
        elif lower.endswith(".gz"):
            with gzip.open(path, "rt", encoding="utf-8") as handle:
                for record in iter_text(handle, path, stats):
                    yield record
        else:
            with open(path, encoding="utf-8") as handle:
                for record in iter_text(handle, path, stats):
                    yield record
    except (OSError, zipfile.BadZipFile, UnicodeDecodeError) as exc:
        fail(f"cannot read {path}: {exc}")


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("files", nargs="+")
    parser.add_argument("--format", required=True, choices=sorted(PRESETS))
    parser.add_argument("--tool", required=True, help="tool name for the inventory, for example amplitude")
    parser.add_argument("--event-key")
    parser.add_argument("--props-key")
    parser.add_argument("--time-key")
    parser.add_argument("--source-key")
    parser.add_argument("--since")
    parser.add_argument("--until")
    parser.add_argument("--values-for", default="")
    parser.add_argument("--occurrence-key")
    parser.add_argument("--identity-keys", default="")
    parser.add_argument("--max-records", type=int)
    args = parser.parse_args()

    preset = dict(PRESETS[args.format])
    event_key = args.event_key or preset["event"]
    props_key = args.props_key or preset["props"]
    time_key = args.time_key or preset["time"]
    source_keys = [args.source_key] if args.source_key else preset["source"]
    if not event_key:
        fail("--format generic needs --event-key (and usually --props-key and --time-key)")
    since = to_date(args.since) if args.since else None
    until = to_date(args.until) if args.until else None
    if (args.since and not since) or (args.until and not until):
        fail("--since and --until must be YYYY-MM-DD")
    values_for = {v.strip() for v in args.values_for.split(",") if v.strip()}
    identity_keys = [v.strip() for v in args.identity_keys.split(",") if v.strip()]

    stats = Counter()
    events = defaultdict(lambda: {"rows": 0, "first": None, "last": None, "max_params": 0, "dups": 0, "tdups": 0})
    props = defaultdict(lambda: {"rows": 0, "pii": 0, "sentinel": 0, "empty": 0})
    values = defaultdict(Counter)
    identity = defaultdict(Counter)
    occurrence_seen, transport_seen = defaultdict(set), defaultdict(set)
    first_keys_checked = False

    for path in args.files:
        for record in iter_records(path, stats):
            if args.max_records and stats["records parsed"] >= args.max_records:
                stats["stopped at --max-records"] = 1
                break
            stats["records parsed"] += 1
            if not isinstance(record, dict):
                stats["records that are not objects"] += 1
                continue
            if args.format == "segment" and record.get("type") in ("page", "screen"):
                name = record.get("name") or record.get("type")
            elif args.format == "segment" and record.get("type") not in (None, "track"):
                stats["segment calls that are not track, page or screen"] += 1
                continue
            else:
                name = get_path(record, event_key)
            if not first_keys_checked:
                first_keys_checked = True
                absent = [event_key] if name is MISSING else []
                absent += [k for k in (props_key, time_key) if k and get_path(record, k) is MISSING]
                if source_keys and all(get_path(record, k) is MISSING for k in source_keys):
                    absent.append(" or ".join(source_keys))
                if absent:
                    print(f"note: the first record has no {', '.join(absent)}; its top-level keys (no values) are: "
                          f"{', '.join(sorted(str(k) for k in record))[:300]}", file=sys.stderr)
            if name is MISSING or name in (None, ""):
                stats["records without an event name"] += 1
                continue
            name = str(name)
            properties = get_path(record, props_key) if props_key else {}
            if isinstance(properties, str) and properties.strip().startswith("{"):
                try:
                    properties = json.loads(properties)
                except ValueError:
                    properties = {}
            if not isinstance(properties, dict):
                properties = {}
            source = ""
            for key in source_keys:
                found = get_path(record, key)
                if found not in (MISSING, None, ""):
                    source = str(found)
                    break
            day = to_date(get_path(record, time_key)) if time_key else None
            if (since or until) and day is None:
                stats["records without a readable time (skipped)"] += 1
                continue
            if until and day > until:
                stats["records after --until (ignored)"] += 1
                continue
            ev = events[(source, name)]
            if day:
                ev["first"] = day if ev["first"] is None or day < ev["first"] else ev["first"]
                ev["last"] = day if ev["last"] is None or day > ev["last"] else ev["last"]
            if since and day < since:
                stats["records before --since (first and last seen only)"] += 1
                continue
            stats["records in the window"] += 1
            ev["rows"] += 1
            ev["max_params"] = max(ev["max_params"], len(properties))
            if args.occurrence_key:
                key = get_path(record, args.occurrence_key)
                if key not in (MISSING, None, ""):
                    digest = hash(str(key))
                    if digest in occurrence_seen[name]:
                        ev["dups"] += 1
                    occurrence_seen[name].add(digest)
            if preset["transport"]:
                parts = [get_path(record, k) for k in preset["transport"]]
                if all(p not in (MISSING, None, "") for p in parts[-1:]):
                    digest = hash(tuple(str(p) for p in parts))
                    if digest in transport_seen[name]:
                        ev["tdups"] += 1
                    transport_seen[name].add(digest)
            for prop, value in properties.items():
                vtype = value_type(value)
                bucket = props[(source, name, prop, vtype)]
                bucket["rows"] += 1
                if isinstance(value, str):
                    if looks_personal(value):
                        bucket["pii"] += 1
                    if value in SENTINELS:
                        bucket["sentinel"] += 1
                    elif not value.strip():
                        bucket["empty"] += 1
                if prop in values_for and value is not None:
                    values[(source, name, prop, vtype)][value_label(value)] += 1
            for key in identity_keys:
                identity[(source, name, key)][id_class(get_path(record, key))] += 1
        else:
            continue
        break

    if not events:  # a wrong format or file type parses to nothing; an empty inventory must not look like a clean run
        seen = ", ".join(f"{label} {stats[label]:,}" for label in ("records parsed", "malformed lines", "records without an event name", "records that are not objects") if stats[label])
        fail(f"no events read from {len(args.files)} file(s) ({seen or 'nothing parsed'}): check --format, --event-key and the file type")
    writer = csv.writer(sys.stdout, lineterminator="\n")
    writer.writerow(HEADER)
    for (source, name) in sorted(events, key=lambda k: (k[1], k[0])):
        ev = events[(source, name)]
        writer.writerow([args.tool, source, name, "", "", ev["rows"], "", ev["first"] or "", ev["last"] or "",
                         "", "", "", "", ev["dups"] if args.occurrence_key else "",
                         ev["tdups"] if preset["transport"] else "", ev["max_params"] or ""])
        for (s, n, prop, vtype) in sorted(k for k in props if k[0] == source and k[1] == name):
            bucket = props[(s, n, prop, vtype)]
            top = values.get((s, n, prop, vtype))
            listed = values_text(top) if top else ""
            writer.writerow([args.tool, s, n, prop, vtype, bucket["rows"], ev["rows"], "", "", listed,
                             bucket["pii"], bucket["sentinel"], bucket["empty"], "", "", ""])
    totals = defaultdict(Counter)
    for (source, name, key), classes in identity.items():
        totals[(source, key)].update(classes)
    # One identifier row per event, then a '*' total per source and key; classes only, never the IDs.
    rows_out = [((key, source, name), identity[(source, name, key)]) for (source, name, key) in identity]
    rows_out += [((key, source, "*"), totals[(source, key)]) for (source, key) in totals]
    for (key, source, name), classes in sorted(rows_out, key=lambda r: (r[0][0], r[0][2] == "*", r[0][2], r[0][1])):
        writer.writerow([args.tool, source, name, key, "identifier", sum(classes.values()), sum(classes.values()),
                         "", "", "|".join(f"{c}={n}" for c, n in classes.most_common()), "", "", "", "", "", ""])

    print(f"profiled {len(args.files)} file(s) as {args.format} for tool {args.tool}", file=sys.stderr)
    for label in ("records parsed", "records in the window", "records before --since (first and last seen only)",
                  "records after --until (ignored)", "records without a readable time (skipped)",
                  "records without an event name", "records that are not objects",
                  "segment calls that are not track, page or screen", "malformed lines", "stopped at --max-records"):
        if stats[label]:
            print(f"  {label}: {stats[label]:,}", file=sys.stderr)
    print(f"  events: {len(events):,}; property and type rows: {len(props):,}; values listed for: "
          f"{', '.join(sorted(values_for)) or 'none'}", file=sys.stderr)
    if stats["stopped at --max-records"]:
        print("  the inventory is partial; label it partial in the report", file=sys.stderr)
    return 0


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