""" Reveal-log integrity checker for downloaded oTree CSV exports. Usage: python reveal_log_integrity_checker.py path/to/custom_export_recovery.csv Warnings are recoverable data-quality issues, usually malformed legacy values or incomplete participants. Fatal errors are consistency failures in completed rounds that should be resolved or explicitly accounted for before analysis. The script never writes to the oTree database or modifies CSV inputs. """ import argparse import csv import json import sys from collections import defaultdict VALID_STAGES = {'resume', 'zoom', 'inperson'} VALID_GATES = {'gate1', 'gate2', 'gate3', 'blind_final'} GATE_ALLOWED_STAGES = { 'gate1': {'resume'}, 'gate2': {'resume', 'zoom'}, 'gate3': {'resume', 'zoom', 'inperson'}, 'blind_final': {'resume', 'zoom', 'inperson'}, } def load_json(value, label, problems, severity='warning'): if value in (None, ''): return None try: return json.loads(value) except Exception as exc: problems[severity].append(f'{label}: invalid JSON ({type(exc).__name__}: {exc})') return None def normalize_reveal_payload(value, label, problems): parsed = load_json(value, label, problems, severity='warning') if parsed is None: return {'revealed_cells': [], 'revealed_stages': [], 'reveal_log': []} if isinstance(parsed, list): parsed = {'reveal_log': parsed} if not isinstance(parsed, dict): problems['warning'].append(f'{label}: reveal payload is not an object/list') return {'revealed_cells': [], 'revealed_stages': [], 'reveal_log': []} return { 'revealed_cells': parsed.get('revealed_cells') if isinstance(parsed.get('revealed_cells'), list) else [], 'revealed_stages': parsed.get('revealed_stages') if isinstance(parsed.get('revealed_stages'), list) else [], 'reveal_log': parsed.get('reveal_log') if isinstance(parsed.get('reveal_log'), list) else [], } def visible_labels_from_context(value, label, problems): parsed = load_json(value, label, problems, severity='warning') if not isinstance(parsed, dict): return set() labels = set() for item in parsed.get('visible_candidates', []) or []: if isinstance(item, dict) and item.get('label'): labels.add(str(item['label'])) for key in ('selected_candidates', 'finalists'): for candidate in parsed.get(key, []) or []: labels.add(str(candidate)) for key in ('hired_candidate', 'chosen_candidate'): if parsed.get(key): labels.add(str(parsed[key])) return labels def unique_in_order(values): result = [] for value in values: if value not in result: result.append(value) return result def duration_ms(row, *field_names): for field_name in field_names: value = row.get(field_name) if value in (None, ''): continue try: return float(value) * 1000 except ValueError: return None return None def check_reveal_payload(payload, gate, visible_labels, row_label, problems, page_duration_ms=None): seen_cells = set() last_event_index = 0 last_elapsed = None event_cells = [] event_stages = [] for event in payload['reveal_log']: if not isinstance(event, dict): problems['warning'].append(f'{row_label}: non-object reveal event ignored') continue event_index = event.get('event_index') if not isinstance(event_index, int) or event_index <= last_event_index: problems['fatal'].append(f'{row_label}: event_index is not monotonically increasing') if isinstance(event_index, int): last_event_index = event_index candidate = str(event.get('candidate') or '') stage = str(event.get('stage') or '') cell_key = str(event.get('cell_key') or '') expected_cell_key = f'{candidate}:{stage}' if candidate and stage else '' if not candidate or (visible_labels and candidate not in visible_labels): problems['fatal'].append(f'{row_label}: invalid reveal candidate {candidate!r}') if stage not in VALID_STAGES: problems['fatal'].append(f'{row_label}: invalid reveal stage {stage!r}') if gate not in VALID_GATES: problems['fatal'].append(f'{row_label}: invalid gate {gate!r}') if stage and gate in GATE_ALLOWED_STAGES and stage not in GATE_ALLOWED_STAGES[gate]: problems['fatal'].append(f'{row_label}: stage {stage!r} not allowed for {gate}') if cell_key != expected_cell_key: problems['fatal'].append(f'{row_label}: invalid cell_key {cell_key!r}; expected {expected_cell_key!r}') if event.get('event_type') not in (None, 'reveal'): problems['fatal'].append(f'{row_label}: invalid event_type {event.get("event_type")!r}') if event.get('is_first_reveal_for_cell') is False: problems['fatal'].append(f'{row_label}: reveal event is not marked as first reveal') if cell_key in seen_cells: problems['fatal'].append(f'{row_label}: duplicate first-reveal event for {cell_key}') seen_cells.add(cell_key) event_cells.append(cell_key) event_stages.append(stage) elapsed = event.get('elapsed_ms_on_page') if elapsed is not None: if not isinstance(elapsed, int) or elapsed < 0: problems['fatal'].append(f'{row_label}: elapsed_ms_on_page is negative or non-integer') if last_elapsed is not None and isinstance(elapsed, int) and elapsed < last_elapsed: problems['fatal'].append(f'{row_label}: elapsed times are not nondecreasing') if isinstance(elapsed, int): last_elapsed = elapsed if page_duration_ms is not None and last_elapsed is not None: if page_duration_ms + 250 < last_elapsed: problems['fatal'].append( f'{row_label}: page duration is shorter than last reveal elapsed time' ) expected_cells = unique_in_order(event_cells) if payload['revealed_cells'] != expected_cells: problems['fatal'].append(f'{row_label}: revealed_cells does not match unique reveal_log cell_key values') expected_stages = [stage for stage in ['resume', 'zoom', 'inperson'] if stage in set(event_stages)] if payload['revealed_stages'] != expected_stages: problems['fatal'].append(f'{row_label}: revealed_stages does not match revealed_cells stages') def row_id(row): return ( row.get('session_code', ''), row.get('participant_code', ''), row.get('round_number', ''), ) def completed_standard(row): return (row.get('condition') == 'standard' and bool(row.get('hired_candidate'))) def completed_blind(row): return (row.get('condition') == 'blind' and bool(row.get('hired_candidate'))) def check_row(row, problems): session_code, participant_code, round_number = row_id(row) label = f'session={session_code} participant={participant_code} round={round_number}' if completed_standard(row): required = [ 'gate1_reveal_log', 'gate2_reveal_log', 'gate3_reveal_log', 'gate1_decision_context', 'gate2_decision_context', 'hire_decision_context', ] for field in required: if not row.get(field): problems['fatal'].append(f'{label}: completed Standard round missing {field}') gate_contexts = { 'gate1': row.get('gate1_decision_context', ''), 'gate2': row.get('gate2_decision_context', ''), 'gate3': row.get('hire_decision_context', ''), } gate_logs = { 'gate1': row.get('gate1_reveal_log', ''), 'gate2': row.get('gate2_reveal_log', ''), 'gate3': row.get('gate3_reveal_log', ''), } gate_durations = { 'gate1': duration_ms(row, 'main_gate1_secs', 'practice_gate1_secs'), 'gate2': duration_ms(row, 'main_gate2_secs', 'practice_gate2_secs'), 'gate3': duration_ms(row, 'main_gate3_secs', 'practice_gate3_secs'), } for gate, log_value in gate_logs.items(): visible = visible_labels_from_context(gate_contexts[gate], f'{label} {gate} context', problems) payload = normalize_reveal_payload(log_value, f'{label} {gate} reveal log', problems) check_reveal_payload(payload, gate, visible, f'{label} {gate}', problems, gate_durations[gate]) elif completed_blind(row): if not row.get('reveal_log'): problems['fatal'].append(f'{label}: completed Blind round missing reveal_log') if not row.get('hire_decision_context'): problems['fatal'].append(f'{label}: completed Blind round missing hire_decision_context') visible = visible_labels_from_context(row.get('hire_decision_context', ''), f'{label} blind context', problems) payload = normalize_reveal_payload(row.get('reveal_log', ''), f'{label} blind reveal log', problems) check_reveal_payload( payload, 'blind_final', visible, f'{label} blind_final', problems, duration_ms(row, 'main_stage3_secs', 'practice_stage3_secs'), ) else: problems['warning'].append(f'{label}: incomplete or non-final row; reveal completeness not enforced') def main(argv=None): parser = argparse.ArgumentParser(description='Check reveal-log integrity in downloaded oTree CSV exports.') parser.add_argument('csv_path') args = parser.parse_args(argv) problems = {'warning': [], 'fatal': []} row_count = 0 participants = set() with open(args.csv_path, newline='', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: row_count += 1 participants.add((row.get('session_code', ''), row.get('participant_code', ''))) check_row(row, problems) print(f'Checked rows: {row_count}') print(f'Participants: {len(participants)}') print(f'Warnings: {len(problems["warning"])}') print(f'Fatal errors: {len(problems["fatal"])}') for severity in ('fatal', 'warning'): if problems[severity]: print(f'\n{severity.upper()}:') for item in problems[severity]: print(f'- {item}') return 1 if problems['fatal'] else 0 if __name__ == '__main__': sys.exit(main())