#!/usr/bin/env python3
"""Scores a SERP-only run against an earlier run of the same cohort. Per route and per date: how
many calls returned data, how many result rows, whether a google.com/goto redirect sits in the
result URL field or elsewhere in the response, latency, and cost. Across routes: how far the
routes agree with each other on the same day, how many distinct result lists each query came back
with, and how far each route's own results moved between the two dates.
Writes src/data/serp-checks/<date>.json and copies the per-call records to
public/evidence/serp-runs/<date>/. Scores are computed on the raw responses; the published copies
drop any result row that names a term on the site's exclusion list (scripts/check-exclusions.mjs)
and record the count in the run header. Deterministic; no provider call.
Usage: scripts/score-serp-check.py evidence/fixtures/seo-cohort-v1.json <baseline-run-dir> <new-run-dir>"""
import json, sys, os, glob, math, statistics, csv, re, collections
from urllib.parse import urlparse
cohort = json.load(open(sys.argv[1])); base_dir, new_dir = sys.argv[2].rstrip('/'), sys.argv[3].rstrip('/')
base_date, date = os.path.basename(base_dir), os.path.basename(new_dir)
Q = cohort['queries']
PRICE = {  # USD per call and where the figure comes from; treg-routed, dataforseo and serpstat are read from the responses
    'cloro': (0.002, 'settled charge on the account ledger; catalog list price is $0.0028'), 'serpapi': (0.015, 'catalog list price, charged per successful search'),
    'anyapi': (0.0004, 'settled charge on the account ledger; catalog list price is $0.0009'),
    'limadata': (0.002, 'settled charge on the account ledger'), 'serper': (0.001, 'settled charge on the account ledger'),
    'scrapecreators': (0.00188, 'catalog list price for one credit; called on a key registered on the account'),
}
KIND = {'serpstat': 'stored index'}
ORDER = ['cloro', 'dataforseo', 'serpapi', 'treg-routed', 'serpstat', 'anyapi', 'limadata', 'scrapecreators', 'serper']
SAME_LIST = 0.2  # two responses are the same result list when their URL sets share at least this Jaccard overlap
def load(d, task='serp'): return [r for r in (json.load(open(f)) for f in sorted(glob.glob(os.path.join(d, '*.json')))) if r['_run'].get('task') == task]
def norm(u):
    if not u: return None
    p = urlparse(u); host = p.netloc.lower().removeprefix('www.'); path = p.path.rstrip('/') or '/'
    return host + path
def is_goto(u):
    p = urlparse(u or ''); return p.netloc.lower().removeprefix('www.') == 'google.com' and p.path.startswith('/goto')
def no_data(r):
    b = r['response']
    return isinstance(b, dict) and isinstance(b.get('error'), dict) and 'not found' in str(b['error'].get('message', '')).lower()
def ok(r):
    b = r['response']
    if not isinstance(b, (dict, list)): return False
    if isinstance(b, dict) and ('detail' in b or 'error' in b or '_nonjson' in b): return False
    if isinstance(b, dict) and b.get('status_code') not in (None, 20000): return False
    if isinstance(b, dict) and b.get('success') is False: return False
    return r['_run']['exit'] == 0
def raw_urls(r, top=10):
    p = r['_run']['provider']; b = r['response']
    try:
        if p == 'cloro': items = [(x['position'], x['link']) for x in b['result']['organicResults']]
        elif p == 'dataforseo': items = [(x['rank_group'], x['url']) for x in b['tasks'][0]['result'][0]['items'] if x.get('type') == 'organic']
        elif p == 'serpstat': items = [(x['position'], x['url']) for x in b['result']['data']['top']]
        elif p == 'serpapi': items = [(x['position'], x['link']) for x in b['organic_results']]
        elif p == 'serper': items = [(x['position'], x['link']) for x in b['organic']]
        elif p == 'limadata': items = [(x.get('position', i+1), x['url']) for i, x in enumerate(b['organic'])]
        elif p == 'anyapi': items = [(x.get('position', i+1), x['link']) for i, x in enumerate(b['output']['data']['results'])]
        elif p == 'scrapecreators': items = [(i+1, x.get('url')) for i, x in enumerate(b['results'])]
        elif p == 'treg-routed':
            rows = [x for x in b['output']['results'] if x.get('type', 'organic') == 'organic']
            items = [(i+1, x.get('url')) for i, x in enumerate(rows)]
        else: items = []
    except Exception: return []
    return [u for _, u in sorted(items, key=lambda t: t[0]) if u][:top]
def urls(r, top=10): return [norm(u) for u in raw_urls(r, top)]
def goto_anywhere(o):
    if isinstance(o, dict): return any(goto_anywhere(v) for v in o.values())
    if isinstance(o, list): return any(goto_anywhere(v) for v in o)
    return isinstance(o, str) and 'google.com/goto' in o
def jaccard(a, b): return len(a & b) / len(a | b)
def pct(lat, q): return lat[max(0, math.ceil(q*len(lat))-1)] if lat else None
def price(p, oks):
    if p == 'treg-routed': return round(statistics.mean((r['response'].get('_treg') or {}).get('charged_micro', 0)/1e6 for r in oks), 5), 'charge reported in each response'
    if p == 'dataforseo': return round(statistics.mean(r['response'].get('cost', 0) for r in oks), 5), 'cost field in each response'
    if p == 'serpstat': return round(statistics.mean(len(r['response']['result']['data']['top']) * 0.0005 for r in oks), 5), 'rows returned at the catalog list price of $0.0005 a row; one call checked against the account ledger'
    return PRICE.get(p, (None, None))

# ---------- per route ----------
def table(recs):
    by = {}
    for r in recs: by.setdefault(r['_run']['provider'], {})[r['_run']['q']] = r
    rows = {}
    for p, byq in by.items():
        oks = [r for r in byq.values() if ok(r)]; nod = [r for r in byq.values() if no_data(r)]
        lat = sorted(r['_run']['latency_ms'] for r in byq.values())
        counts = [len(raw_urls(r)) for r in oks]
        served = {}
        for r in oks:
            s = (r['response'].get('_treg') or {}).get('served_by') if p == 'treg-routed' else None
            if s: served[s.split('.')[0]] = served.get(s.split('.')[0], 0) + 1
        usd, basis = price(p, oks)
        rows[p] = {'provider': p, 'kind': KIND.get(p, 'live fetch'), 'calls': len(byq), 'ok': len(oks), 'no_data': len(nod), 'errors': len(byq) - len(oks) - len(nod),
                   'empty': sum(1 for c in counts if c == 0), 'median_results': statistics.median(counts) if counts else None,
                   'goto_in_url_field': sum(1 for r in oks for u in raw_urls(r) if is_goto(u)), 'responses_with_goto_elsewhere': sum(1 for r in oks if goto_anywhere(r['response'])),
                   'p50_ms': lat[len(lat)//2] if lat else None, 'p95_ms': pct(lat, 0.95), 'cost_usd_per_call': usd, 'cost_basis': basis, 'served_by': served or None}
    return by, rows
base_by, base_rows = table(load(base_dir)); new_recs = load(new_dir); new_by, new_rows = table(new_recs)
def sets(by, routes):
    return {q: {p: set(urls(by[p][q])) for p in routes if p in by and q in by[p] and ok(by[p][q]) and urls(by[p][q])} for q in Q}

# ---------- agreement between routes on the same day ----------
def agreement(by):
    ps = [p for p in ORDER if p in by]; s = sets(by, ps); out = []
    for a in ps:
        row = {'provider': a}
        for b in ps:
            v = [jaccard(s[q][a], s[q][b]) for q in Q if a in s[q] and b in s[q]]
            row[b] = round(statistics.mean(v), 2) if v else None
        out.append(row)
    return out

# ---------- how many distinct result lists came back per query ----------
def result_lists(by, routes):
    s = sets(by, routes); n = collections.Counter(); big = collections.Counter(); tot = collections.Counter(); disjoint = 0
    for q in Q:
        groups = []
        for p, u in s[q].items():
            hit = [g for g in groups if any(jaccard(u, t) >= SAME_LIST for _, t in g)]
            new = [(p, u)]
            for g in hit: new += g; groups.remove(g)
            groups.append(new)
        groups.sort(key=len, reverse=True); n[min(len(groups), 3)] += 1
        for p, _ in groups[0]: big[p] += 1
        for p in s[q]: tot[p] += 1
        us = list(s[q].values())
        disjoint += 1 if any(not (a & b) for i, a in enumerate(us) for b in us[i+1:]) else 0
    return {'routes': [p for p in ORDER if p in routes and p in by], 'queries_with_one_list': n[1], 'queries_with_two_lists': n[2], 'queries_with_three_or_more_lists': n[3],
            'queries_with_a_pair_sharing_no_url': disjoint, 'in_largest_group': {p: f'{big[p]}/{tot[p]}' for p in ORDER if p in tot}}
same = [p for p in ORDER if p in base_by and p in new_by]
lists = {'same_list_threshold_jaccard': SAME_LIST, f'{base_date}_same_routes': result_lists(base_by, same), f'{date}_same_routes': result_lists(new_by, same), f'{date}_all_routes': result_lists(new_by, list(new_by))}

# ---------- each route against its own earlier run, and against the earlier day's common page ----------
LIVE_BASE = [p for p in base_by if base_rows[p]['kind'] == 'live fetch']
base_sets = sets(base_by, LIVE_BASE); page = {}; first = {}
for q in Q:  # the page at least three live routes returned on the baseline date, and the first result most of them returned
    c = collections.Counter(u for s in base_sets[q].values() for u in s)
    page[q] = {u for u, k in c.items() if k >= min(3, len(LIVE_BASE))}
    f = collections.Counter(urls(base_by[p][q])[0] for p in base_sets[q])
    first[q] = f.most_common(1)[0][0] if f else None
def against_page(by):
    out = {}
    for p, byq in by.items():
        share, top = [], 0
        for q, r in byq.items():
            u = urls(r) if ok(r) else []
            if not u or not page.get(q): continue
            share.append(len(set(u) & page[q]) / len(page[q])); top += 1 if u[0] == first[q] else 0
        out[p] = {'queries_compared': len(share), 'mean_share_of_baseline_page_returned': round(statistics.mean(share), 2) if share else None, 'same_first_result_as_baseline_page': f'{top}/{len(share)}' if share else None}
    return out
base_page, new_page = against_page(base_by), against_page(new_by)
moved = []
for p in [x for x in ORDER if x in new_rows and x in base_by]:
    kept, top, n = [], 0, 0
    for q in Q:
        a, b_ = base_by[p].get(q), new_by[p].get(q)
        if not a or not b_ or not ok(a) or not ok(b_): continue
        ua, ub = urls(a), urls(b_)
        if not ua or not ub: continue
        n += 1; kept.append(len(set(ua) & set(ub)) / len(ua)); top += 1 if ua[0] == ub[0] else 0
    moved.append({'provider': p, 'queries_compared': n, 'share_of_urls_kept': round(statistics.mean(kept), 2) if kept else None, 'same_first_result': f'{top}/{n}' if n else None})

# ---------- depth checks: where each route's first page sits in a deeper fetch of the same query ----------
depth = []
for r in load(new_dir, 'serp-depth'):
    q = r['_run']['q']; row = {'query': q, 'route': r['_run']['provider'], 'label': r['_run']['label'], 'ok': ok(r), 'latency_ms': r['_run']['latency_ms']}
    if ok(r):
        deep = urls(r, top=1000); row['organic_returned'] = len(deep)
        row['first_page_urls_found'] = {p: f"{len(set(urls(new_by[p][q])) & set(deep))}/{len(urls(new_by[p][q]))}" for p in ORDER if p in new_by and q in new_by[p] and ok(new_by[p][q]) and urls(new_by[p][q])}
    depth.append(row)

BK = ('calls', 'ok', 'no_data', 'errors', 'empty', 'median_results', 'goto_in_url_field', 'responses_with_goto_elsewhere', 'p50_ms', 'p95_ms', 'served_by', 'mean_share_of_baseline_page_returned', 'same_first_result_as_baseline_page')
rows = []
for p in [x for x in ORDER if x in new_rows] + [x for x in new_rows if x not in ORDER]:
    b = {**base_rows[p], **base_page[p]} if p in base_rows else None
    rows.append({**new_rows[p], **new_page[p], 'baseline': ({k: b[k] for k in BK} if b else None)})
summary = {'date': date, 'baseline_date': base_date, 'cohort': cohort['version'], 'queries': len(Q), 'calls': len(new_recs), 'ok': sum(ok(r) for r in new_recs),
           'no_data': sum(r['no_data'] for r in rows), 'errors': sum(r['errors'] for r in rows), 'est_cost_usd': round(sum(r['ok'] * (r['cost_usd_per_call'] or 0) for r in rows), 3),
           'routes': rows, 'agreement': {date: agreement(new_by), base_date: agreement(base_by)}, 'result_lists': lists, 'moved': moved, 'depth': depth}

# ---------- publish ----------
EXCLUDED = [t.lower() for t in re.findall(r"'([^']+)'", re.search(r'EXCLUDED = \[(.*?)\]', open(os.path.join(os.path.dirname(os.path.abspath(__file__)), 'check-exclusions.mjs')).read()).group(1))]
def named(s): return isinstance(s, str) and any(t in s.lower() for t in EXCLUDED)
def withhold(o, n):
    """Returns (copy, hit): list items that name an excluded term are dropped, n[0] counts them."""
    if isinstance(o, list):
        out = []
        for x in o:
            y, hit = withhold(x, n)
            if hit or named(x): n[0] += 1
            else: out.append(y)
        return out, False
    if isinstance(o, dict):
        out, hit = {}, False
        for k, v in o.items():
            if named(v) or named(k): hit = True; out['withheld' if named(k) else k] = '[withheld]'
            else:
                y, h = withhold(v, n); out[k] = y; hit = hit or h
        return out, hit
    return o, False
pub = f'public/evidence/serp-runs/{date}'; os.makedirs(os.path.join(pub, 'tables'), exist_ok=True)
withheld_rows = withheld_files = 0; published = new_recs + load(new_dir, 'serp-depth')
for r in published:
    n = [0]; body, _ = withhold(r['response'], n)
    head = {**r['_run'], 'withheld_rows': n[0]} if n[0] else r['_run']
    withheld_rows += n[0]; withheld_files += 1 if n[0] else 0
    txt = json.dumps({'_run': head, 'response': body})
    assert not named(txt), r['_run']['label']
    open(os.path.join(pub, r['_run']['label'] + '.json'), 'w').write(txt)
summary['published'] = {'files': len(published), 'withheld_rows': withheld_rows, 'files_with_withheld_rows': withheld_files}
def write(name, rows_):
    keys = list(rows_[0].keys())
    with open(os.path.join(pub, 'tables', name), 'w', newline='') as fh:
        w = csv.DictWriter(fh, fieldnames=keys); w.writeheader()
        for r in rows_: w.writerow({k: (json.dumps(v) if isinstance(v, (dict, list)) else v) for k, v in r.items()})
write('routes.csv', [{**{k: v for k, v in r.items() if k != 'baseline'}, **{f'baseline_{k}': (r['baseline'] or {}).get(k) for k in BK}} for r in rows])
for d_, rows_ in summary['agreement'].items(): write(f'agreement-{d_}.csv', rows_)
write('result-lists.csv', [{'run': k, **{a: b for a, b in v.items()}} for k, v in lists.items() if isinstance(v, dict)])
write('moved.csv', moved)
if depth: write('depth.csv', [{'first_page_urls_found': None, 'organic_returned': None, **r} for r in depth])
summary['files'] = {'tables': sorted(os.listdir(os.path.join(pub, 'tables'))), 'raw_count': len(published)}
os.makedirs('src/data/serp-checks', exist_ok=True); json.dump(summary, open(f'src/data/serp-checks/{date}.json', 'w'), indent=1)
print(json.dumps({k: summary[k] for k in ('calls', 'ok', 'no_data', 'errors', 'est_cost_usd', 'published')}))
for r in rows: print('SERP', {k: v for k, v in r.items() if k not in ('baseline', 'cost_basis')}); print('  was', r['baseline'])
for k, v in lists.items(): print('LISTS', k, v)
for a in summary['agreement'][date]: print('AGREE', a)
for m in moved: print('MOVED', m)
for d_ in depth: print('DEPTH', d_)