#!/usr/bin/env python3 """ Bol.com CSV Feed Importer for SneakerPicks ------------------------------------------- Reads the bol.com fashion-footwear CSV feed, filters sneakers, and upserts one offer per bol.com productId into the SneakerPicks DB. Fix B: per-product importing (one row = one product/offer). productId is the unique bol.com identifier per variant. Prerequisites: - SSH tunnel: ssh -i ~/.ssh/hostinger_vps -N -L 5433:172.18.0.5:5432 root@72.62.45.83 - Feed file: D:\bol-footwear-feed.csv.gz """ import sys, io, os, re, gzip, csv, json, time, argparse from decimal import Decimal, InvalidOperation sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace', line_buffering=True) sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace', line_buffering=True) import psycopg2 import psycopg2.extras DB_URL = os.environ.get( "DATABASE_URL", "postgresql://supabase_admin:HzWrn2rMMxUYWO6OT5X0PScIVEvSjQMup9YSLIrkPQ@localhost:5433/sneakerpicks" ) FEED_FILE = r"D:\bol-footwear-feed.csv.gz" # Brands we want to import TARGET_BRANDS = { 'nike', 'adidas', 'adidas originals', 'new balance', 'puma', 'asics', 'jordan', 'vans', 'converse', 'reebok', 'skechers', 'hoka', 'on running', 'on', 'salomon', 'diadora', 'karhu', 'filling pieces', 'autry', 'common projects', 'saucony', 'hummel', 'geox', 'floris van bommel', 'nubikk', 'cruyff', 'fila', 'lacoste', 'tommy hilfiger', 'reebok classic', } # Title keywords that indicate this is NOT a sneaker (whole-word or context-aware) EXCLUDE_PATTERNS = [ r'\bsandaa?l\b', r'\bslipper[s]?\b', r'\bslide[s]?\b', r'\bflip.?flop', r'\bmuil(en)?\b', r'\bclog[s]?\b', r'\blaars(jes|en)?\b', r'\bballerina\b', r'\bespadrille\b', r'\bloafer[s]?\b', r'\bmocassin\b', r'\bpantoffel[s]?\b', r'\bwandelschoen\b', r'\bhiking\b', r'\bvoetbal(schoen)?\b', r'\bfootball\b', r'\brugby\b', r'\bzaalschoen\b', r'\bzaalsport\b', r'\bzaal\b', r'\bbadschoen\b', r'\bwaterschoen\b', r'\bbadeschuhe?\b', r'\bbadslippers?\b', r'\bteenslippers?\b', r'\bklompen\b', r'\bsokken\b', r'\binlegzool\b', r'\bschoenlepel\b', r'\brugzak\b', ] import re as _re _EXCLUDE_RE = _re.compile('|'.join(EXCLUDE_PATTERNS), _re.IGNORECASE) def is_real_sneaker(title: str, brick: str) -> bool: """Filter out non-sneaker footwear that slips through category filters.""" text = f"{title} {brick}" return not _EXCLUDE_RE.search(text) BOL_AFFILIATE_SITE_ID = "1507527" def make_affiliate_url(product_url: str) -> str: """Wrap a bol.com URL in the affiliate tracking redirect.""" from urllib.parse import quote return f"https://partnerprogramma.bol.com/click/click?p=1&t=url&s={BOL_AFFILIATE_SITE_ID}&url={quote(product_url, safe='')}&f=API" GENDER_MAP = { 'mannen': 'men', 'vrouwen': 'women', 'unisex': 'unisex', 'jongens': 'boys', 'meisjes': 'girls', } def slugify(text: str) -> str: """Convert text to URL-friendly slug.""" s = text.lower().strip() s = re.sub(r'[^a-z0-9\s-]', '', s) s = re.sub(r'[\s-]+', '-', s) return s.strip('-')[:200] def clean_family_name(name: str, brand: str) -> str: """Remove brand prefix and size/color suffixes from familyName.""" # Remove brand prefix if name.lower().startswith(brand.lower()): name = name[len(brand):].strip(' -') # Remove trailing " - EU", " - NL", size info name = re.sub(r'\s*-\s*(EU|NL|Maat\s*\d+).*$', '', name, flags=re.IGNORECASE) # Remove trailing size like "- 42.5" name = re.sub(r'\s*-\s*\d{2,3}(\.5)?$', '', name) return name.strip(' -') or name def parse_price(val: str) -> Decimal | None: """Parse price string to Decimal.""" if not val or not val.strip(): return None try: return Decimal(val.strip()) except InvalidOperation: return None def main(): parser = argparse.ArgumentParser(description="Import Bol.com footwear feed") parser.add_argument("--feed", default=FEED_FILE, help="Path to feed CSV.gz") parser.add_argument("--dry-run", action="store_true", help="Don't write to DB") parser.add_argument("--limit", type=int, default=0, help="Limit number of rows to process (0=all)") args = parser.parse_args() print("=== Bol.com Feed Importer (Fix B: per-product) ===") print(f" Feed: {args.feed}") print(f" Dry run: {args.dry_run}") # --- Phase 1: Read CSV, collect flat list of per-product rows --- print("\n[1/3] Reading and filtering CSV...") t0 = time.time() products = [] # flat list — one entry per bol.com productId total_rows = 0 skipped_no_price = 0 with gzip.open(args.feed, 'rt', encoding='utf-8', errors='replace') as f: reader = csv.DictReader(f, delimiter='|', quotechar='"') for row in reader: total_rows += 1 if total_rows % 500000 == 0: print(f" Scanned {total_rows} rows...") # Filter: sneakers only subgroup = row.get('Category.subgroup', '').lower() if 'sneaker' not in subgroup: continue # Filter: deliverable, new condition if row.get('OfferNL.isDeliverable') != 'Y': continue if row.get('OfferNL.condition') != 'new': continue # Filter: target brands only brand = row.get('brand', '').strip() if brand.lower() not in TARGET_BRANDS: continue # Filter: must have price price = parse_price(row.get('OfferNL.sellingPrice', '')) if not price or price <= 0: skipped_no_price += 1 continue # Filter: exclude non-sneaker items by title/brick title = row.get('title', '') brick = row.get('Gpc.brickName', '') if not is_real_sneaker(title, brick): continue # Collect all fields we need per product row list_price = parse_price(row.get('OfferNL.listPrice', '')) colour = row.get('Colour', '').strip() gender_raw = row.get('Gender', '').strip().lower() family_name = row.get('familyName', '').strip() or title.strip() model_name = row.get('Model', '').strip() size = row.get('ShoeSize', '').strip() products.append({ 'external_id': row.get('productId', '').strip(), 'brand_name': brand, 'title': title, 'colour': colour, 'gender': GENDER_MAP.get(gender_raw, gender_raw), 'price': price, 'list_price': list_price, 'size': size, 'image_url': row.get('imageUrl', '').strip(), 'raw_url': row.get('productPageUrlNL', '').strip(), 'ean': row.get('ean', '').strip() or None, 'model_name': model_name, 'family_name': family_name, }) # Apply row limit after filtering if args.limit and len(products) >= args.limit: break elapsed = time.time() - t0 filtered_rows = len(products) print(f" Total rows scanned: {total_rows}") print(f" Filtered sneaker rows: {filtered_rows}") print(f" Skipped (no price): {skipped_no_price}") print(f" Time: {elapsed:.1f}s") if args.dry_run: from collections import defaultdict brand_counts = defaultdict(int) for p in products: brand_counts[p['brand_name']] += 1 print("\n Brand distribution:") for b, c in sorted(brand_counts.items(), key=lambda x: -x[1])[:20]: print(f" {b}: {c} rows") return # --- Phase 2: Connect to DB and prepare --- print("\n[2/3] Connecting to DB...") conn = psycopg2.connect(DB_URL) conn.autocommit = False cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor) # Create/find provider for Bol.com cur.execute("SELECT id FROM providers WHERE name = 'Bol.com'") row = cur.fetchone() if row: provider_id = row['id'] else: cur.execute("INSERT INTO providers (name, slug, adapter_type) VALUES ('Bol.com', 'bolcom', 'csv') RETURNING id") provider_id = cur.fetchone()['id'] conn.commit() print(f" Provider ID: {provider_id}") # Create/find feed for Bol.com cur.execute("SELECT id FROM feeds WHERE provider_id = %s AND name = 'Bol.com Footwear'", (provider_id,)) row = cur.fetchone() if row: feed_id = row['id'] else: cur.execute(""" INSERT INTO feeds (provider_id, name, merchant_name, feed_url, sync_frequency, is_enabled) VALUES (%s, 'Bol.com Footwear', 'Bol.com', 'ftp://apm-feed.unftp.bol.com', 'daily', true) RETURNING id """, (provider_id,)) feed_id = cur.fetchone()['id'] conn.commit() print(f" Feed ID: {feed_id}") # Cache existing brands cur.execute("SELECT id, slug, name FROM brands") brand_cache = {} # slug -> id for r in cur.fetchall(): brand_cache[r['slug']] = r['id'] print(f" Existing brands: {len(brand_cache)}") # --- Phase 3: Import per-product rows --- print("\n[3/3] Importing products...") stats = { 'total_rows': total_rows, 'filtered_rows': filtered_rows, 'new_products': 0, 'updated_offers': 0, 'new_brands': 0, 'skipped': skipped_no_price, 'errors': 0, } processed = 0 for p in products: processed += 1 try: brand_name = p['brand_name'] brand_slug = slugify(brand_name) # Find/create brand if brand_slug not in brand_cache: cur.execute( "INSERT INTO brands (name, slug) VALUES (%s, %s) ON CONFLICT (slug) DO NOTHING RETURNING id", (brand_name, brand_slug) ) result = cur.fetchone() if result: brand_cache[brand_slug] = result['id'] stats['new_brands'] += 1 else: cur.execute("SELECT id FROM brands WHERE slug = %s", (brand_slug,)) brand_cache[brand_slug] = cur.fetchone()['id'] brand_id = brand_cache[brand_slug] # Determine canonical product name and slug family_name = p['family_name'] colour = p['colour'] cleaned_family = clean_family_name(family_name, brand_name) product_name = f"{brand_name} {cleaned_family}" if cleaned_family else family_name if colour: product_name = f"{product_name} {colour}" product_slug = slugify(product_name) # Clean up model_name model_name = p['model_name'] if model_name.lower() in ('geen', '', 'nvt', 'n.v.t.', 'overig'): model_name = cleaned_family # Look up product by slug, or insert cur.execute("SELECT id FROM products WHERE slug = %s", (product_slug,)) existing = cur.fetchone() if existing: product_id = existing['id'] else: cur.execute(""" INSERT INTO products (name, slug, brand_id, brand, model, model_name, colorway, gender, image_url, ean, is_sneaker, brand_normalized, model_normalized, colorway_normalized, first_seen_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, true, %s, %s, %s, now()) ON CONFLICT (slug) DO UPDATE SET updated_at = now() RETURNING id """, ( product_name, product_slug, brand_id, brand_name, model_name, model_name, colour, p['gender'], p['image_url'], p['ean'], brand_slug, slugify(model_name) if model_name else None, slugify(colour) if colour else None, )) product_id = cur.fetchone()['id'] stats['new_products'] += 1 # Build sizes JSON: single size for this row, or empty array size = p['size'] sizes_json = json.dumps([size]) if size else json.dumps([]) # Prices price = p['price'] list_price = p['list_price'] original_price = float(list_price) if list_price and list_price > price else None # Affiliate URL for this specific row's product page product_url = make_affiliate_url(p['raw_url']) if p['raw_url'] else '' # Upsert offer — external_id = this row's productId (the key fix) cur.execute(""" INSERT INTO product_offers (product_id, feed_id, external_id, price, original_price, currency, product_url, image_url, in_stock, sizes, last_seen_at) VALUES (%s, %s, %s, %s, %s, 'EUR', %s, %s, true, %s::jsonb, now()) ON CONFLICT (feed_id, external_id) DO UPDATE SET product_id = EXCLUDED.product_id, price = EXCLUDED.price, original_price = EXCLUDED.original_price, product_url = EXCLUDED.product_url, image_url = EXCLUDED.image_url, in_stock = true, sizes = EXCLUDED.sizes, last_seen_at = now(), updated_at = now() """, ( product_id, feed_id, p['external_id'], float(price), original_price, product_url, p['image_url'], sizes_json, )) stats['updated_offers'] += 1 except Exception as e: stats['errors'] += 1 if stats['errors'] <= 10: print(f" ERROR [{processed}]: {e}") conn.rollback() # Commit every 500 if processed % 500 == 0: conn.commit() print(f" [{processed}/{filtered_rows}] committed... (products: {stats['new_products']}, offers: {stats['updated_offers']}, errors: {stats['errors']})") # Final commit conn.commit() # Update feed stats cur.execute("UPDATE feeds SET product_count = %s, last_sync_at = now(), last_sync_status = 'success' WHERE id = %s", (stats['updated_offers'], feed_id)) conn.commit() total_time = time.time() - t0 print(f"\n{'='*50}") print(f" Total rows scanned: {stats['total_rows']}") print(f" Filtered rows: {stats['filtered_rows']}") print(f" New products: {stats['new_products']}") print(f" Offers upserted: {stats['updated_offers']}") print(f" New brands: {stats['new_brands']}") print(f" Skipped (no price): {stats['skipped']}") print(f" Errors: {stats['errors']}") print(f" Total time: {total_time:.0f}s") cur.close() conn.close() if __name__ == "__main__": main()