#!/usr/bin/env python3
"""How long do Meta Ad Library creative URLs stay valid? Every image and video URL the library returns
points at Meta's CDN and carries its own expiry in the query string: `oe=` is the expiry as a hex Unix
timestamp, `oh=` is the signature that covers it. This script reads any Meta ad-library receipts written by
scripts/run-ads-cohort.py (files named meta-*.json with a `_run.ts` fetch time), decodes the expiry of every
CDN URL, reports the signed lifetime per route and asset type, and optionally fetches a sample live to show
which links still resolve.

Usage: scripts/check-meta-creative-urls.py --out evidence/ads-url-checks/YYYY-MM-DD [--live 60] RUN_DIR [RUN_DIR ...]
Writes summary.json (aggregates), urls.csv (one row per URL; add --full-urls to include the link itself, otherwise the
asset id and CDN host identify it inside the raw receipt) and live.csv (one row per live probe, with the link). Free:
the live probe is a 1-byte ranged GET against Meta's CDN; no provider call is made."""
import argparse, csv, datetime as dt, glob, json, os, re, ssl, statistics, urllib.error, urllib.request
import concurrent.futures as cf

URL = re.compile(r'https?://[^"\\\s]*fbcdn\.net[^"\\\s]*')
OE = re.compile(r'[?&]oe=([0-9A-Fa-f]{8})')
ASSET = re.compile(r'/([0-9]+_[0-9]+_[0-9]+_n\.(?:jpg|png|mp4))')

def classify(u):
    host = re.match(r'https?://([^/]+)/', u).group(1)
    if host.startswith('video'): kind = 'video'
    elif re.search(r'/v/t0\.\d+-\d+/', u): kind = 'image (dynamic creative)'
    elif re.search(r'/v/t1\.\d+-\d+/', u): kind = 'image (page picture)'
    elif 's60x60' in u: kind = 'image thumbnail'
    else: kind = 'image'
    return kind

def rows_for(run_dir):
    label = os.path.basename(run_dir.rstrip('/'))
    out = []
    for f in sorted(glob.glob(os.path.join(run_dir, 'meta-*.json'))):
        j = json.load(open(f))
        ts = dt.datetime.strptime(j['_run']['ts'], '%Y-%m-%dT%H:%M:%SZ')
        seen = set()
        for u in URL.findall(json.dumps(j['response'])):
            if u in seen: continue
            seen.add(u)
            m = OE.search(u)
            oe = dt.datetime.utcfromtimestamp(int(m.group(1), 16)) if m else None
            a = ASSET.search(u)
            out.append({'run': label, 'route': j['_run']['provider'], 'adv': j['_run']['adv'], 'fetched': ts.isoformat() + 'Z',
                        'kind': classify(u), 'asset': a.group(1) if a else '', 'expires': oe.isoformat() + 'Z' if oe else '',
                        'lifetime_h': round((oe - ts).total_seconds() / 3600, 1) if oe else '', 'url': u})
    return out

CTX = ssl.create_default_context()
def probe(u):
    req = urllib.request.Request(u, headers={'User-Agent': 'Mozilla/5.0', 'Range': 'bytes=0-0'})
    try:
        with urllib.request.urlopen(req, timeout=20, context=CTX) as r: return r.status
    except urllib.error.HTTPError as e: return e.code
    except Exception as e: return type(e).__name__

def stats(v):
    return {'n': len(v), 'min_h': round(min(v), 1), 'median_h': round(statistics.median(v), 1), 'max_h': round(max(v), 1)}

def main():
    ap = argparse.ArgumentParser(); ap.add_argument('--out', required=True); ap.add_argument('--live', type=int, default=0)
    ap.add_argument('--seed', type=int, default=1); ap.add_argument('--full-urls', action='store_true'); ap.add_argument('runs', nargs='+'); a = ap.parse_args()
    os.makedirs(a.out, exist_ok=True)
    rows = [r for d in a.runs for r in rows_for(d)]
    now = dt.datetime.utcnow().replace(microsecond=0)
    summary = {'checked_at': now.isoformat() + 'Z', 'runs': [], 'urls': len(rows), 'urls_without_expiry': sum(1 for r in rows if not r['expires'])}
    by_run = {}
    for r in rows: by_run.setdefault(r['run'], []).append(r)
    assets = {k: {r['asset'] for r in v if r['asset']} for k, v in by_run.items()}
    for run, v in by_run.items():
        s = {'run': run, 'files': len({(r['route'], r['adv']) for r in v}), 'urls': len(v), 'assets': len(assets[run]), 'by_kind': {}, 'by_route': {}}
        for key, field in (('by_kind', 'kind'), ('by_route', 'route')):
            for g in sorted({r[field] for r in v}):
                s[key][g] = stats([r['lifetime_h'] for r in v if r[field] == g and r['lifetime_h'] != ''])
        s['lifetime_buckets'] = {}
        for r in v:
            if r['lifetime_h'] == '': continue
            b = 'about 32 h' if r['lifetime_h'] < 48 else 'about 106 h' if r['lifetime_h'] < 200 else 'about 720 h'
            s['lifetime_buckets'][b] = s['lifetime_buckets'].get(b, 0) + 1
        summary['runs'].append(s)
    names = list(by_run)
    summary['assets_shared_between_runs'] = {f'{x} & {y}': len(assets[x] & assets[y]) for i, x in enumerate(names) for y in names[i + 1:]}
    # A shared asset re-signed at the later fetch proves the link is refreshed per response, not per asset.
    resigned = 0
    for i, x in enumerate(names):
        for y in names[i + 1:]:
            ex = {r['asset']: r['expires'] for r in by_run[x] if r['asset']}
            ey = {r['asset']: r['expires'] for r in by_run[y] if r['asset']}
            resigned += sum(1 for k in ex.keys() & ey.keys() if ex[k] != ey[k])
    summary['shared_assets_with_new_expiry'] = resigned
    live = []
    if a.live:
        import random; random.seed(a.seed)
        sample = []
        for run, v in by_run.items():
            for b in ('about 32 h', 'about 106 h', 'about 720 h'):
                pool = [r for r in v if r['lifetime_h'] != '' and (('about 32 h' if r['lifetime_h'] < 48 else 'about 106 h' if r['lifetime_h'] < 200 else 'about 720 h') == b)]
                sample += random.sample(pool, min(a.live, len(pool)))
        with cf.ThreadPoolExecutor(16) as ex:
            for r, st in zip(sample, ex.map(lambda r: probe(r['url']), sample)):
                expired = dt.datetime.fromisoformat(r['expires'].rstrip('Z')) < now
                live.append({**{k: r[k] for k in ('run', 'route', 'kind', 'fetched', 'expires', 'lifetime_h')}, 'probed_at': now.isoformat() + 'Z', 'past_expiry': expired, 'status': st, 'url': r['url']})
        summary['live'] = {}
        for r in live:
            k = f"{r['run']} · {'past expiry' if r['past_expiry'] else 'before expiry'}"
            d = summary['live'].setdefault(k, {'probed': 0, 'ok_2xx': 0, 'forbidden_403': 0, 'other': 0})
            d['probed'] += 1
            d['ok_2xx' if isinstance(r['status'], int) and 200 <= r['status'] < 300 else 'forbidden_403' if r['status'] == 403 else 'other'] += 1
    json.dump(summary, open(os.path.join(a.out, 'summary.json'), 'w'), indent=1)
    slim = [{**{k: v for k, v in r.items() if k != 'url'}, 'host': re.match(r'https?://([^/]+)/', r['url']).group(1)} for r in rows]
    with open(os.path.join(a.out, 'urls.csv'), 'w', newline='') as f:
        w = csv.DictWriter(f, fieldnames=list((rows if a.full_urls else slim)[0].keys())); w.writeheader(); w.writerows(rows if a.full_urls else slim)
    if live:
        with open(os.path.join(a.out, 'live.csv'), 'w', newline='') as f:
            w = csv.DictWriter(f, fieldnames=list(live[0].keys())); w.writeheader(); w.writerows(live)
    print(json.dumps(summary, indent=1))

if __name__ == '__main__': main()