`fix_dwkk_misclassified.py`

163 lines
#!/usr/bin/env python3
"""
Fix DWKK products whose Shopify productType says wallcovering but the Kravet
master sheet classifies the item as MULTIPURPOSE / UPHOLSTERY / DRAPERY.

Per-row action (idempotent):
  1. Pull live product from Shopify (productType, tags, title, status)
  2. Target productType = 'Fabric'
  3. Target tags = existing - {Wallcovering} + {Fabric, <TitleUse>}
  4. Target title = strip stray "Wallpaper" / "Wallcovering" tokens
  5. Skip the write if nothing differs
  6. Mirror product_type='Fabric' back to kravet_catalog

Run:
  python3 fix_dwkk_misclassified.py --limit 5          # pilot
  python3 fix_dwkk_misclassified.py --limit 0          # full batch
  python3 fix_dwkk_misclassified.py --dry              # no writes
"""
import argparse, csv, json, re, subprocess, sys, time, urllib.request

SHOP  = 'designer-laboratory-sandbox.myshopify.com'
TOKEN = os.environ.get('SHOPIFY_ADMIN_TOKEN','')
GQL   = f'https://{SHOP}/admin/api/2024-10/graphql.json'
SSH   = ['ssh', 'root@45.61.58.125']
CSV_PATH = '/Users/macstudio3/Projects/kravet-fabric-fix-2026-04-23/dwkk_misclassified_2026-04-23.csv'
RESULTS  = '/Users/macstudio3/Projects/kravet-fabric-fix-2026-04-23/fix_results.jsonl'

TITLE_USE = {'MULTIPURPOSE': 'Multipurpose', 'UPHOLSTERY': 'Upholstery', 'DRAPERY': 'Drapery'}
STRIP_TITLE = re.compile(r'\s*\|\s*(wallpaper|wallcovering)s?\b', re.I)
STRIP_TITLE_TAIL = re.compile(r'\b(wallpaper|wallcovering)s?\s*$', re.I)


def gql(query, variables=None):
    body = json.dumps({'query': query, 'variables': variables or {}}).encode()
    req = urllib.request.Request(GQL, data=body, method='POST', headers={
        'Content-Type': 'application/json',
        'X-Shopify-Access-Token': TOKEN,
    })
    resp = urllib.request.urlopen(req, timeout=30)
    data = json.loads(resp.read())
    if 'errors' in data:
        raise RuntimeError(data['errors'])
    return data['data']


def pg_exec(sql):
    subprocess.run(SSH + [
        f"PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified -c \"{sql}\""
    ], check=True, capture_output=True)


def compute_target(product, sheet_use):
    cur_type = product['productType']
    cur_tags = list(product['tags'])
    cur_title = product['title']

    new_type = 'Fabric'
    tag_set = set(cur_tags)
    tag_set.discard('Wallcovering')
    tag_set.discard('wallcovering')
    tag_set.add('Fabric')
    tag_set.add(TITLE_USE[sheet_use])
    new_tags = sorted(tag_set)

    new_title = STRIP_TITLE.sub('', cur_title)
    new_title = STRIP_TITLE_TAIL.sub('', new_title).strip().rstrip('|').strip()

    changes = {}
    if cur_type != new_type:
        changes['productType'] = (cur_type, new_type)
    if sorted(cur_tags) != new_tags:
        changes['tags'] = (cur_tags, new_tags)
    if cur_title != new_title:
        changes['title'] = (cur_title, new_title)
    return new_type, new_tags, new_title, changes


PRODUCT_Q = '''query($id: ID!) {
  product(id: $id) { id title productType status tags vendor handle }
}'''

PRODUCT_UPDATE = '''mutation($input: ProductInput!) {
  productUpdate(input: $input) {
    product { id productType tags title }
    userErrors { field message }
  }
}'''


def fix_one(row, dry=False):
    pid = row['shopify_product_id']
    dw_sku = row['dw_sku']
    sheet_use = row['sheet_use']

    try:
        data = gql(PRODUCT_Q, {'id': pid})
    except Exception as e:
        return {'dw_sku': dw_sku, 'pid': pid, 'status': 'fetch_error', 'error': str(e)}
    p = data['product']
    if p is None:
        return {'dw_sku': dw_sku, 'pid': pid, 'status': 'not_found'}

    new_type, new_tags, new_title, changes = compute_target(p, sheet_use)
    if not changes:
        return {'dw_sku': dw_sku, 'pid': pid, 'status': 'noop', 'title': p['title']}

    if dry:
        return {'dw_sku': dw_sku, 'pid': pid, 'status': 'would_change', 'changes': changes}

    variables = {'input': {
        'id': pid,
        'productType': new_type,
        'tags': new_tags,
        'title': new_title,
    }}
    try:
        upd = gql(PRODUCT_UPDATE, variables)
    except Exception as e:
        return {'dw_sku': dw_sku, 'pid': pid, 'status': 'update_error', 'error': str(e), 'changes': changes}
    errs = upd['productUpdate']['userErrors']
    if errs:
        return {'dw_sku': dw_sku, 'pid': pid, 'status': 'user_errors', 'errors': errs, 'changes': changes}

    sql = f"UPDATE kravet_catalog SET product_type='Fabric', updated_at=now() WHERE dw_sku='{dw_sku}'"
    try:
        pg_exec(sql)
    except Exception as e:
        return {'dw_sku': dw_sku, 'pid': pid, 'status': 'shopify_ok_pg_fail', 'error': str(e), 'changes': changes}

    return {'dw_sku': dw_sku, 'pid': pid, 'status': 'ok', 'changes': changes}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--limit', type=int, default=5)
    ap.add_argument('--dry', action='store_true')
    ap.add_argument('--offset', type=int, default=0)
    args = ap.parse_args()

    with open(CSV_PATH) as f:
        rows = list(csv.DictReader(f))
    if args.offset:
        rows = rows[args.offset:]
    if args.limit:
        rows = rows[:args.limit]

    print(f'Processing {len(rows)} products{" (DRY)" if args.dry else ""}')
    counters = {}
    with open(RESULTS, 'a') as out:
        for i, row in enumerate(rows, 1):
            result = fix_one(row, dry=args.dry)
            counters[result['status']] = counters.get(result['status'], 0) + 1
            out.write(json.dumps(result) + '\n')
            out.flush()
            if i % 25 == 0 or i <= 10:
                print(f'  [{i}/{len(rows)}] {row["dw_sku"]}: {result["status"]}')
            time.sleep(0.15)
    print('Done:', counters)


if __name__ == '__main__':
    main()