`bulk_image_check.py`

138 lines
#!/usr/bin/env python3
"""
Bulk-validate Kravet image URLs for every wallcovering row (~10,184).
URL pattern: https://www.kravet.com/media/catalog/product{image_file_hires}
Runs HEAD requests, persists result to kravet_sheet_fields:
  - image_url_hires (computed)
  - image_url_lores (computed)
  - image_hires_status (http code)
  - image_checked_at
Concurrency: 20 threads, ~20 rps → ~10min total.
"""
import json, subprocess, urllib.request, urllib.error, sys, time
from concurrent.futures import ThreadPoolExecutor, as_completed

BASE = "https://www.kravet.com/media/catalog/product"
SSH  = ["ssh", "root@45.61.58.125"]
UA   = {"User-Agent": "Mozilla/5.0 (dw-sync bot)"}


def pg_cmd(sql: str) -> str:
    r = subprocess.run(
        SSH + [f'PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified -At -F "|" -c "{sql}"'],
        capture_output=True, text=True, check=True)
    return r.stdout


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


# 1) Add columns
print("[1/4] adding columns to kravet_sheet_fields ...")
pg_exec("""
  ALTER TABLE kravet_sheet_fields
    ADD COLUMN IF NOT EXISTS image_url_hires text,
    ADD COLUMN IF NOT EXISTS image_url_lores text,
    ADD COLUMN IF NOT EXISTS image_hires_status int,
    ADD COLUMN IF NOT EXISTS image_checked_at timestamptz
""")

# 2) Fetch ALL rows with any image filename (hires OR lores)
print("[2/4] pulling all rows with any image filename ...")
rows_out = pg_cmd("""
  SELECT mfr_sku_norm, COALESCE(image_file_hires,''), COALESCE(image_file_lores,'')
  FROM kravet_sheet_fields
  WHERE COALESCE(image_file_hires,'') <> '' OR COALESCE(image_file_lores,'') <> ''
""")
rows = [line.split("|") for line in rows_out.strip().split("\n") if line]
print(f"       {len(rows)} rows to check")


def head(url: str) -> int:
    try:
        req = urllib.request.Request(url, headers=UA, method="HEAD")
        with urllib.request.urlopen(req, timeout=10) as r:
            return r.status
    except urllib.error.HTTPError as e:
        return e.code
    except Exception:
        return 0


# 3) HEAD the best URL: prefer hires, fall back to lores
print("[3/4] validating URLs (40 concurrent, hires→lores fallback) ...")
results = []  # (norm, hires_url, lores_url, status_of_best, best_kind)
done = 0
t0 = time.time()


def check_one(norm, hires, lores):
    hu = BASE + hires if hires else ""
    lu = BASE + lores if lores else ""
    status = 0; kind = ""
    if hu:
        status = head(hu); kind = "hires"
    if status != 200 and lu:
        ls = head(lu)
        if ls == 200:
            status = ls; kind = "lores"
        elif status == 0:
            status = ls; kind = "lores"
    return (norm, hu, lu, status, kind)


with ThreadPoolExecutor(max_workers=40) as ex:
    futs = [ex.submit(check_one, n, h, l) for n, h, l in rows]
    for f in as_completed(futs):
        results.append(f.result())
        done += 1
        if done % 1000 == 0:
            rps = done / max(1, time.time() - t0)
            print(f"       {done}/{len(rows)}  {rps:.1f}/s")

ok = sum(1 for r in results if r[3] == 200)
err = len(results) - ok
hires_count = sum(1 for r in results if r[4] == "hires")
lores_count = sum(1 for r in results if r[4] == "lores")
print(f"       DONE: {ok} OK ({hires_count} hires / {lores_count} lores), {err} failed, {time.time()-t0:.0f}s")

# 4) Upsert back to PG
print("[4/4] writing back to PG ...")

import tempfile, csv
with tempfile.NamedTemporaryFile("w", suffix=".csv", delete=False, newline="") as tf:
    w = csv.writer(tf)
    for norm, hu, lu, status, kind in results:
        w.writerow([norm, hu, lu, status, kind])
    tmp = tf.name
subprocess.run(["scp", tmp, "root@45.61.58.125:/tmp/kravet_image_check.csv"], check=True)
pg_exec("""
  DROP TABLE IF EXISTS _img_tmp;
  CREATE TEMP TABLE _img_tmp (mfr_sku_norm text, hu text, lu text, status int) ON COMMIT DROP;
""")
# Use a single session (temp table needs same session)
pg_exec("ALTER TABLE kravet_sheet_fields ADD COLUMN IF NOT EXISTS image_best_kind text")
r = subprocess.run(SSH + ["""PGPASSWORD=DW2024! psql -h 127.0.0.1 -U dw_admin -d dw_unified <<'SQL'
CREATE TEMP TABLE _img_tmp (mfr_sku_norm text, hu text, lu text, status int, kind text);
\\copy _img_tmp FROM '/tmp/kravet_image_check.csv' WITH (FORMAT csv);
UPDATE kravet_sheet_fields s
   SET image_url_hires    = NULLIF(t.hu,''),
       image_url_lores    = NULLIF(t.lu,''),
       image_hires_status = t.status,
       image_best_kind    = NULLIF(t.kind,''),
       image_checked_at   = now()
  FROM _img_tmp t
 WHERE s.mfr_sku_norm = t.mfr_sku_norm;
SELECT COUNT(*) FILTER (WHERE image_hires_status=200) AS ok,
       COUNT(*) FILTER (WHERE image_best_kind='hires') AS hires,
       COUNT(*) FILTER (WHERE image_best_kind='lores') AS lores,
       COUNT(*) FILTER (WHERE image_hires_status<>200 AND image_hires_status IS NOT NULL) AS bad
  FROM kravet_sheet_fields;
SQL
"""], capture_output=True, text=True)
print(r.stdout)
print("done.")