Lumotia/scripts/parse-activation-log.py

#!/usr/bin/env python3
"""Summarise a Lumotia activation log for tester review. Stdlib only.

Usage:
  python3 scripts/parse-activation-log.py path/to/tester-activation.json
  python3 scripts/parse-activation-log.py --paste   # reads stdin (tab/pipe table or JSON)
"""
import json, re, sys
from datetime import datetime, timezone
from pathlib import Path

# Activation metric thresholds (v0.1 tester runbook)
WARMUP_MINUTES  = 3    # first capture < N min from open = activated
CORE_VALUE_COUNT = 3   # captures in first 24h
DAY = 86400

# Event kinds that count as "capture completed"
CAPTURE_KINDS = {
    'first_capture', 'recording_completed', 'recording_saved',
    'capture_completed', 'transcript_saved',
}

# Deny patterns — scrub payload strings before printing anything
_DENY = re.compile(
    r'transcripts?/|captures?/|audio/|\.wav\b|\.mp3\b|\.opus\b|\.ogg\b|\.flac\b|\.db\b',
    re.IGNORECASE,
)

def _utc(ts): return datetime.fromtimestamp(ts, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')

def _delta(a, b):
    d = abs(b - a)
    if d < 60:   return f"{d}s"
    if d < 3600: return f"{d//60} min"
    if d < DAY:  return f"{d//3600}h {(d%3600)//60}m"
    return f"{d//DAY}d {(d%DAY)//3600}h"

def _parse_json(text):
    data = json.loads(text)
    if not isinstance(data, list): raise ValueError("Expected JSON array")
    return data

def _parse_table(text):
    """Best-effort: extract rows from a pipe/tab-delimited table or HTML paste."""
    try: return _parse_json(text)
    except (json.JSONDecodeError, ValueError): pass
    html = re.compile(r'<[^>]+>')
    events = []
    for line in text.splitlines():
        line = html.sub('', line).strip()
        cells = [c.strip() for c in re.split(r'\||\t', line) if c.strip()]
        if len(cells) < 3 or cells[0].lower() in ('id', '#'): continue
        try:
            raw_ts = cells[2]
            if re.match(r'^\d+$', raw_ts):
                ts = int(raw_ts)
            else:
                ts = int(datetime.fromisoformat(raw_ts.replace('Z','+00:00'))
                          .astimezone(timezone.utc).timestamp())
            events.append({'id': int(cells[0]), 'kind': cells[1], 'occurred_at': ts,
                           'payload': cells[3] if len(cells) > 3 else None})
        except (ValueError, IndexError):
            continue
    return events

def _metrics(events):
    if not events: return {}
    ev = sorted(events, key=lambda e: e.get('occurred_at', 0))
    by_kind = {}
    for e in ev: by_kind.setdefault(e.get('kind',''), []).append(e)

    first_ts, last_ts = ev[0]['occurred_at'], ev[-1]['occurred_at']

    def _first(*kinds):
        for e in ev:
            if e.get('kind') in set().union(*kinds): return e['occurred_at']
        return None

    fc = _first({'first_capture'}, CAPTURE_KINDS)
    fe = _first({'first_export','transcript_exported','export_completed'})
    ft = _first({'first_task_extract','task_extracted','tasks_extracted'})

    cap_24h = sum(1 for e in ev if e['kind'] in CAPTURE_KINDS and e['occurred_at']-first_ts <= DAY)
    cap_7d  = sum(1 for e in ev if e['kind'] in CAPTURE_KINDS and e['occurred_at']-first_ts <= 7*DAY)
    returned = (last_ts - first_ts) <= 7*DAY if last_ts != first_ts else None
    return dict(ev=ev, by_kind=by_kind, first_ts=first_ts, last_ts=last_ts,
                fc=fc, fe=fe, ft=ft, cap_24h=cap_24h, cap_7d=cap_7d, returned=returned)

def _render(m, label):
    if not m: return "ERROR: no events found — nothing to summarise.\n"
    L = []
    hdr = f"LUMOTIA ACTIVATION LOG SUMMARY ({label})"
    L += [hdr, '='*len(hdr), '']
    L.append(f"Total events:         {len(m['ev'])}")
    L.append(f"First event:          {_utc(m['first_ts'])}")
    L.append(f"Last event:           {_utc(m['last_ts'])}  (span: {_delta(m['first_ts'], m['last_ts'])})")
    L.append('')

    def _ms(label, ts, ref=None):
        if ts is None: return f"{label:<22}  (not recorded)"
        s = f"{label:<22}  {_utc(ts)}"
        if ref and ref != ts: s += f"  ({_delta(ref, ts)} after first capture)"
        return s

    L += [_ms("First capture:", m['fc']),
          _ms("First export:", m['fe'], m['fc']),
          _ms("First task extract:", m['ft'], m['fc']), '']
    L.append(f"Captures (24h):       {m['cap_24h']}   (target: >= {CORE_VALUE_COUNT})")
    L.append(f"Captures (7-day):     {m['cap_7d']}")
    ret = m['returned']
    if ret is True:  ret_s = f"yes  (last event {_utc(m['last_ts'])})"
    elif ret is False: ret_s = f"no   (last seen {_utc(m['last_ts'])} — {_delta(m['first_ts'],m['last_ts'])} after first)"
    else:            ret_s = "n/a  (only one event recorded)"
    L += [f"Returned within 7d:   {ret_s}", '', "ACTIVATION METRICS:"]

    def ok(flag, label, detail):
        t = "  ✓" if flag else "  ×"
        return f"{t} {label}: {detail}"
    def q(label, detail): return f"  ? {label}: {detail}"

    L.append(ok(m['fc'] is not None, "Activation",
                "first_capture event present" if m['fc'] else "no capture event recorded"))
    L.append(ok(m['cap_24h'] >= CORE_VALUE_COUNT, "Core value",
                f"{m['cap_24h']} captures in first 24h (target: >= {CORE_VALUE_COUNT})"))
    if ret is True:   L.append(ok(True,  "Retention", "returned within 7 days"))
    elif ret is False: L.append(ok(False, "Retention", "no return event within 7 days"))
    else:              L.append(q("Retention", "only one session — check day-3"))
    L.append(q("Quality", "extracted tasks accepted/edited (not in activation log — ask tester)"))
    L.append(q("Trust",   "can articulate 'what stays local' (not in activation log — ask tester)"))
    L += ['', "EVENT BREAKDOWN:"]
    for kind, evs in sorted(m['by_kind'].items()):
        L.append(f"  {kind:<35}  x{len(evs)}")
    L += ['', "NOTE: ? items require qualitative follow-up per the tester-onboarding-kit."]
    return '\n'.join(L) + '\n'

def main():
    args = sys.argv[1:]
    if not args: print(__doc__); sys.exit(1)
    if args[0] == '--paste':
        raw, label = sys.stdin.read(), 'stdin (paste)'
        parse = _parse_table
    else:
        p = Path(args[0])
        if not p.exists(): print(f"ERROR: file not found: {p}", file=sys.stderr); sys.exit(2)
        raw, label = p.read_text(encoding='utf-8', errors='replace'), p.name
        parse = _parse_table  # tries JSON first, then table
    try:
        events = parse(raw)
    except Exception as exc:
        print(f"ERROR: could not parse input — {exc}", file=sys.stderr); sys.exit(2)
    for ev in events:
        if isinstance(ev.get('payload'), str):
            ev['payload'] = _DENY.sub('[redacted]', ev['payload'])
    print(_render(_metrics(events), label))

if __name__ == '__main__':
    main()