#!/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()