"""Offline frozen comparison: python3 products/apibench/compare.py. No vendor calls."""
from collections import Counter
from datetime import datetime
import hashlib
import json
import math
from pathlib import Path
import statistics

ROOT = Path(__file__).resolve().parent
VENDORS = ('scraperapi', 'scrapingbee')
START = '2026-10-18T00:00:00+00:00'
END = '2026-10-19T05:08:06.276135+00:00'
RAW = ROOT / 'evidence/results-2026-10-19.jsonl'
OUTPUT = ROOT / 'evidence/scraperapi-vs-scrapingbee-2026-10-19.json'


def utc(value):
    stamp = datetime.fromisoformat(value.replace('Z', '+00:00'))
    if stamp.utcoffset() is None:
        raise ValueError('UTC offset required')
    return stamp


def metrics(rows):
    unavailable = [r for r in rows if r.get('result') == 'unavailable' or
                   (r['mode'] == 'premium' and r['http_status'] in (401, 402, 403))]
    evaluated = [r for r in rows if r not in unavailable]
    passed = sum(isinstance(r['http_status'], int) and 200 <= r['http_status'] < 300
                 and r['validation_passed'] is True for r in evaluated)
    latencies = sorted(r['seconds'] for r in rows)
    return dict(attempts=len(rows), evaluated=len(evaluated), valid=passed,
                failures=len(evaluated) - passed, unavailable=len(unavailable),
                success_rate=passed / len(evaluated) if evaluated else None,
                http_200=sum(r['http_status'] == 200 for r in rows),
                http_200_content_failures=sum(r['http_status'] == 200 and
                                             not r['validation_passed'] for r in rows),
                median_seconds=statistics.median(latencies) if rows else None,
                p90_seconds=latencies[math.ceil(.9 * len(rows)) - 1] if rows else None,
                first_capture=min((r['timestamp'] for r in rows), default=None),
                last_capture=max((r['timestamp'] for r in rows), default=None))


def compare(rows, targets, start=START, end=END):
    """Match target+mode+URL, retain every eligible attempt, never pair on outcome."""
    definitions = {t['id']: t for t in targets}
    candidates, excluded = [], []
    for line, row in enumerate(rows, 1):
        reason = None
        target = definitions.get(row['target'])
        if row['vendor'] not in VENDORS:
            reason = 'other_vendor'
        elif not utc(start) <= utc(row['timestamp']) < utc(end):
            reason = 'outside_window'
        elif row.get('result') == 'timeout_60s_old':
            reason = 'legacy_premium_60s_timeout'  # protocol changed to 150s
        elif (not target or row['url'] != target['url'] or
              row['mode'] not in ([target['mode'], 'premium'] if target.get('challenge')
                                  else [target['mode']])):
            reason = 'unknown_target_url_or_mode'
        elif not isinstance(row.get('validation_passed'), bool):
            reason = 'missing_content_validation'
        if reason:
            excluded.append(dict(source_line=line, reason=reason, observation=row))
        else:
            candidates.append((line, row))
    key = lambda r: (r['target'], r['mode'], r['url'])
    keys = {v: {key(r) for _, r in candidates if r['vendor'] == v} for v in VENDORS}
    shared = keys[VENDORS[0]] & keys[VENDORS[1]]
    matched = []
    for line, row in candidates:
        if key(row) in shared:
            matched.append(dict(source_line=line, observation=row))
        else:
            excluded.append(dict(source_line=line, reason='unmatched_stratum', observation=row))
    observations = [x['observation'] for x in matched]
    strata = []
    for target, mode, url in sorted(shared):
        strata.append(dict(target=target, mode=mode, url=url,
                           validation=definitions[target]['validation'],
                           vendors={v: metrics([r for r in observations if key(r) == (target, mode, url)
                                                and r['vendor'] == v]) for v in VENDORS}))
    return dict(window=dict(start_inclusive=start, end_exclusive=end),
                matching='target ID + exact URL + exact mode; all eligible attempts in window',
                success_definition='2xx AND recorded visible-text validation_passed; not semantic correctness',
                latency_population='all eligible matched attempts, including failures and unavailable; nearest-rank p90',
                rate_denominator='evaluated = eligible matched attempts minus unavailable; zero -> null',
                exclusions=dict(Counter(x['reason'] for x in excluded)),
                strata=strata,
                totals={mode: {v: metrics([r for r in observations if r['vendor'] == v and
                                         ((r['mode'] != 'premium') if mode == 'standard' else
                                          (r['mode'] == 'premium'))]) for v in VENDORS}
                        for mode in ('standard', 'premium')},
                matched_observations=matched, excluded_observations=excluded,
                input_observations=len(rows), matched_attempts=len(matched))


def main():
    rows = [json.loads(line) for line in RAW.read_text().splitlines() if line.strip()]
    result = compare(rows, json.loads((ROOT / 'targets.json').read_text()))
    result['evidence_date'] = '2026-10-19'
    result['inputs'] = {str(p.relative_to(ROOT)): hashlib.sha256(p.read_bytes()).hexdigest()
                       for p in (RAW, ROOT / 'targets.json', ROOT / 'run.py', ROOT / 'report.py',
                                 ROOT / 'vendors.py', ROOT / 'data/summary.json', Path(__file__),
                                 ROOT / 'evidence/provenance-2026-10-19.json')}
    result['provenance'] = json.loads((ROOT / 'evidence/provenance-2026-10-19.json').read_text())
    OUTPUT.write_text(json.dumps(result, indent=2) + '\n')
    print(f'{len(result["strata"])} shared strata; {result["matched_attempts"]} matched attempts; {result["exclusions"]}')


if __name__ == '__main__':
    main()