`collect.py`

88 lines
#!/usr/bin/env python3
"""Bounded read-only queries; no API or database mutation path."""
import datetime, hashlib, json, os, pathlib, subprocess, time, urllib.request

BASE = pathlib.Path(__file__).resolve().parent
PRIOR = pathlib.Path.home() / 'Projects/ticket-system/data/codex-yoloforever/evidence/cycle-20260905T0721Z.n3ijiG/11246'
OUT = BASE / 'verification'
OUT.mkdir(exist_ok=True)

def save(name, value):
    (OUT / name).write_text(json.dumps(value, indent=2) + '\n')

def token():
    env = {}
    for line in (pathlib.Path.home() / 'Projects/secrets-manager/.env').read_text().splitlines():
        if '=' in line and not line.startswith('#'):
            k, v = line.split('=', 1)
            env[k] = v.strip().strip('\"\'')
    return env.get('SHOPIFY_ADMIN_TOKEN') or env['SHOPIFY_ADMIN_API_TOKEN']

TOKEN = token()
def gql(query, variables=None):
    assert query.lstrip().startswith('query '), 'Only explicit read queries allowed'
    for attempt in range(4):
        req = urllib.request.Request(
            'https://designer-laboratory-sandbox.myshopify.com/admin/api/2026-07/graphql.json',
            data=json.dumps({'query': query, 'variables': variables or {}}).encode(),
            headers={'X-Shopify-Access-Token': TOKEN, 'Content-Type': 'application/json'})
        with urllib.request.urlopen(req, timeout=35) as response:
            result = json.load(response)
        if result.get('errors'):
            if all(e.get('extensions', {}).get('code') == 'THROTTLED' for e in result['errors']):
                time.sleep(2 + attempt * 2)
                continue
            raise RuntimeError(json.dumps(result['errors']))
        assert result.get('data') is not None
        return result['data']
    raise RuntimeError('Read query throttle exhausted')

FIELDS = '''id title handle vendor status tags updatedAt
custom:metafield(namespace:"custom",key:"manufacturer_sku"){id value type updatedAt}
dwc:metafield(namespace:"dwc",key:"manufacturer_sku"){id value type updatedAt}
variants(first:100){pageInfo{hasNextPage} nodes{id sku title}}'''

def main():
    prior = json.loads((PRIOR / 'identity-report.json').read_text())
    save('prior-identity-report.json', prior)
    skus = [r['dw_sku'] for r in prior['rows']]
    assert len(skus) == len(set(skus)) == 26
    ids = list(dict.fromkeys('gid://shopify/Product/' + r[k] for r in prior['rows']
        for k in ['candidate_shopify_product_id', 'earlier_generation_product_id']))
    historical = []
    for offset in range(0, len(ids), 10):
        chunk = ids[offset:offset+10]
        data = gql('query History($ids:[ID!]!){nodes(ids:$ids){... on Product{' + FIELDS + '}}}', {'ids':chunk})
        historical.extend(dict(requested_id=i, product=p) for i,p in zip(chunk,data['nodes']))
        time.sleep(0.5)
    save('historical-products.json', historical)
    searches = {}
    for n, sku in enumerate(skus):
        data = gql('query Current($q:String!){productVariants(first:100,query:$q){pageInfo{hasNextPage} nodes{id sku product{' + FIELDS + '}}}}', {'q':'sku:' + sku})
        result = data['productVariants']
        assert not result['pageInfo']['hasNextPage'], 'Truncated SKU search'
        exact = [v for v in result['nodes'] if v['sku'] == sku]
        assert exact, 'No exact match for ' + sku
        for v in exact:
            assert v['product']['vendor'] == 'Carnegie'
            assert not v['product']['variants']['pageInfo']['hasNextPage']
        searches[sku] = exact
        print(f'{n+1}/26 {sku}: {len(exact)} exact variants', flush=True)
        save('current-by-sku.json', searches)
        time.sleep(0.5)
    missing = gql('query Missing{node(id:"gid://shopify/Product/1"){id}}')
    assert missing['node'] is None
    save('negative-missing-product.json', missing)
    sql = "SELECT json_build_object('observed_at',now(),'read_only',current_setting('default_transaction_read_only'),'catalog',(SELECT json_agg(t) FROM (SELECT id,dw_sku,mfr_sku,pattern_number,color_number,parent_sku,product_type,product_url,updated_at FROM carnegie_catalog ORDER BY id) t));"
    (OUT/'current-readonly.sql').write_text(sql+'\n')
    env = dict(os.environ, PGOPTIONS='-c default_transaction_read_only=on -c statement_timeout=5000', PGCONNECT_TIMEOUT='3')
    db = subprocess.run(['psql','-X','-d','dw_unified','-At','-v','ON_ERROR_STOP=1','-f',str(OUT/'current-readonly.sql')],env=env,text=True,capture_output=True,check=True)
    save('current-catalog.json',json.loads(db.stdout))
    save('collection.json', {'observed_at':datetime.datetime.now(datetime.timezone.utc).isoformat(),
        'ticket':'TK-11246','shop':'designer-laboratory-sandbox.myshopify.com','api':'2026-07',
        'historical_ids':len(ids),'sku_searches':len(searches),'business_writes':0,
        'source_sha256':hashlib.sha256((PRIOR/'identity-report.json').read_bytes()).hexdigest()})

if __name__ == '__main__':
    main()