#!/usr/bin/env python3 """ SneakerPicks Feed Sync v2 — Cursor-based pagination Fetches ALL products from Webgains feeds using search_cursor pagination. """ import requests import psycopg2 import psycopg2.extras import re import hashlib import time import sys # ── Config ── WEBGAINS_API = "https://platform-api.webgains.com" PUBLISHER_ID = "1347930" WEBGAINS_TOKEN = "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJhdWQiOiIxMDAwMzEiLCJqdGkiOiIwNjM5ODEwYzQ0YzM1MTBlZmZiMGRkYWNiM2EyNWFlNTg2YzFmNmQ5ODNjNGQ2MDA4NzkwZmM0NTkzY2EyNjg4ZGViZDE4NTgwOThmYjJjYiIsImlhdCI6MTc3MTUxNjc4OS44MzAxMzUsIm5iZiI6MTc3MTUxNjc4OS44MzAxMzcsImV4cCI6MjA4NzA0OTU4OS44MjcwODgsInN1YiI6IjIwMTcwOCIsInNjb3BlcyI6W119.ZxlCDsyBI7_SNUWjO-PUxDYKvQVfuUBzU5n-SgqHdAZCnOj9253yVcJIS5NRXJLrnyx0E_9bn8ymfXS71cvPISA4hPAlKCSMjdkv9f2k2FFwdAlQQ6mdhBB2m9dzHUIn34QwA8s4-ikB1ATKIZeqiJL-3Gh9eiMYPpvaJdCCjN-oGPMhhlehkgA04cZSgwy6HfKgb6uPvUSG9w_tvN8U06uEHKV11DedQHNWWGzuXpd2FEL3baV0z9KjSNMZsJZHx2yHUb6gVWJwaR7_louahMvrPXG-YLgnGvNMcPDPjm-z5HJ-JKcPv9_FBqAYGItZxgeGgVoS8MVhW6lc1a-6CvYtVanfXF42MPPNP-Qq76qRtHPLRBdkIATMuBxVJrMzlprT3n4F0sUUltGR4ixZuzMUm5LeyPMHgmjy2UfYzsl9eVqx0oJCmKUyY9maBXwFLJa8J1NHM_nWp3hPpCGkMyhzHK1DIf_z8Wirg1hp5vKts93eERWJBlQwhucUUS_KnZeC_gVaQ7kT4jCwaEWkpPxPxyXWRoK0mcYS_JGz3JbpVhRLj5DVYIB9k54DXqjYlCx8ezJZE9sf2SX3VWvPvD1xSQD1caTCm0To0B-MkeU1YpEfWdUslpCJXrpOy2Wbs8iDHR1BzNrMnOEr4zbz_EoB4JJe3tGTGjzNLNIw6XI" DB_URL = "postgresql://supabase_admin:HzWrn2rMMxUYWO6OT5X0PScIVEvSjQMup9YSLIrkPQ@172.18.0.5:5432/sneakerpicks" CAMPAIGNS = { "manify.nl": "1622230", "ladify.nl": "1622250", "thehike.nl": "1622240", "mellowed.nl": "1622260", } HEADERS = { "Authorization": f"Bearer {WEBGAINS_TOKEN}", "Content-Type": "application/json", "Accept": "application/json", } PRODUCT_FIELDS = ["title", "image_link", "link", "price", "brand", "description", "id", "sale_price", "gtin", "mpn", "colour", "size", "availability", "condition", "product_type", "google_product_category"] def slugify(text): text = text.lower().strip() text = re.sub(r'[^\w\s-]', '', text) text = re.sub(r'[\s_]+', '-', text) text = re.sub(r'-+', '-', text) return text[:200] def parse_price(val): if not val: return None val = str(val).replace(',', '.').strip() match = re.search(r'[\d.]+', val) if match: try: return float(match.group()) except: return None return None def get_feeds(): """Get all unique feeds across all campaigns.""" all_feeds = {} for site, camp_id in CAMPAIGNS.items(): url = f"{WEBGAINS_API}/auth/publishers/{PUBLISHER_ID}/campaigns/{camp_id}/feeds" try: r = requests.get(url, headers=HEADERS, params={"size": 100}, timeout=30) r.raise_for_status() data = r.json() for f in data.get("data", []): fid = f["id"] if fid not in all_feeds: all_feeds[fid] = { "id": fid, "name": f.get("name", ""), "program_name": f.get("program", {}).get("name", ""), "program_id": str(f.get("program", {}).get("id", "")), "product_count": f.get("products_successfully_stored", 0), "campaign_id": camp_id, } print(f" {site} ({camp_id}): {len(data.get('data', []))} feeds") except Exception as e: print(f" {site} ERROR: {e}") return all_feeds def fetch_all_products_cursor(campaign_id, feed_id, max_pages=200): """Fetch ALL products using cursor-based pagination.""" url = f"{WEBGAINS_API}/auth/publishers/{PUBLISHER_ID}/campaigns/{campaign_id}/feeds/products" all_products = [] seen_ids = set() cursor = None for page in range(max_pages): params = { "feeds[]": [feed_id], "size": 1000, "fields[]": PRODUCT_FIELDS, } if cursor: params["search_cursor"] = cursor try: r = requests.get(url, headers=HEADERS, params=params, timeout=60) r.raise_for_status() data = r.json() except Exception as e: print(f" Page {page} error: {e}") break products = data.get("data", []) if not products: break pagination = data.get("pagination", {}) new_cursor = pagination.get("search_cursor") # Deduplicate new_products = [] for p in products: pid = p.get("id") if pid and pid not in seen_ids: seen_ids.add(pid) new_products.append(p) if not new_products: print(f" Page {page}: no new products, done") break all_products.extend(new_products) if page % 10 == 0 or len(new_products) < 900: print(f" Page {page}: +{len(new_products)} new (total: {len(all_products)})") if new_cursor == cursor or not new_cursor: print(f" Cursor stopped changing at page {page}") break cursor = new_cursor time.sleep(0.3) # Rate limit return all_products def main(): print("=== SneakerPicks Feed Sync v2 (Cursor Pagination) ===") print() conn = psycopg2.connect(DB_URL) conn.autocommit = False cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor) # Clear old data for clean import print("Clearing old data...") cur.execute("TRUNCATE product_offers, offers, price_history, featured_products, collection_products, collection_overrides, product_categories, product_tags, price_alerts, products, feeds, sync_logs, field_mappings RESTART IDENTITY CASCADE") cur.execute("DELETE FROM brands WHERE slug != 'unknown'") conn.commit() print("Done.\n") cur.execute("SELECT id FROM providers WHERE slug = 'webgains'") provider_id = cur.fetchone()["id"] print("Fetching feeds from Webgains API...") feeds = get_feeds() print(f"Found {len(feeds)} unique feeds\n") # Brand cache brand_cache = {} cur.execute("SELECT id, slug FROM brands") for row in cur.fetchall(): brand_cache[row["slug"]] = row["id"] # Product slug cache to avoid duplicates slug_cache = set() # EAN to product_id cache ean_cache = {} total_products = 0 total_offers = 0 for feed_ext_id, feed_info in sorted(feeds.items()): fname = feed_info["name"] pcount = feed_info["product_count"] camp_id = feed_info["campaign_id"] print(f"\n{'='*60}") print(f"Feed: {fname} (ext_id={feed_ext_id}, ~{pcount} products)") print(f"{'='*60}") # Create feed in DB cur.execute(""" INSERT INTO feeds (provider_id, external_id, name, merchant_name, product_count) VALUES (%s, %s, %s, %s, %s) RETURNING id """, (provider_id, str(feed_ext_id), fname, feed_info["program_name"], pcount or 0)) db_feed_id = cur.fetchone()["id"] conn.commit() # Fetch ALL products with cursor pagination products = fetch_all_products_cursor(camp_id, feed_ext_id) if not products: print(" No products, skipping") continue print(f" Total fetched: {len(products)}") feed_new = 0 feed_offers = 0 batch_size = 500 for i in range(0, len(products), batch_size): batch = products[i:i+batch_size] for p in batch: title = (p.get("title") or "").strip() brand_name = (p.get("brand") or "").strip() image = (p.get("image_link") or "").strip() link = (p.get("link") or "").strip() price = parse_price(p.get("sale_price") or p.get("price")) original_price = parse_price(p.get("price")) if p.get("sale_price") else None ext_id = str(p.get("id", "")).strip() gtin_val = p.get("gtin") or "" if isinstance(gtin_val, list): gtin_val = gtin_val[0] if gtin_val else "" ean = str(gtin_val).strip() or None colour = (p.get("colour") or "").strip() if not title or not link or not price: continue # Brand brand_slug = slugify(brand_name) if brand_name else "unknown" if brand_slug not in brand_cache: try: cur.execute(""" INSERT INTO brands (name, slug) VALUES (%s, %s) ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name RETURNING id """, (brand_name or "Unknown", brand_slug)) brand_cache[brand_slug] = cur.fetchone()["id"] except Exception: conn.rollback() cur.execute("SELECT id FROM brands WHERE slug = %s", (brand_slug,)) row = cur.fetchone() if row: brand_cache[brand_slug] = row["id"] else: continue brand_id = brand_cache[brand_slug] # Check EAN match product_id = None if ean and ean in ean_cache: product_id = ean_cache[ean] elif ean: cur.execute("SELECT id FROM products WHERE ean = %s", (ean,)) row = cur.fetchone() if row: product_id = row["id"] ean_cache[ean] = product_id if not product_id: # Generate unique slug base_slug = slugify(f"{brand_name}-{title}")[:180] slug_hash = hashlib.md5(f"{feed_ext_id}-{ext_id}".encode()).hexdigest()[:8] product_slug = f"{base_slug}-{slug_hash}" if product_slug in slug_cache: # Already created this product, get its ID cur.execute("SELECT id FROM products WHERE slug = %s", (product_slug,)) row = cur.fetchone() if row: product_id = row["id"] if not product_id: try: cur.execute(""" INSERT INTO products (name, slug, brand_id, brand, image_url, description, ean, colorway, brand_normalized, first_seen_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) RETURNING id """, (title, product_slug, brand_id, brand_name, image, (p.get("description") or "")[:500], ean, colour, brand_name.lower() if brand_name else None)) product_id = cur.fetchone()["id"] slug_cache.add(product_slug) if ean: ean_cache[ean] = product_id feed_new += 1 except psycopg2.errors.UniqueViolation: conn.rollback() cur.execute("SELECT id FROM products WHERE slug = %s", (product_slug,)) row = cur.fetchone() if row: product_id = row["id"] else: continue except Exception as e: conn.rollback() continue # Insert offer try: cur.execute(""" INSERT INTO product_offers (product_id, feed_id, external_id, price, original_price, currency, product_url, image_url, in_stock, last_seen_at) VALUES (%s, %s, %s, %s, %s, 'EUR', %s, %s, true, NOW()) ON CONFLICT (feed_id, external_id) DO UPDATE SET price = EXCLUDED.price, original_price = EXCLUDED.original_price, product_url = EXCLUDED.product_url, image_url = EXCLUDED.image_url, in_stock = true, last_seen_at = NOW(), updated_at = NOW() """, (product_id, db_feed_id, ext_id, price, original_price, link, image)) feed_offers += 1 except Exception as e: conn.rollback() continue # Commit per batch conn.commit() # Update feed stats cur.execute(""" UPDATE feeds SET product_count = %s, last_sync_at = NOW(), last_sync_status = 'success' WHERE id = %s """, (feed_offers, db_feed_id)) conn.commit() total_products += feed_new total_offers += feed_offers print(f" ✓ {feed_new} new products, {feed_offers} offers") print(f"\n{'='*60}") print(f"SYNC COMPLETE") print(f"{'='*60}") print(f"New products created: {total_products}") print(f"Total offers: {total_offers}") cur.execute("SELECT COUNT(*) FROM products") print(f"Products in DB: {cur.fetchone()[0]}") cur.execute("SELECT COUNT(*) FROM product_offers") print(f"Offers in DB: {cur.fetchone()[0]}") cur.execute("SELECT COUNT(*) FROM brands") print(f"Brands in DB: {cur.fetchone()[0]}") cur.execute("SELECT COUNT(*) FROM feeds") print(f"Feeds in DB: {cur.fetchone()[0]}") cur.execute("SELECT brand, COUNT(*) c FROM products GROUP BY brand ORDER BY c DESC LIMIT 20") print("\nTop brands:") for row in cur.fetchall(): print(f" {row[0] or '(empty)'}: {row[1]}") cur.close() conn.close() if __name__ == "__main__": main()