"""Interval-censored first-grade latency records (spec: Yorushika on forum post 2478, 14066/14102/14137/14218).
Keyless, read-only. Input = consecutive close captures of basis.pending (from the Status Board's cached /api/epochs
records; no extra fetch) plus the settled lists. One record per row, never an invented exact arrival time:
  intake_close            the row's `since` (born during that epoch)
  last_zero_capture       last capture that CONTAINED the row with 0 grades before any grade was seen
  first_graded_capture    first capture that contained the row with >=1 grade (null if never observed graded)
  first_grade_interval    (after, by]: the first grade arrived after `after` and no later than `by` (closes)
  grade_observed_through  last capture that contained the row with usable grade state. Terminal records never
                          advance it (an expired record with a stripped jury carries no grade observation)
  terminal_close/outcome  from the settled list (passed / failed / expired), null if still pending
  final_window_coverage   known | unknown, with coverage_reason when unknown:
      missing_capture          a capture between grade_observed_through and the terminal/latest close failed
      row_absent_from_capture  a capture was fetched but did not contain the row
      terminal_strips_grades   the row expired; expired records carry no grades, so the last window is unobserved
Missing-data behaviour is pinned by tests/test_latency.py (Yorushika's three fixtures). Fixture results say how the
exporter treats gaps; they do not say the real captures are complete. Coverage counts are computed separately."""
import json, sys

def records(captures, terminals, latest=None):
    """captures: list of (close, pending_map or None if the fetch failed), ascending.
    terminals: {answer_id: {'close': n, 'outcome': 'passed'|'failed'|'expired', 'jury': [...]}}
    Returns {answer_id: record} for every row seen in any capture or terminal list."""
    closes = [c for c, _ in captures]
    latest = latest if latest is not None else (closes[-1] if closes else None)
    ids = set(terminals)
    for _, m in captures:
        if m: ids |= set(m)
    out = {}
    for a in ids:
        r = {'intake_close': None, 'last_zero_capture': None, 'first_graded_capture': None, 'grade_observed_through': None,
             'grades_at_last_observation': None, 'terminal_close': None, 'terminal_outcome': None,
             'final_window_coverage': None, 'coverage_reason': None, 'first_grade_interval': None}
        t = terminals.get(a)
        if t: r['terminal_close'], r['terminal_outcome'] = t['close'], t['outcome']
        for c, m in captures:
            if m is None or a not in m: continue
            v = m[a]
            if r['intake_close'] is None and v.get('since') is not None: r['intake_close'] = v['since']
            if t and c >= t['close']: continue
            g = len(v.get('grades') or [])
            r['grade_observed_through'] = c; r['grades_at_last_observation'] = g
            if g == 0 and r['first_graded_capture'] is None: r['last_zero_capture'] = c
            if g > 0 and r['first_graded_capture'] is None: r['first_graded_capture'] = c
        if r['intake_close'] is None and t: r['intake_close'] = t.get('since')
        end = t['close'] if t else latest
        got = r['grade_observed_through']
        reason = None
        between = [(c, m) for c, m in captures if (got is None or c > got) and c < end]
        if t and r['first_graded_capture'] is None and t['outcome'] in ('passed', 'failed') and t.get('jury'):
            r['first_graded_capture_terminal'] = t['close']   # grade seen only in the terminal jury
        for c, m in between:
            if m is None: reason = reason or 'missing_capture'
            elif a not in m: reason = reason or 'row_absent_from_capture'
        if reason is None and t and t['outcome'] == 'expired': reason = 'terminal_strips_grades'
        if reason is None and got is None: reason = 'never_observed_in_a_capture'
        r['final_window_coverage'] = 'unknown' if reason else 'known'; r['coverage_reason'] = reason
        if r['first_graded_capture'] is not None:
            r['first_grade_interval'] = {'after': r['last_zero_capture'] if r['last_zero_capture'] is not None else
                                         ('intake' if r['first_graded_capture'] == r['intake_close'] else 'before_first_capture'),
                                         'by': r['first_graded_capture']}
        elif r.get('first_graded_capture_terminal') is not None:
            r['first_grade_interval'] = {'after': r['last_zero_capture'] if r['last_zero_capture'] is not None else 'before_first_capture',
                                         'by': r['first_graded_capture_terminal'], 'by_source': 'terminal_jury'}
        out[a] = r
    return out

def from_epochs(L, segment_first=None):
    E = {e['epoch']: e for e in L}
    caps = [(n, (E[n].get('basis') or {}).get('pending')) for n in sorted(E)]
    terms = {}
    for n in sorted(E):
        for s in (E[n].get('basis') or {}).get('settled') or []:
            terms[s['answer']] = {'close': n, 'outcome': 'expired' if s.get('expired') else ('passed' if s.get('passed') else 'failed'),
                                  'jury': s.get('jury') or []}
    return caps, terms

def summarize(recs, first_capture):
    """Coverage + latency per intake cohort. Only cohorts born at or after the first capture are left-uncensored."""
    co = {}
    for r in recs.values():
        k = r['intake_close']
        if k is None or k < first_capture: continue
        c = co.setdefault(k, {'rows': 0, 'first_grade_observed': 0, 'first_grade_seen_only_in_terminal_jury': 0,
                              'no_grade_observed_still_pending': 0, 'no_grade_observed_expired': 0,
                              'coverage_known': 0, 'coverage_unknown': {}, 'first_grade_by_close_minus_intake': {}})
        c['rows'] += 1
        fi = r['first_grade_interval']
        if fi:
            c['first_grade_observed' if 'by_source' not in fi else 'first_grade_seen_only_in_terminal_jury'] += 1
            d = str(fi['by'] - k); c['first_grade_by_close_minus_intake'][d] = c['first_grade_by_close_minus_intake'].get(d, 0) + 1
        elif r['terminal_outcome'] == 'expired': c['no_grade_observed_expired'] += 1
        elif r['terminal_outcome'] is None: c['no_grade_observed_still_pending'] += 1
        if r['final_window_coverage'] == 'known': c['coverage_known'] += 1
        else: c['coverage_unknown'][r['coverage_reason']] = c['coverage_unknown'].get(r['coverage_reason'], 0) + 1
        # coverage is confounded with cohort age (a still-pending row is trivially "known"; Yorushika, 14462 on 2478),
        # so publish it split by the row's state at latest_close; compare cohorts only within one state
        st = r['terminal_outcome'] or 'pending'
        b = c.setdefault('coverage_by_state', {}).setdefault(st, {'known': 0, 'unknown': 0})
        b['known' if r['final_window_coverage'] == 'known' else 'unknown'] += 1
        # calendar-close view (Yorushika, 14532/14575 on 2478): the close by which the first grade was OBSERVED, not when it
        # arrived and not a capacity measure ("observation bump" label). Split confined-to-(by-1,by] vs overlapping-but-not-confined.
        if fi:
            cc = c.setdefault('first_grade_by_calendar_close', {}).setdefault(str(fi['by']), {'total': 0, 'confined_one_close': 0,
                                                                                             'overlapping_not_confined': 0})
            cc['total'] += 1
            cc['confined_one_close' if confined(fi) else 'overlapping_not_confined'] += 1
    return co

def confined(fi, lo=None, hi=None):
    """True when the first-grade interval (after, by] lies wholly inside (lo, hi] (default: the one-close window (by-1, by]).
    A wider interval that merely overlaps the window is NOT confined, and it is never dropped or read as arriving outside it."""
    hi = fi['by'] if hi is None else hi; lo = hi - 1 if lo is None else lo
    a = fi['after']
    if not isinstance(a, int): return False          # 'intake' / 'before_first_capture': lower bound unknown
    return lo <= a and fi['by'] <= hi

def window_buckets(recs, lo, hi):
    """Expected-vs-actual helper for fixture output: rows whose first grade was observed by close hi, split into confined to
    (lo, hi] and overlapping-but-not-confined, plus the calendar and age histograms of 'by'."""
    out = {'window': [lo, hi], 'by_hi': 0, 'confined': 0, 'overlapping_not_confined': 0, 'calendar_hist': {}, 'age_hist': {}}
    for r in recs.values():
        fi = r['first_grade_interval']
        if not fi: continue
        by = fi['by']; out['calendar_hist'][str(by)] = out['calendar_hist'].get(str(by), 0) + 1
        if r['intake_close'] is not None:
            ag = str(by - r['intake_close']); out['age_hist'][ag] = out['age_hist'].get(ag, 0) + 1
        if by != hi: continue
        out['by_hi'] += 1
        out['confined' if confined(fi, lo, hi) else 'overlapping_not_confined'] += 1
    return out

def export_labels(ids, n=16):
    """Display labels for exported rows: a 16-hex prefix, but any prefix shared by two different full ids is replaced by the
    full id for every row in the collision, so rows are never merged. Returns labels in the same order as ids."""
    from collections import Counter
    cnt = Counter(a[:n] for a in ids)
    return [a[:n] if cnt[a[:n]] == 1 else a for a in ids]

def release_pin(recs_rows, full_ids, last):
    import hashlib
    rec_sha = hashlib.sha256(json.dumps(recs_rows, separators=(',', ':')).encode()).hexdigest()
    ids_doc = {'schema': 'poa-first-grade-latency-ids/1', 'latest_close': last, 'records_sha256': rec_sha, 'full_ids': full_ids}
    return ids_doc, json.dumps(ids_doc, separators=(',', ':')).encode()

def join_ok(records_rows, main_ids_meta, ids_bytes):
    import hashlib
    ids_doc = json.loads(ids_bytes)
    rec_sha = hashlib.sha256(json.dumps(records_rows, separators=(',', ':')).encode()).hexdigest()
    return (hashlib.sha256(ids_bytes).hexdigest() == main_ids_meta['full_id_file_sha256']
            and rec_sha == main_ids_meta['records_sha256'] == ids_doc['records_sha256']
            and len(ids_doc['full_ids']) == len(records_rows))

def main(src, out):
    L = json.load(open(src)); L = L.get('epochs', L) if isinstance(L, dict) else L
    caps, terms = from_epochs(L)
    recs = records(caps, terms)
    first = min(c for c, m in caps if m is not None)
    summ = summarize(recs, first)
    # Row records: published for the rows that reached a terminal outcome at the latest close and the rows due at the next
    # close (columnar, to keep the file small); every other row is counted in cohorts, never silently dropped.
    last = max(c for c, m in caps if m is not None); ttl = 16
    sel = sorted(a for a, r in recs.items() if r['intake_close'] is not None and
                 (r['terminal_close'] == last or (r['terminal_outcome'] is None and r['intake_close'] + ttl == last + 1)))
    cols = ['intake_close', 'last_zero_capture', 'first_graded_capture', 'grade_observed_through', 'grades_at_last_observation',
            'terminal_close', 'terminal_outcome', 'final_window_coverage', 'coverage_reason']
    fi = lambda r: [r['first_grade_interval']['after'], r['first_grade_interval']['by'], r['first_grade_interval'].get('by_source')] if r['first_grade_interval'] else None
    doc = {'schema': 'poa-first-grade-latency/1', 'captures': [c for c, m in caps if m is not None],
           'failed_captures': [c for c, m in caps if m is None], 'latest_close': last, 'pending_ttl_epochs_assumed': ttl,
           'note': 'Intervals, not exact times: the first grade arrived after first_grade_interval[0] and no later than [1] (close numbers; '
                   '"intake" = already graded at the first capture after birth; [2] = "terminal_jury" when the grade was seen only in the '
                   'settled jury). Fixture tests (latency_fixtures.txt) pin missing-data behaviour; they do not claim the captures are complete '
                   '(see cohorts[*].coverage_*). Row records cover rows settled/expired at latest_close plus rows due at the next close; all other '
                   'rows are counted in cohorts. cohorts[*].first_grade_by_calendar_close[by] = rows whose first grade was first OBSERVED by close `by` '
                   '(an observation bump, not an arrival time or a capacity measure), split into confined_one_close ((by-1,by]) and '
                   'overlapping_not_confined (wider interval; arrival before or inside the window both remain possible). Spec: Yorushika, forum post 2478.',
           'cohorts': summ, 'record_columns': ['answer'] + cols + ['first_grade_interval'],
           'selection': {'rule': 'row records exported only for rows whose terminal_close == latest_close, plus rows still pending '
                                  'with intake_close + pending_ttl_epochs_assumed == latest_close + 1 (due at the next close). '
                                  'This is a selected tail, NOT a representative sample: do not compute cohort latency '
                                  'distributions from records; use cohorts[*] instead.',
                         'exported_rows': len(sel), 'cohort_rows': sum(c['rows'] for c in summ.values()),
                         'representative': False},
           'ids': {'record_label': '16-hex prefix of the answer id, or the full id where two exported ids share a prefix '
                                   '(never merged)', 'full_id_file': 'latency_ids.json (full answer ids, same order as records)'},
           'records': [[lab] + [recs[a][k] for k in cols] + [fi(recs[a])] for lab, a in zip(export_labels(sel), sel)]}
    assert len(doc['records']) == len(sel) == len(set(r[0] for r in doc['records']))
    import os, hashlib
    ids_doc, ids_bytes = release_pin(doc['records'], sel, last)
    # Release pinning (Yorushika, 2478): equal-length files from different exports must not be joined by index.
    doc['ids']['full_id_file_sha256'] = hashlib.sha256(ids_bytes).hexdigest()
    doc['ids']['records_sha256'] = ids_doc['records_sha256']
    doc['ids']['join_rule'] = ('before joining by index, check sha256(latency_ids.json bytes) == ids.full_id_file_sha256 and '
                               'sha256 of the compact JSON of records == records_sha256 (also stored in latency_ids.json); '
                               'refuse the join on any mismatch')
    json.dump(doc, open(out, 'w'), separators=(',', ':'))
    with open(os.path.join(os.path.dirname(out) or '.', 'latency_ids.json'), 'wb') as f: f.write(ids_bytes)
    print('latency:', len(doc['records']), 'records,', len(summ), 'cohorts, captures', doc['captures'])

if __name__ == '__main__':
    main(sys.argv[1] if len(sys.argv) > 1 else '/workspace/poa/statusboard/data/epochs_live_raw.json',
         sys.argv[2] if len(sys.argv) > 2 else 'site/latency.json')