← back to Kravet Fabric Fix 2026 04 23
archive_no_sheet_match.py
184 lines
#!/usr/bin/env python3
"""For DWKKs flagged `no_sheet_match` in orphan_fix_results.jsonl:
1. Recheck by catalog.mfr_sku (maybe the product just has missing/renamed tags)
2. If sheet truly has no match → archive on Shopify + set on_shopify=false in catalog
3. If sheet DOES match via mfr_sku → fix product_type + tags per Use column
"""
import csv, json, re, subprocess, time
from concurrent.futures import ThreadPoolExecutor, as_completed
import importlib.util
spec = importlib.util.spec_from_file_location('fx',
'/Users/macstudio3/Projects/kravet-fabric-fix-2026-04-23/fix_dwkk_misclassified.py')
fx = importlib.util.module_from_spec(spec); spec.loader.exec_module(fx)
SHEET_CSV = '/Users/macstudio3/Projects/kravet-sheet-sync-2026-04-20/kravet_sheet_2026-04-20.csv'
sheet = {}
with open(SHEET_CSV) as f:
for row in csv.DictReader(f):
norm = re.sub(r'\.0$', '', re.sub(r'-', '.', row['Item'].upper().strip()))
sheet[norm] = row['Use'].upper().strip()
print(f'Sheet rows: {len(sheet)}')
# No-sheet-match DWKKs from previous run
no_match = []
with open('/Users/macstudio3/Projects/kravet-fabric-fix-2026-04-23/orphan_fix_results.jsonl') as f:
for line in f:
r = json.loads(line)
if r['status'] == 'no_sheet_match' and r.get('dw_sku'):
no_match.append(r['dw_sku'])
print(f'no_sheet_match DWKKs: {len(no_match)}')
# Bulk lookup catalog mfr_sku for all these
sku_in = "','".join(no_match)
r = subprocess.run(fx.SSH + [
f"PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified --csv -t -c "
f"\"SELECT dw_sku, mfr_sku, shopify_product_id FROM kravet_catalog WHERE dw_sku IN ('{sku_in}')\""
], capture_output=True, text=True, check=True)
import io
mfr = {}
for row in csv.reader(io.StringIO(r.stdout)):
if row and row[0]:
mfr[row[0]] = {'mfr_sku': row[1], 'pid': row[2]}
print(f'catalog rows found: {len(mfr)}')
# Bucketize: recoverable via mfr_sku vs truly-gone
recoverable = [] # mfr_sku IS in sheet → fix type
truly_gone = [] # mfr_sku NOT in sheet (or no catalog row) → archive
for dw in no_match:
info = mfr.get(dw)
if not info or not info['mfr_sku']:
truly_gone.append({'dw_sku': dw, 'pid': None})
continue
norm = re.sub(r'\.0$', '', re.sub(r'-', '.', info['mfr_sku'].upper().strip()))
use = sheet.get(norm)
if use:
recoverable.append({'dw_sku': dw, 'pid': info['pid'], 'use': use, 'mfr_sku': info['mfr_sku']})
else:
truly_gone.append({'dw_sku': dw, 'pid': info['pid'], 'mfr_sku': info['mfr_sku']})
print(f'recoverable (fix type): {len(recoverable)}')
print(f'truly gone (archive): {len(truly_gone)}')
ARCHIVE_Q = '''mutation($input: ProductInput!) {
productUpdate(input: $input) {
product { id status }
userErrors { field message }
}
}'''
PRODUCT_Q = '''query($id: ID!) {
product(id: $id) { id title productType status tags vendor handle
variants(first: 2) { nodes { sku } }
}
}'''
LOOKUP_BY_SKU = '''query($q: String!) {
productVariants(first: 10, query: $q) {
nodes { sku product { id } }
}
}'''
out = open('/Users/macstudio3/Projects/kravet-fabric-fix-2026-04-23/archive_results.jsonl', 'w')
def archive_one(item):
dw = item['dw_sku']; pid = item.get('pid')
if not pid:
# Look up by SKU
try:
d = fx.gql(LOOKUP_BY_SKU, {'q': f'sku:{dw}'})
except Exception as e:
return {'dw_sku': dw, 'status': 'lookup_error', 'error': str(e)}
match = next((n for n in d['productVariants']['nodes'] if n['sku'] == dw), None)
if not match:
return {'dw_sku': dw, 'status': 'not_on_shopify'}
pid = match['product']['id']
try:
p = fx.gql(PRODUCT_Q, {'id': pid})['product']
if p is None:
return {'dw_sku': dw, 'status': 'not_found'}
if p['status'] == 'ARCHIVED':
return {'dw_sku': dw, 'status': 'already_archived'}
upd = fx.gql(ARCHIVE_Q, {'input': {'id': pid, 'status': 'ARCHIVED'}})
errs = upd['productUpdate']['userErrors']
if errs:
return {'dw_sku': dw, 'status': 'user_errors', 'errors': errs}
return {'dw_sku': dw, 'status': 'archived', 'pid': pid}
except Exception as e:
return {'dw_sku': dw, 'status': 'update_error', 'error': str(e)}
def recover_one(item):
dw, pid, use = item['dw_sku'], item['pid'], item['use']
if not pid:
try:
d = fx.gql(LOOKUP_BY_SKU, {'q': f'sku:{dw}'})
except Exception as e:
return {'dw_sku': dw, 'status': 'lookup_error', 'error': str(e)}
match = next((n for n in d['productVariants']['nodes'] if n['sku'] == dw), None)
if not match:
return {'dw_sku': dw, 'status': 'not_on_shopify'}
pid = match['product']['id']
try:
p = fx.gql(PRODUCT_Q, {'id': pid})['product']
if p is None:
return {'dw_sku': dw, 'status': 'not_found'}
target_type = 'Fabric' if use in fx.TITLE_USE else 'Wallcovering'
tag_set = set(p['tags'])
if use == 'WALLCOVERING':
tag_set.add('Wallcovering')
tag_set.discard('Fabric')
for u in fx.TITLE_USE.values():
tag_set.discard(u)
else:
tag_set.discard('Wallcovering'); tag_set.discard('wallcovering')
tag_set.add('Fabric'); tag_set.add(fx.TITLE_USE[use])
new_tags = sorted(tag_set)
if p['productType'] == target_type and sorted(p['tags']) == new_tags:
return {'dw_sku': dw, 'status': 'noop'}
upd = fx.gql(fx.PRODUCT_UPDATE, {'input': {'id': pid, 'productType': target_type, 'tags': new_tags}})
errs = upd['productUpdate']['userErrors']
if errs:
return {'dw_sku': dw, 'status': 'user_errors', 'errors': errs}
return {'dw_sku': dw, 'status': 'recovered', 'use': use, 'pid': pid}
except Exception as e:
return {'dw_sku': dw, 'status': 'update_error', 'error': str(e)}
counters = {}
print(f'Recovering {len(recoverable)}...')
with ThreadPoolExecutor(max_workers=8) as pool:
for i, res in enumerate(as_completed([pool.submit(recover_one, it) for it in recoverable]), 1):
r = res.result()
counters[r['status']] = counters.get(r['status'], 0) + 1
out.write(json.dumps({'phase': 'recover', **r}) + '\n')
if i % 200 == 0:
print(f' recover [{i}/{len(recoverable)}]: {counters}', flush=True)
out.flush()
print(f'Archiving {len(truly_gone)}...')
archive_pids = []
with ThreadPoolExecutor(max_workers=8) as pool:
for i, res in enumerate(as_completed([pool.submit(archive_one, it) for it in truly_gone]), 1):
r = res.result()
counters[r['status']] = counters.get(r['status'], 0) + 1
if r['status'] == 'archived':
archive_pids.append(r['dw_sku'])
out.write(json.dumps({'phase': 'archive', **r}) + '\n')
if i % 200 == 0:
print(f' archive [{i}/{len(truly_gone)}]: {counters}', flush=True)
out.flush()
out.close()
# Flip on_shopify=false for archived
if archive_pids:
sku_list = "','".join(archive_pids)
fx.pg_exec(f"UPDATE kravet_catalog SET on_shopify=false, orphaned_at=now(), updated_at=now() WHERE dw_sku IN ('{sku_list}')")
print(f'Marked {len(archive_pids)} catalog rows on_shopify=false')
print('FINAL:', counters)