`scripts/select_push_targets.py`

95 lines
#!/usr/bin/env python3
"""DTD D1=(c) target selector — build dw_color_staging.push_targets.

Policy (DTD 3/3, 2026-06-20): push a product's corrected color metafield ONLY when
  - v2 real_hex is a valid #rrggbb, AND
  - v2 image_family is known (not null / 'unknown'), AND
  - product is ACTIVE, AND
  - the cached hex was the near-white defect  OR  v2 differs MATERIALLY from cached
    (cached null/empty, or RGB euclidean distance(cached, real) > 40).
Never overwrite an already-good cached hex with a low-confidence v2 guess.
All local + reversible (isolated dw_color_staging DB)."""
import subprocess, math, sys

WHITE_SET = {'#F2F2F2', '#FFFFFF', '#FAFAFA', '#F5F5F5', '#FFFFFE', '#FEFEFE'}
MATERIAL_DIST = 40.0

def rgb(h):
    if not h or len(h) != 7 or h[0] != '#':
        return None
    try:
        return (int(h[1:3], 16), int(h[3:5], 16), int(h[5:7], 16))
    except ValueError:
        return None

def is_valid_hex(h):
    return rgb(h) is not None

def near_white(h):
    if not h:
        return False
    H = h.upper()
    if H in WHITE_SET:
        return True
    # pattern #FxFxFx (every other nibble F) — matches analyze.sql near-white capture
    return len(H) == 7 and H[1] == 'F' and H[3] == 'F' and H[5] == 'F'

def dist(a, b):
    ra, rb = rgb(a), rgb(b)
    if ra is None or rb is None:
        return None
    return math.sqrt(sum((x - y) ** 2 for x, y in zip(ra, rb)))

def q(sql, inp=None):
    p = subprocess.run(['psql', '-d', 'dw_color_staging', '-v', 'ON_ERROR_STOP=1', '-c', sql],
                       input=inp, text=True, capture_output=True)
    if p.returncode:
        sys.stderr.write(p.stdout + p.stderr); sys.exit(1)
    return p.stdout

# pull candidates
rows = subprocess.run(
    ['psql', '-d', 'dw_color_staging', '-tAF', '\t', '-c',
     "SELECT shopify_product_id, vendor, status, declared, cached_hex, real_hex, image_family "
     "FROM product_colors_v2 WHERE lower(status)='active'"],
    text=True, capture_output=True).stdout.strip().split('\n')

targets, skipped = [], {'bad_hex': 0, 'unknown_fam': 0, 'good_cached_kept': 0}
for line in rows:
    if not line:
        continue
    spid, vendor, status, declared, cached, real, fam = (line.split('\t') + [''] * 7)[:7]
    if not is_valid_hex(real):
        skipped['bad_hex'] += 1; continue
    if not fam or fam.lower() == 'unknown':
        skipped['unknown_fam'] += 1; continue
    nw = near_white(cached)
    d = dist(cached, real)
    material = (not cached) or nw or (d is not None and d > MATERIAL_DIST) or (d is None)
    if not material:
        skipped['good_cached_kept'] += 1; continue
    targets.append((spid, vendor, status, declared, cached or '', real, fam,
                    'near_white' if nw else ('cached_missing' if not cached else f'dist={d:.0f}')))

# (re)build push_targets
q("""DROP TABLE IF EXISTS push_targets;
CREATE TABLE push_targets(
  shopify_product_id text PRIMARY KEY, vendor text, status text, declared text,
  cached_hex text, color_hex text NOT NULL, color_name text, reason text,
  backup_color_hex text, backup_color_details text, backed_up_at timestamptz,
  push_status text NOT NULL DEFAULT 'pending', pushed_at timestamptz,
  response text, is_canary boolean DEFAULT false);""")

buf = []
for spid, vendor, status, declared, cached, real, fam, reason in targets:
    name = fam[:1].upper() + fam[1:] if fam else ''
    cells = [spid, vendor, status, declared, cached, real, name, reason]
    buf.append('\t'.join((c or '').replace('\\', '\\\\').replace('\t', ' ').replace('\n', ' ') for c in cells))

p = subprocess.run(
    ['psql', '-d', 'dw_color_staging', '-c',
     "COPY push_targets(shopify_product_id,vendor,status,declared,cached_hex,color_hex,color_name,reason) FROM STDIN"],
    input='\n'.join(buf) + '\n', text=True, capture_output=True)
sys.stderr.write(p.stdout + p.stderr)
print(f"targets={len(targets)}  skipped={skipped}")