`verify_live.py`

76 lines
#!/usr/bin/env python3
"""Independent REST/SQL reader; imports no repair implementation."""
import argparse, datetime, json, os, pathlib, subprocess, time, urllib.error, urllib.request
from collect import token
BASE=pathlib.Path(__file__).resolve().parent
RUN=BASE/'execution-20260911'
PLAN=json.loads((BASE/'verification/repair-plan.json').read_text())
TOKEN=token()
def get(path):
    for attempt in range(5):
        req=urllib.request.Request('https://designer-laboratory-sandbox.myshopify.com/admin/api/2026-07/'+path,headers={'X-Shopify-Access-Token':TOKEN})
        try:
            with urllib.request.urlopen(req,timeout=35) as r:
                result=json.load(r)
                assert 'rel="next"' not in r.headers.get('Link',''), 'Truncated REST response'
            time.sleep(.55);return result
        except urllib.error.HTTPError as e:
            if e.code!=429:raise
            time.sleep(max(1,float(e.headers.get('Retry-After','2'))))
    raise RuntimeError('REST throttle exhausted')
def check_product(p,baseline,target=None):
    numeric=p['id'].split('/')[-1]
    actual=get('products/'+numeric+'.json')['product']
    metafields=get('products/'+numeric+'/metafields.json?limit=250')['metafields']
    for key in ['title','handle','vendor','status']:
        assert str(actual[key]).lower()==str(baseline[key]).lower(),(numeric,key)
    assert actual['body_html']==baseline['descriptionHtml']
    assert actual['product_type']==baseline['productType']
    assert sorted(x.strip() for x in actual['tags'].split(',') if x.strip())==sorted(baseline['tags'])
    av={str(v['id']):(v['sku'],v['title'],v['price'],v['compare_at_price']) for v in actual['variants']}
    bv={v['id'].split('/')[-1]:(v['sku'],v['title'],v['price'],v['compareAtPrice']) for v in baseline['variants']['nodes']}
    assert av==bv,(numeric,'variant or price drift')
    live={(m['namespace'],m['key']):(str(m['id']),m['type'],m['value']) for m in metafields}
    expected={(m['namespace'],m['key']):(m['id'].split('/')[-1],m['type'],m['value']) for m in baseline['metafields']['nodes']}
    if target is not None:
        for ns in ['custom','dwc']:
            ident,typ,_=expected[(ns,'manufacturer_sku')]
            expected[(ns,'manufacturer_sku')]=(ident,typ,target)
    assert live==expected,(numeric,'metafield mismatch')
    return dict(id=p['id'],status=actual['status'],metafields=metafields,variant_prices_unchanged=True,unrelated_fields_unchanged=True)
def main():
    ap=argparse.ArgumentParser();ap.add_argument('phase',choices=['canary','rollback-canary','all','monitor']);args=ap.parse_args()
    before=json.loads((RUN/'preflight.json').read_text())
    if (RUN/'products-baseline.json').exists():before['products']=json.loads((RUN/'products-baseline.json').read_text())['products']
    canary=next((r,p) for r in PLAN['rows'] for p in r['products'] if p['status']=='ACTIVE')
    chosen=[canary] if args.phase in ['canary','rollback-canary'] else [(r,p) for r in PLAN['rows'] for p in r['products']]
    if args.phase=='monitor':
        chosen=[canary,next((r,p) for r in PLAN['rows'] for p in r['products'] if p['status']=='ARCHIVED'),next((r,p) for r in reversed(PLAN['rows']) for p in r['products'] if p['status']=='ACTIVE')]
    results=[]
    for r,p in chosen:
        results.append(check_product(p,before['products'][p['id']],r['before'] if args.phase=='rollback-canary' else r['after']))
        print('REST verified '+p['id'],flush=True)
    extra={r['excluded_grouped_product'] for r in PLAN['rows']}
    for pid in extra:check_product({'id':pid},before['products'][pid])
    ids=','.join(str(r['catalog_id']) for r in PLAN['rows'])
    query=f"SELECT json_build_object('rows',(SELECT json_agg(t) FROM (SELECT * FROM carnegie_catalog WHERE id IN ({ids}) ORDER BY dw_sku) t),'count',(SELECT count(*) FROM carnegie_catalog),'distinct',(SELECT count(DISTINCT mfr_sku) FROM carnegie_catalog),'other_digest',(SELECT md5(string_agg(id::text||':'||dw_sku||':'||mfr_sku,'|' ORDER BY id)) FROM carnegie_catalog WHERE id NOT IN ({ids})));"
    q=subprocess.run(['psql','-X','-q','-d','dw_unified','-At','-v','ON_ERROR_STOP=1','-c',query],capture_output=True,text=True,check=True,env=dict(os.environ,PGOPTIONS='-c default_transaction_read_only=on -c statement_timeout=15000'))
    db=json.loads(q.stdout);expected_rows=before['database']['rows']
    target={r['catalog_id']:r['after'] for r in PLAN['rows']}
    for r in expected_rows:r['mfr_sku']=target[r['id']]
    assert db['rows']==expected_rows, 'Unexpected local column change'
    assert db['count']==db['distinct']==5928
    assert db['other_digest']==before['database']['other_digest']
    canonical_verified=False
    if args.phase in ['all','monitor']:
        remote_sql=f"SELECT json_agg(t) FROM (SELECT id,dw_sku,mfr_sku FROM carnegie_catalog WHERE id IN ({ids}) ORDER BY id) t;"
        remote_cmd="PGOPTIONS='-c default_transaction_read_only=on -c statement_timeout=10000' psql -d dw_unified -At -c '"+remote_sql+"'"
        result=subprocess.run(['ssh','-o','BatchMode=yes','-o','ConnectTimeout=8','kamatera',remote_cmd],capture_output=True,text=True,check=True)
        remote=json.loads(result.stdout)
        assert {(r['id'],r['dw_sku'],r['mfr_sku']) for r in remote}=={(r['catalog_id'],r['dw_sku'],r['after']) for r in PLAN['rows']}
        canonical_verified=True
    proof=dict(verdict='PASS',at=datetime.datetime.now(datetime.timezone.utc).isoformat(),phase=args.phase,reader='Independent REST GET plus read-only SQL; no repair implementation imported',products=len(results),metafields=len(results)*2,database_rows=26,unique_identities=5928,grouped_product_unchanged=True,canonical_unchanged_verified=canonical_verified,all_unrelated_metafields_and_product_fields_unchanged=True,results=results)
    (RUN/('independent-'+args.phase+'.json')).write_text(json.dumps(proof,indent=2)+'\n')
    print(json.dumps({k:v for k,v in proof.items() if k!='results'},indent=2))
if __name__=='__main__':main()