scripts/validate_events.py

"""Validate an event-log JSONL file against the official sample_events.jsonl schema.

The challenge's held-out set (and our submitted event log) uses a multi-source
schema with three event families:

  entry / exit                  -> id_token, store_code, event_timestamp, is_staff, demographics
  zone_entered / zone_exited    -> track_id, store_id, zone_id, event_time, zone_* , hotspots
  queue_completed / queue_abandoned -> queue_event_id, queue_*_ts, wait_seconds, abandoned

Usage:
    python scripts/validate_events.py data/sample_events.jsonl [more.jsonl ...]

Exit code is non-zero if any HARD error is found (bad JSON, unknown event_type,
missing required field, unparseable timestamp). "Recommended" fields that the
official sample always carries (camera_id, demographics, zone hotspots, ...) are
reported as WARNINGS only — they enrich the stream but are not required to parse.
"""
from __future__ import annotations

import json
import sys
from datetime import datetime
from pathlib import Path

# Windows consoles default to cp1252 and choke on the ✓/⚠/✗ glyphs below.
try:
    sys.stdout.reconfigure(encoding="utf-8")
except (AttributeError, ValueError):
    pass

OFFICIAL_TYPES = {
    "entry", "exit", "zone_entered", "zone_exited",
    "queue_completed", "queue_abandoned",
}

# required = must be present & non-null to be a valid event of this family
# recommended = present in the official sample; absence is a warning, not an error
SCHEMA = {
    "entry": {
        "required": {"event_type", "id_token", "store_code", "event_timestamp", "is_staff"},
        "recommended": {"camera_id", "gender_pred", "age_pred", "age_bucket",
                        "is_face_hidden", "group_id", "group_size"},
        "ts": ["event_timestamp"],
    },
    "exit": {
        "required": {"event_type", "id_token", "store_code", "event_timestamp", "is_staff"},
        "recommended": {"camera_id", "gender_pred", "age_pred", "age_bucket",
                        "is_face_hidden", "group_id", "group_size"},
        "ts": ["event_timestamp"],
    },
    "zone_entered": {
        "required": {"event_type", "store_id", "zone_id", "event_time"},
        "id_one_of": {"track_id", "id_token"},
        "recommended": {"camera_id", "zone_name", "zone_type", "is_revenue_zone",
                        "zone_hotspot_x", "zone_hotspot_y", "gender", "age", "age_bucket"},
        "ts": ["event_time"],
    },
    "zone_exited": {
        "required": {"event_type", "store_id", "zone_id", "event_time"},
        "id_one_of": {"track_id", "id_token"},
        "recommended": {"camera_id", "zone_name", "zone_type", "is_revenue_zone",
                        "zone_hotspot_x", "zone_hotspot_y", "gender", "age", "age_bucket"},
        "ts": ["event_time"],
    },
    "queue_completed": {
        "required": {"queue_event_id", "event_type", "store_id", "zone_id",
                     "queue_join_ts", "abandoned"},
        "id_one_of": {"track_id", "id_token"},
        "recommended": {"camera_id", "zone_name", "zone_type", "is_revenue_zone",
                        "queue_served_ts", "queue_exit_ts", "wait_seconds",
                        "queue_position_at_join", "zone_hotspot_x", "zone_hotspot_y",
                        "gender", "age", "age_bucket"},
        "ts": ["queue_join_ts"],
    },
    "queue_abandoned": {
        "required": {"queue_event_id", "event_type", "store_id", "zone_id",
                     "queue_join_ts", "abandoned"},
        "id_one_of": {"track_id", "id_token"},
        "recommended": {"camera_id", "zone_name", "zone_type", "is_revenue_zone",
                        "queue_served_ts", "queue_exit_ts", "wait_seconds",
                        "queue_position_at_join", "zone_hotspot_x", "zone_hotspot_y",
                        "gender", "age", "age_bucket"},
        "ts": ["queue_join_ts"],
    },
}


def _parse_ts(v: object) -> bool:
    if not isinstance(v, str) or not v:
        return False
    try:
        datetime.fromisoformat(v.replace("Z", "+00:00"))
        return True
    except ValueError:
        return False


def validate_file(path: Path) -> tuple[int, int, int]:
    """Returns (n_events, n_errors, n_warnings) and prints a report for `path`."""
    errors: list[str] = []
    warnings: list[str] = []
    type_counts: dict[str, int] = {}
    missing_recommended: dict[str, int] = {}
    n = 0

    if not path.exists():
        print(f"  ✗ file not found: {path}")
        return (0, 1, 0)

    for lineno, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
        if not raw.strip():
            continue
        n += 1
        try:
            ev = json.loads(raw)
        except json.JSONDecodeError as exc:
            errors.append(f"line {lineno}: invalid JSON ({exc})")
            continue
        if not isinstance(ev, dict):
            errors.append(f"line {lineno}: not a JSON object")
            continue

        et = ev.get("event_type")
        type_counts[et] = type_counts.get(et, 0) + 1
        if et not in OFFICIAL_TYPES:
            errors.append(f"line {lineno}: unknown event_type {et!r} "
                          f"(expected one of {sorted(OFFICIAL_TYPES)})")
            continue

        rules = SCHEMA[et]
        for f in rules["required"]:
            if f not in ev or ev[f] is None:
                errors.append(f"line {lineno} [{et}]: missing required field {f!r}")
        if "id_one_of" in rules and not (rules["id_one_of"] & {k for k, v in ev.items() if v is not None}):
            errors.append(f"line {lineno} [{et}]: needs one of "
                          f"{sorted(rules['id_one_of'])} for visitor identity")
        for f in rules["ts"]:
            if f in ev and ev[f] is not None and not _parse_ts(ev[f]):
                errors.append(f"line {lineno} [{et}]: unparseable timestamp {f}={ev[f]!r}")
        for f in rules["recommended"]:
            if f not in ev:
                missing_recommended[f] = missing_recommended.get(f, 0) + 1

    # roll missing-recommended into warnings (one line per field)
    for f, c in sorted(missing_recommended.items()):
        warnings.append(f"recommended field {f!r} absent on {c} event(s)")

    print(f"\n{'='*70}\n{path}")
    print(f"  events: {n}   types: " +
          ", ".join(f"{k}={v}" for k, v in sorted(type_counts.items(), key=lambda x: str(x[0]))))
    if errors:
        print(f"  ✗ {len(errors)} ERROR(S):")
        for e in errors[:25]:
            print(f"      - {e}")
        if len(errors) > 25:
            print(f"      … and {len(errors) - 25} more")
    else:
        print("  ✓ no hard errors — valid JSONL, all events match the official schema")
    if warnings:
        print(f"  ⚠ {len(warnings)} warning(s) (enrichment fields the sample carries):")
        for w in warnings:
            print(f"      - {w}")
    return (n, len(errors), len(warnings))


def main(argv: list[str]) -> int:
    files = argv or ["data/sample_events.jsonl"]
    total_err = 0
    total_events = 0
    for f in files:
        n, e, _w = validate_file(Path(f))
        total_events += n
        total_err += e
    print(f"\n{'='*70}\nTOTAL: {total_events} events across {len(files)} file(s), "
          f"{total_err} hard error(s) → {'PASS ✓' if total_err == 0 else 'FAIL ✗'}")
    return 0 if total_err == 0 else 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))