`scripts/edges-batch-scan.py`

88 lines
#!/usr/bin/env python3
"""Read-only batch edges-scan over unpublished EDGES_FAIL rows.

Samples N unpublished EDGES_FAIL designs, resolves each PNG on local disk, runs
the 6-lens edges-scan.py, and tallies PASS/WARN/FAIL/missing. Writes a per-row
report JSON. DOES NOT touch the DB, does not flip is_published, does not deploy.
Its only purpose is to gauge the PASS-rate before deciding to promote.

Usage: edges-batch-scan.py [N]   (default 200)
"""
import os, sys, json, subprocess, re

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
GEN = os.path.join(ROOT, 'data', 'generated')
N = int(sys.argv[1]) if len(sys.argv) > 1 else 200


def db_url():
    for line in open(os.path.join(ROOT, '.env')):
        if line.startswith('DATABASE_URL='):
            return line.split('=', 1)[1].strip().strip('"')
    raise SystemExit('no DATABASE_URL in .env')


def fetch_rows(n):
    q = ("select id, coalesce(local_path,'') from all_designs "
         "where is_published=false and coalesce(user_removed,false)=false "
         "and notes ilike '%EDGES_FAIL%' order by id desc limit " + str(int(n)))
    out = subprocess.check_output(['psql', db_url(), '-tAc', q], text=True)
    rows = []
    for line in out.splitlines():
        line = line.strip()
        if '|' in line:
            did, lp = line.split('|', 1)
            rows.append((int(did), lp))
    return rows


def resolve(lp):
    if not lp:
        return None
    cands = [lp, lp if lp.endswith('.png') else lp + '.png']
    base = os.path.basename(lp.rstrip('/'))
    cands += [os.path.join(GEN, base + '.png'),
              os.path.join(GEN, base, 'final.png')]
    return next((c for c in cands if os.path.isfile(c)), None)


def scan(path):
    try:
        out = subprocess.check_output(
            ['python3', os.path.join(ROOT, 'scripts', 'edges-scan.py'), '--path', path],
            text=True, stderr=subprocess.DEVNULL, timeout=60)
        return json.loads(out)
    except Exception as e:
        return {'verdict': 'ERROR', 'error': str(e)[:120]}


def main():
    rows = fetch_rows(N)
    tally = {'PASS': 0, 'WARN': 0, 'FAIL': 0, 'ERROR': 0, 'MISSING': 0}
    detail = []
    passes = []
    for did, lp in rows:
        path = resolve(lp)
        if not path:
            tally['MISSING'] += 1
            detail.append({'id': did, 'verdict': 'MISSING'})
            continue
        r = scan(path)
        v = r.get('verdict', 'ERROR')
        tally[v] = tally.get(v, 0) + 1
        rec = {'id': did, 'verdict': v, 'score': r.get('overall_max') or r.get('score')}
        detail.append(rec)
        if v == 'PASS':
            passes.append(did)
    report = {'requested': N, 'scanned': len(rows), 'tally': tally,
              'pass_ids': passes, 'detail': detail}
    outp = os.path.join(ROOT, 'data', 'edges-batch-scan-report.json')
    with open(outp, 'w') as f:
        json.dump(report, f, indent=2)
    print(json.dumps({'tally': tally, 'pass_count': len(passes),
                      'report': outp}, indent=2))


if __name__ == '__main__':
    main()