#!/usr/bin/env python3 """ Bol.com CSV Feed Importer v2 for SneakerPicks ---------------------------------------------- Key improvements over v1: - EAN-first product matching (same as feed-sync-v4.py) - Fingerprint matching as fallback - Groups sizes per product instead of 1 offer per size - Better model/name extraction from familyName Usage: python3 import-bol-feed-v2.py --feed /path/to/bol-footwear-feed.csv.gz python3 import-bol-feed-v2.py --feed /path/to/bol-footwear-feed.csv.gz --dry-run """ import sys, io, os, re, gzip, csv, json, time, argparse, hashlib from decimal import Decimal, InvalidOperation from collections import defaultdict from urllib.parse import quote 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", "") if not DB_URL: # Try loading from .env file env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '.env') if os.path.exists(env_path): with open(env_path) as f: for line in f: line = line.strip() if line.startswith('DATABASE_URL='): DB_URL = line.split('=', 1)[1] if not DB_URL: print("ERROR: DATABASE_URL not set") sys.exit(1) # Brands we want to import (same as feed-sync-v4.py SNEAKER_BRANDS) 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', 'mizuno', 'k-swiss', 'le coq sportif', 'ellesse', 'under armour', 'brooks', 'golden goose', 'axel arigato', 'philippe model', 'alexander mcqueen', 'balenciaga', 'versace', 'gucci', 'off-white', 'stone island', 'moncler', 'premiata', 'copenhagen', 'copenhagen studios', 'mason garments', 'mercer amsterdam', 'yeezy', 'ugg', 'birkenstock', 'clarks', 'hugo boss', 'boss', 'calvin klein', 'umbro', 'lotto', 'altra', 'topo athletic', 'hi-tec', 'li-ning', } # Title keywords that indicate this is NOT a sneaker 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', r'\bpump[s]?\b', r'\bhak(?:ken)?\b', r'\bstiletto\b', ] _EXCLUDE_RE = re.compile('|'.join(EXCLUDE_PATTERNS), re.IGNORECASE) BOL_AFFILIATE_SITE_ID = "1507527" GENDER_MAP = { 'mannen': 'men', 'vrouwen': 'women', 'unisex': 'unisex', 'jongens': 'boys', 'meisjes': 'girls', } def slugify(text): s = text.lower().strip() s = re.sub(r'[^a-z0-9\s-]', '', s) s = re.sub(r'[\s-]+', '-', s) return s.strip('-')[:200] def make_fingerprint(brand, name, colorway): """Same fingerprint function as feed-sync-v4.py for cross-feed matching.""" parts = [ (brand or '').lower(), re.sub(r'[^\w\s]', '', (name or '').lower()), (colorway or '').lower() ] return hashlib.md5('|'.join(p.strip() for p in parts).encode()).hexdigest() def make_affiliate_url(product_url): return f"https://partnerprogramma.bol.com/click/click?p=1&t=url&s={BOL_AFFILIATE_SITE_ID}&url={quote(product_url, safe='')}&f=API" def clean_title(title, brand): """Clean the product title for matching: remove brand prefix, size/color suffixes.""" name = title.strip() # Remove brand prefix (case-insensitive) if name.lower().startswith(brand.lower()): name = name[len(brand):].strip(' -') # Remove trailing size like "- 42.5" or "- Maat 42" name = re.sub(r'\s*-\s*(?:Maat\s*)?\d{2,3}(?:[.,]5)?\s*$', '', name, flags=re.I) # Remove trailing EU size like "- EU 37.5" name = re.sub(r'\s*-\s*EU\s*\d{2,3}(?:[.,]5)?\s*$', '', name, flags=re.I) # Remove trailing gender/category like "- Unisex" or "- Dames" name = re.sub(r'\s*-\s*(?:Unisex|Dames|Heren|Kinderen|Kids|Junior)\s*$', '', name, flags=re.I) # Remove trailing color like "- Zwart/Wit" name = re.sub(r'\s*-\s*(?:Maat\s+)?(?:EU\s+)?\d{2,3}(?:[.,]5)?\s*$', '', name, flags=re.I) # Remove trailing material info name = re.sub(r'\s*-\s*(?:Mesh|Synthetisch|Leer|Canvas|Textiel|Suède|Rubber|Nubuck)(?:/\w+)*\s*$', '', name, flags=re.I) return name.strip(' -') or title.strip() def parse_price(val): if not val or not val.strip(): return None try: return Decimal(val.strip()) except InvalidOperation: return None def normalize_size(raw): """Normalize a raw size string to a clean EU size. Simplified version — Bol.com sizes are usually clean numbers like '42.5'. """ if not raw or not isinstance(raw, str): return None s = raw.strip().replace(',', '.') if not s or s.lower() in ('one size', 'onesize', 'os'): return None # Remove "Maat"/"EU"/"EUR" prefix s = re.sub(r'^(?:Maat|EU|EUR|Size)\s*', '', s, flags=re.I).strip() # Handle adidas fractional like "42 2/3" frac_m = re.match(r'^(\d{2})\s+(\d)/(\d)$', s) if frac_m: whole = int(frac_m.group(1)) frac = int(frac_m.group(2)) / int(frac_m.group(3)) return f"{whole}.5" if frac > 0.25 else str(whole) # Plain number m = re.match(r'^(\d{2,3}(?:\.\d{1,2})?)$', s) if m: n = float(m.group(1)) if 17 <= n <= 52: n = round(n * 2) / 2 return str(int(n)) if n == int(n) else f"{n:.1f}" return None def is_real_sneaker(title, brick): text = f"{title} {brick}" return not _EXCLUDE_RE.search(text) def main(): parser = argparse.ArgumentParser(description="Import Bol.com footwear feed v2") parser.add_argument("--feed", required=True, 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 rows (0=all)") args = parser.parse_args() print("=== Bol.com Feed Importer v2 (EAN-first matching) ===") print(f" Feed: {args.feed}") print(f" Dry run: {args.dry_run}") # --- Phase 1: Read CSV and group by familyName+brand+colour --- print("\n[1/4] Reading and filtering CSV...") t0 = time.time() # Group products: key = (brand, familyName, colour) -> list of size variants groups = defaultdict(list) total_rows = 0 skipped_category = 0 skipped_brand = 0 skipped_exclude = 0 skipped_no_price = 0 skipped_not_deliverable = 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 category subgroup = row.get('Category.subgroup', '').lower() if 'sneaker' not in subgroup: skipped_category += 1 continue # Filter: deliverable, new condition if row.get('OfferNL.isDeliverable') != 'Y': skipped_not_deliverable += 1 continue if row.get('OfferNL.condition') != 'new': continue # Filter: target brands brand = row.get('brand', '').strip() if brand.lower() not in TARGET_BRANDS: skipped_brand += 1 continue # Filter: price price = parse_price(row.get('OfferNL.sellingPrice', '')) if not price or price <= 0: skipped_no_price += 1 continue # Filter: exclude non-sneaker items title = row.get('title', '').strip() brick = row.get('Gpc.brickName', '').strip() if not is_real_sneaker(title, brick): skipped_exclude += 1 continue # Extract fields family_name = row.get('familyName', '').strip() or title colour = row.get('Colour', '').strip() raw_size = row.get('ShoeSize', '').strip() size = normalize_size(raw_size) if raw_size else None ean = row.get('ean', '').strip() or None list_price = parse_price(row.get('OfferNL.listPrice', '')) gender_raw = row.get('Gender', '').strip().lower() model_name = row.get('Model', '').strip() product_id_bol = row.get('productId', '').strip() image_url = row.get('imageUrl', '').strip() raw_url = row.get('productPageUrlNL', '').strip() # Group key: brand + cleaned familyName + colour cleaned_name = clean_title(family_name, brand) group_key = (brand.lower(), cleaned_name.lower(), colour.lower()) groups[group_key].append({ 'external_id': product_id_bol, 'brand': brand, 'title': title, 'family_name': family_name, 'cleaned_name': cleaned_name, 'colour': colour, 'gender': GENDER_MAP.get(gender_raw, gender_raw), 'price': price, 'list_price': list_price, 'size': size, 'image_url': image_url, 'raw_url': raw_url, 'ean': ean, 'model_name': model_name, }) if args.limit and sum(len(v) for v in groups.values()) >= args.limit: break elapsed = time.time() - t0 total_variants = sum(len(v) for v in groups.values()) print(f" Total rows scanned: {total_rows}") print(f" Skipped (category): {skipped_category}") print(f" Skipped (brand): {skipped_brand}") print(f" Skipped (excluded): {skipped_exclude}") print(f" Skipped (no price): {skipped_no_price}") print(f" Skipped (not deliverable): {skipped_not_deliverable}") print(f" Groups (unique products): {len(groups)}") print(f" Total variants: {total_variants}") print(f" Avg sizes per product: {total_variants/max(len(groups),1):.1f}") print(f" Time: {elapsed:.1f}s") if args.dry_run: brand_counts = defaultdict(int) for variants in groups.values(): brand_counts[variants[0]['brand']] += 1 print("\n Brand distribution:") for b, c in sorted(brand_counts.items(), key=lambda x: -x[1])[:20]: print(f" {b}: {c} products") return # --- Phase 2: Connect to DB and build caches --- print("\n[2/4] Connecting to DB and building caches...") conn = psycopg2.connect(DB_URL) conn.autocommit = False cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor) # Provider + Feed cur.execute("SELECT id FROM providers WHERE slug = 'bolcom'") 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() 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" Provider ID: {provider_id}, Feed ID: {feed_id}") # Build EAN cache from existing products cur.execute("SELECT id, ean, match_fingerprint FROM products WHERE ean IS NOT NULL OR match_fingerprint IS NOT NULL") ean_cache = {} fp_cache = {} for r in cur.fetchall(): if r['ean']: ean_cache[r['ean']] = r['id'] if r['match_fingerprint']: fp_cache[r['match_fingerprint']] = r['id'] print(f" EAN cache: {len(ean_cache)} entries") print(f" Fingerprint cache: {len(fp_cache)} entries") # Brand cache cur.execute("SELECT id, slug, name FROM brands") brand_cache = {} for r in cur.fetchall(): brand_cache[r['slug']] = r['id'] print(f" Brand cache: {len(brand_cache)} entries") # Mark all existing bol.com offers as not seen (we'll set last_seen_at for ones we find) cur.execute("UPDATE product_offers SET in_stock = false WHERE feed_id = %s", (feed_id,)) conn.commit() stale_count = cur.rowcount print(f" Marked {stale_count} existing offers as out of stock") # --- Phase 3: Import grouped products --- print("\n[3/4] Importing products with EAN-first matching...") stats = { 'matched_ean': 0, 'matched_fp': 0, 'new_products': 0, 'offers_upserted': 0, 'errors': 0, } feed_used_pids = set() # prevent same product being used twice in this feed processed = 0 for group_key, variants in groups.items(): processed += 1 try: # Pick representative variant (first one, or one with image) rep = variants[0] for v in variants: if v['image_url']: rep = v break brand_name = rep['brand'] brand_slug = slugify(brand_name) cleaned_name = rep['cleaned_name'] colour = rep['colour'] gender = rep['gender'] model_name = rep['model_name'] if model_name.lower() in ('geen', '', 'nvt', 'n.v.t.', 'overig'): model_name = cleaned_name # Collect all EANs and all sizes all_eans = set() size_entries = [] min_price = None min_orig = None best_url = '' best_image = '' for v in variants: if v['ean']: all_eans.add(v['ean']) size = v['size'] if size: size_entries.append({ 'size': size, 'ean': v['ean'] or '', }) if min_price is None or v['price'] < min_price: min_price = v['price'] best_url = v['raw_url'] best_image = v['image_url'] if v['list_price'] and v['list_price'] > v['price']: if min_orig is None or v['list_price'] < min_orig: min_orig = v['list_price'] if not min_price: continue primary_ean = next(iter(all_eans), None) # --- TIER 1: EAN match --- product_id = None for ean in all_eans: if ean in ean_cache: product_id = ean_cache[ean] stats['matched_ean'] += 1 break cur.execute("SELECT id FROM products WHERE ean = %s", (ean,)) r = cur.fetchone() if r: product_id = r['id'] ean_cache[ean] = product_id stats['matched_ean'] += 1 break # --- TIER 2: Fingerprint match --- if not product_id: fp = make_fingerprint(brand_name, cleaned_name, colour) if fp in fp_cache: product_id = fp_cache[fp] stats['matched_fp'] += 1 else: cur.execute("SELECT id FROM products WHERE match_fingerprint = %s", (fp,)) r = cur.fetchone() if r: product_id = r['id'] fp_cache[fp] = product_id stats['matched_fp'] += 1 # Prevent same-feed duplicates if product_id and product_id in feed_used_pids: product_id = None # --- TIER 3: Create new product --- if not product_id: 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, brand_slug) ) brand_cache[brand_slug] = cur.fetchone()['id'] except: conn.rollback() cur.execute("SELECT id FROM brands WHERE slug = %s", (brand_slug,)) r = cur.fetchone() if r: brand_cache[brand_slug] = r['id'] brand_id = brand_cache.get(brand_slug) fp = make_fingerprint(brand_name, cleaned_name, colour) product_title = f"{brand_name} {cleaned_name}".strip() product_slug = f"{slugify(product_title)[:180]}-{hashlib.md5(f'bol-{group_key[0]}-{group_key[1]}-{group_key[2]}'.encode()).hexdigest()[:8]}" try: cur.execute(""" INSERT INTO products (name, slug, brand_id, brand, image_url, ean, colorway, gender, brand_normalized, model_normalized, colorway_normalized, match_fingerprint, model, model_name, is_sneaker, needs_cutout, first_seen_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, true, true, NOW()) RETURNING id """, ( product_title, product_slug, brand_id, brand_name, best_image, primary_ean, colour, gender or None, brand_name.lower(), cleaned_name.lower(), colour.lower() if colour else None, fp, model_name, model_name, )) product_id = cur.fetchone()['id'] stats['new_products'] += 1 fp_cache[fp] = product_id for ean in all_eans: ean_cache[ean] = product_id except psycopg2.errors.UniqueViolation: conn.rollback() cur.execute("SELECT id FROM products WHERE slug = %s", (product_slug,)) r = cur.fetchone() if r: product_id = r['id'] else: continue except Exception as e: conn.rollback() stats['errors'] += 1 if stats['errors'] <= 10: print(f" Error creating '{product_title}': {e}") continue feed_used_pids.add(product_id) # Build sizes JSON (grouped) # Deduplicate sizes seen_sizes = set() unique_sizes = [] for s in size_entries: if s['size'] not in seen_sizes: seen_sizes.add(s['size']) unique_sizes.append(s) sizes_json = json.dumps(unique_sizes, ensure_ascii=False) # Prices original_price = float(min_orig) if min_orig and min_orig > min_price else None affiliate_url = make_affiliate_url(best_url) if best_url else '' # Use first variant's external_id as the offer key external_id = f"bol-{variants[0]['external_id']}" # Upsert offer 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, external_id, float(min_price), original_price, affiliate_url, best_image, sizes_json, )) stats['offers_upserted'] += 1 except Exception as e: stats['errors'] += 1 if stats['errors'] <= 10: print(f" ERROR [{processed}]: {e}") conn.rollback() if processed % 500 == 0: conn.commit() print(f" [{processed}/{len(groups)}] EAN:{stats['matched_ean']} FP:{stats['matched_fp']} New:{stats['new_products']} Offers:{stats['offers_upserted']} Err:{stats['errors']}") conn.commit() # --- Phase 4: Cleanup and stats --- print("\n[4/4] Cleanup...") # Remove offers not seen in this sync (stale) cur.execute("DELETE FROM product_offers WHERE feed_id = %s AND in_stock = false", (feed_id,)) removed = cur.rowcount print(f" Removed {removed} stale offers") # Update feed stats cur.execute("UPDATE feeds SET product_count = %s, last_sync_at = now(), last_sync_status = 'success' WHERE id = %s", (stats['offers_upserted'], feed_id)) conn.commit() total_time = time.time() - t0 print(f"\n{'='*50}") print(f" Total rows scanned: {total_rows}") print(f" Product groups: {len(groups)}") print(f" Matched via EAN: {stats['matched_ean']}") print(f" Matched via FP: {stats['matched_fp']}") print(f" New products created: {stats['new_products']}") print(f" Offers upserted: {stats['offers_upserted']}") print(f" Stale offers removed: {removed}") print(f" Errors: {stats['errors']}") print(f" Total time: {total_time:.0f}s") cur.close() conn.close() if __name__ == "__main__": main()