[object Object]

← back to Dw Photo Capture

visual-search: product_image_fingerprints PG table + disk-guarded idempotent loader + flagged SQL-filtered PG search path (PG_FINGERPRINTS=1, default off) (TK-12090 Lane P)

6a07319010ec2f7c4e93b67264ad7b9cd217c9e9 · 2026-09-23 14:52:33 -0700 · Steve Abrams

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KXUyzc9vybUz39rhnNJdwY

Files touched

Diff

commit 6a07319010ec2f7c4e93b67264ad7b9cd217c9e9
Author: Steve Abrams <steve@designerwallcoverings.com>
Date:   Wed Sep 23 14:52:33 2026 -0700

    visual-search: product_image_fingerprints PG table + disk-guarded idempotent loader + flagged SQL-filtered PG search path (PG_FINGERPRINTS=1, default off) (TK-12090 Lane P)
    
    Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
    Claude-Session: https://claude.ai/code/session_01KXUyzc9vybUz39rhnNJdwY
---
 visual-search/load_pg_fingerprints.py | 104 ++++++++++++++++++++++++++++++++++
 visual-search/search_service.py       |  42 +++++++++++++-
 2 files changed, 143 insertions(+), 3 deletions(-)

diff --git a/visual-search/load_pg_fingerprints.py b/visual-search/load_pg_fingerprints.py
new file mode 100644
index 0000000..e656448
--- /dev/null
+++ b/visual-search/load_pg_fingerprints.py
@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""Populate product_image_fingerprints (LOCAL Mac2 dw_unified mirror) from the existing CLIP
+fingerprint store (image_embeddings — the table search_service.py loads into RAM). TK-12090 Lane P.
+
+NOTE: vc_ids can be NEGATIVE (embed_shopify_gap.py placeholder keys for shopify-only actives,
+~27k rows) — the scan cursor starts at bigint MIN and skips already-loaded rows by anti-join.
+
+Set-based, idempotent (ON CONFLICT (source_vc_id, model_tag) DO NOTHING), resumable by vc_id.
+Adds what image_embeddings lacks per the handbag-matcher design: shopify_product_id linkage,
+product_status for SQL pre-filtering, model_tag versioning, created_at.
+
+  DW_UNIFIED_DB=<uri> python3 load_pg_fingerprints.py --limit 5000            # sample run
+  DW_UNIFIED_DB=<uri> python3 load_pg_fingerprints.py --vc-ids ids.txt        # explicit rows
+  DW_UNIFIED_DB=<uri> python3 load_pg_fingerprints.py --full                  # all ~336k (disk-guarded)
+
+A full load duplicates ~1 GB of vectors; it REFUSES unless the data volume has >= 3x the
+estimated size free (Mac2 ran at 99% full on 2026-09-23). Undo: DROP TABLE product_image_fingerprints.
+"""
+import argparse, os, shutil, sys, time
+import psycopg2
+
+DB = os.environ.get("DW_UNIFIED_DB") or "postgresql://%s@/dw_unified?host=/tmp" % (
+    os.environ.get("PGUSER") or os.environ.get("USER") or "")
+# The catalog was embedded with open_clip ViT-B-32/laion2b + dw_clip_ft.pt when present
+# (embed_catalog.py / search_service._get_model). Tag the weights so a re-embed never mixes spaces.
+HERE = os.path.dirname(os.path.abspath(__file__))
+MODEL_TAG = os.environ.get("FP_MODEL_TAG") or (
+    "open_clip:ViT-B-32/laion2b_s34b_b79k" + ("+dw_clip_ft" if os.path.exists(os.path.join(HERE, "dw_clip_ft.pt")) else ""))
+
+INSERT = """
+INSERT INTO product_image_fingerprints
+  (source_vc_id, shopify_product_id, dw_sku, mfr_sku, vendor_code, pattern_name, image_url,
+   product_status, model_tag, dim, embedding)
+SELECT e.vc_id,
+       NULLIF(regexp_replace(sp.shopify_id, '\\D', '', 'g'), '')::bigint,
+       e.dw_sku, e.mfr_sku, e.vendor_code, e.pattern_name, e.image_url,
+       sp.status, %(tag)s, COALESCE(e.dim, 512), e.embedding
+FROM image_embeddings e
+LEFT JOIN LATERAL (
+  SELECT s.shopify_id, s.status FROM shopify_products s
+  WHERE s.dw_sku = e.dw_sku AND e.dw_sku IS NOT NULL
+  ORDER BY (s.status = 'ACTIVE') DESC, s.id DESC LIMIT 1) sp ON true
+WHERE e.embedding IS NOT NULL AND {where}
+ON CONFLICT (source_vc_id, model_tag) DO NOTHING
+"""
+
+def main():
+    ap = argparse.ArgumentParser()
+    g = ap.add_mutually_exclusive_group(required=True)
+    g.add_argument("--limit", type=int)
+    g.add_argument("--vc-ids")
+    g.add_argument("--full", action="store_true")
+    ap.add_argument("--batch", type=int, default=20000)
+    ap.add_argument("--check-disk", action="store_true", help="with --full: run only the disk guard, load nothing")
+    a = ap.parse_args()
+    conn = psycopg2.connect(DB); conn.autocommit = False; cur = conn.cursor()
+    cur.execute("select to_regclass('product_image_fingerprints')")
+    if cur.fetchone()[0] is None:
+        sys.exit("product_image_fingerprints missing — apply sql/product_image_fingerprints.sql first")
+    t0 = time.time(); inserted = 0
+    if a.vc_ids:
+        ids = [int(x) for x in open(a.vc_ids).read().split() if x.strip().lstrip("-").isdigit()]   # vc_ids may be negative
+        cur.execute(INSERT.format(where="e.vc_id = ANY(%(ids)s)"), {"tag": MODEL_TAG, "ids": ids})
+        inserted = cur.rowcount; conn.commit()
+    else:
+        if a.full:
+            cur.execute("select pg_total_relation_size('image_embeddings'), current_setting('data_directory')")
+            est, datadir = cur.fetchone()
+            du = shutil.disk_usage(datadir)
+            # Need room for table + WAL churn (~3x est) AND must leave >= 10% of the volume free
+            # afterwards. (A plain `free < 3*est` test PASSED at 99% full on 2026-09-23 — 9 GB free
+            # vs a 1 GB table — and pushed the disk to 3.4 GB free. Headroom is judged vs the VOLUME.)
+            after = du.free - 3 * est
+            if after < 0.10 * du.total:
+                sys.exit("REFUSING --full: %.1f GB free, needs ~%.1f GB, would leave %.1f GB < 10%% of %.0f GB volume (%s)"
+                         % (du.free / 1e9, 3 * est / 1e9, after / 1e9, du.total / 1e9, datadir))
+            if a.check_disk:
+                print("disk guard OK: %.1f GB would remain" % (after / 1e9)); return
+        # Cursor starts at bigint MIN every run (vc_ids can be negative) and skips rows already
+        # loaded via an anti-join — so an earlier out-of-order --vc-ids load can't make a max()
+        # cursor jump past unloaded rows.
+        last = -9223372036854775808
+        remaining = a.limit if a.limit else None
+        while remaining is None or remaining > 0:
+            n = min(a.batch, remaining) if remaining is not None else a.batch
+            cur.execute("select e.vc_id from image_embeddings e where e.vc_id > %s and e.embedding is not null "
+                        "and not exists (select 1 from product_image_fingerprints f where f.source_vc_id = e.vc_id and f.model_tag = %s) "
+                        "order by e.vc_id limit %s", (last, MODEL_TAG, n))
+            ids = [r[0] for r in cur.fetchall()]
+            if not ids: break
+            cur.execute(INSERT.format(where="e.vc_id = ANY(%(ids)s)"), {"tag": MODEL_TAG, "ids": ids})
+            inserted += cur.rowcount; conn.commit(); last = ids[-1]
+            if remaining is not None: remaining -= len(ids)
+            print("  batch -> vc_id<=%s inserted_total=%d" % (last, inserted), flush=True)
+    cur.execute("select count(*), count(shopify_product_id), count(*) filter (where product_status='ACTIVE'), "
+                "pg_size_pretty(pg_total_relation_size('product_image_fingerprints')) "
+                "from product_image_fingerprints where model_tag=%s", (MODEL_TAG,))
+    total, linked, active, size = cur.fetchone()
+    print("model_tag=%s inserted=%d table_rows=%d linked_shopify=%d active=%d size=%s in %.1fs"
+          % (MODEL_TAG, inserted, total, linked, active, size, time.time() - t0))
+    conn.close()
+
+if __name__ == "__main__":
+    main()
diff --git a/visual-search/search_service.py b/visual-search/search_service.py
index 9812abc..a0511ad 100644
--- a/visual-search/search_service.py
+++ b/visual-search/search_service.py
@@ -18,6 +18,12 @@ PORT = int(os.environ.get("VS_PORT", "9914"))
 # VS_HOST=0.0.0.0 to expose on the tailnet so a remote host (e.g. Kamatera
 # pairs-well) can reach it — only do that when the exposure is intended.
 HOST = os.environ.get("VS_HOST", "127.0.0.1")
+# PG_FINGERPRINTS=1 (default OFF): source vectors from product_image_fingerprints (TK-12090 Lane P,
+# handbag-matcher design port) instead of image_embeddings, and enable SQL-FILTERED matching on
+# /search via a "filter" body key ({vendor, active_only, dw_sku_prefix}). No pgvector on the mirror,
+# so filtering happens in SQL and cosine in-process over just the filtered rows. OFF = byte-for-byte
+# the previous behaviour, so the running :9914 service is unaffected until restarted with the flag.
+PG_FINGERPRINTS = os.environ.get("PG_FINGERPRINTS") == "1"
 
 _MODEL = {"model": None, "preprocess": None}
 _MODEL_LOCK = threading.Lock()
@@ -100,9 +106,11 @@ def load():
     # handle joins ONLY to ACTIVE + Online-Store-published products: /similar's DW-safe
     # guard drops handleless rows, so archived/draft/unpublished products can never be
     # suggested as clickable results (their vectors still serve as query anchors).
-    cur.execute("""select e.vc_id, e.dw_sku, e.mfr_sku, e.vendor_code, e.pattern_name, e.image_url,
+    src = ("(select source_vc_id as vc_id, dw_sku, mfr_sku, vendor_code, pattern_name, image_url, embedding "
+           "from product_image_fingerprints)") if PG_FINGERPRINTS else "image_embeddings"
+    cur.execute(f"""select e.vc_id, e.dw_sku, e.mfr_sku, e.vendor_code, e.pattern_name, e.image_url,
                           e.embedding, v.dominant_color_hex, v.ai_styles, sp.handle
-                   from image_embeddings e
+                   from {src} e
                    left join vendor_catalog v on v.dw_sku = e.dw_sku
                    left join shopify_products sp on sp.dw_sku = e.dw_sku
                         and sp.status = 'ACTIVE' and sp.online_store_published is true
@@ -131,6 +139,28 @@ def load():
         STATE["sku_ix"] = sku_ix; STATE["fam"] = fam; STATE["styles"] = styles
     print(f"loaded {len(meta)} embeddings · {len(sku_ix)} skus indexed", flush=True)
 
+def pg_search(q, k=8, vendor=None, active_only=False, dw_sku_prefix=None):
+    """SQL-filtered match (PG_FINGERPRINTS path): pre-filter product_image_fingerprints in SQL on
+    indexed columns, then brute-force cosine in-process over only the surviving rows. Returns the
+    same result shape as /search. Vectors are the SAME bytes as image_embeddings (loader copies them)."""
+    where, args = ["true"], []
+    if vendor: where.append("vendor_code = %s"); args.append(vendor)
+    if active_only: where.append("product_status = 'ACTIVE'")
+    if dw_sku_prefix: where.append("dw_sku like %s"); args.append(dw_sku_prefix.replace("%", "") + "%")
+    conn = psycopg2.connect(DB); cur = conn.cursor()
+    cur.execute("select source_vc_id, dw_sku, mfr_sku, vendor_code, pattern_name, image_url, embedding, "
+                "shopify_product_id, product_status from product_image_fingerprints where " + " and ".join(where), args)
+    rows = cur.fetchall(); cur.close(); conn.close()
+    if not rows: return []
+    M = np.vstack([np.frombuffer(bytes(r[6]), dtype="float32") for r in rows])
+    sims = M @ q
+    k = min(k, len(rows))
+    idx = np.argpartition(-sims, k - 1)[:k] if k < len(rows) else np.arange(len(rows))
+    idx = idx[np.argsort(-sims[idx])]
+    return [{"vc_id": rows[i][0], "dw_sku": rows[i][1], "mfr_sku": rows[i][2], "vendor": rows[i][3],
+             "pattern": rows[i][4], "image": rows[i][5], "shopify_product_id": rows[i][7],
+             "status": rows[i][8], "score": round(float(sims[i]), 4)} for i in idx]
+
 def embed_photo(b64):
     import torch
     model, preprocess = _get_model()
@@ -147,7 +177,8 @@ class H(BaseHTTPRequestHandler):
         self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(b)))
         self.end_headers(); self.wfile.write(b)
     def do_GET(self):
-        if self.path.startswith("/health"): return self._send(200, {"ok": True, "n": STATE["n"], "loaded_at": STATE["loaded_at"]})
+        if self.path.startswith("/health"): return self._send(200, {"ok": True, "n": STATE["n"], "loaded_at": STATE["loaded_at"],
+                                                                    "source": "product_image_fingerprints" if PG_FINGERPRINTS else "image_embeddings"})
         if self.path.startswith("/reload"):
             try: load(); return self._send(200, {"ok": True, "n": STATE["n"]})
             except Exception as e: return self._send(500, {"ok": False, "err": str(e)})
@@ -162,6 +193,11 @@ class H(BaseHTTPRequestHandler):
             k = int(p.get("k", 8))
             if not b64: return self._send(400, {"err": "image required"})
             q = embed_photo(b64)
+            flt = p.get("filter")
+            if PG_FINGERPRINTS and isinstance(flt, dict):
+                res = pg_search(q, k, vendor=flt.get("vendor"), active_only=bool(flt.get("active_only")),
+                                dw_sku_prefix=flt.get("dw_sku_prefix"))
+                return self._send(200, {"ok": True, "source": "pg", "filter": flt, "results": res})
             with LOCK: M, meta = STATE["M"], STATE["meta"]
             if M is None or not len(M): return self._send(200, {"ok": True, "results": [], "n": 0})
             sims = M @ q                                   # cosine (both L2-normalized)

← b1c94e3 auto-data-snapshot: 2026-09-23T14:48:42 (1 data files) — vis  ·  back to Dw Photo Capture  ·  visual-search: optional pattern-sibling stage (PATTERN_SIBLI 95babfe →