""" Seeder — reads accessories_articles.csv, uploads each image to Cloudinary, generates a CLIP vector, and upserts to Pinecone with metadata. Runs with a thread pool so Cloudinary uploads and CLIP inference happen in parallel. Usage: python seeder.py Prerequisites: pip install cloudinary pinecone-client transformers torch pillow python-dotenv """ import os import csv import base64 import logging import threading from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed import cloudinary import cloudinary.uploader from dotenv import load_dotenv from services.clip_service import ClipService from services.pinecone_service import PineconeService load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) DATA_DIR = Path(__file__).parent / "data" CSV_PATH = DATA_DIR / "accessories_articles.csv" IMAGES_DIR = DATA_DIR / "compressed_images" CONCURRENCY = 10 # parallel workers (tune based on CPU/network) BATCH_SIZE = 100 # Pinecone upsert batch size def configure_cloudinary(): cloudinary.config( cloud_name=os.environ["CLOUDINARY_CLOUD_NAME"], api_key=os.environ["CLOUDINARY_API_KEY"], api_secret=os.environ["CLOUDINARY_API_SECRET"], secure=True, ) def upload_to_cloudinary(image_path: Path, article_id: str) -> str: result = cloudinary.uploader.upload( str(image_path), public_id=f"lensify/{article_id}", overwrite=False, resource_type="image", ) return result["secure_url"] def build_metadata(row: dict, cloudinary_url: str) -> dict: return { "article_id": row["article_id"], "product_code": row["product_code"], "name": row["prod_name"], "product_type": row["product_type_name"], "product_type_no": row["product_type_no"], "product_group": row["product_group_name"], "graphical_appearance": row["graphical_appearance_name"], "graphical_appearance_no": row["graphical_appearance_no"], "colour": row["colour_group_name"], "colour_group_code": row["colour_group_code"], "perceived_colour_value": row["perceived_colour_value_name"], "perceived_colour_master": row["perceived_colour_master_name"], "department": row["department_name"], "department_no": row["department_no"], "index_code": row["index_code"], "index_name": row["index_name"], "index_group": row["index_group_name"], "index_group_no": row["index_group_no"], "section": row["section_name"], "section_no": row["section_no"], "garment_group": row["garment_group_name"], "garment_group_no": row["garment_group_no"], "description": row["detail_desc"], "imageUrl": cloudinary_url, } def process_row(row: dict, clip: ClipService) -> dict | None: """Upload to Cloudinary + generate CLIP vector for one row. Returns None if image missing.""" article_id = row["article_id"].strip() image_path = IMAGES_DIR / f"{article_id}.jpg" if not image_path.exists(): return None cloudinary_url = upload_to_cloudinary(image_path, article_id) with open(image_path, "rb") as f: image_b64 = base64.b64encode(f.read()).decode("utf-8") vector = clip.generate_image_vector(image_b64) return { "id": article_id, "values": vector, "metadata": build_metadata(row, cloudinary_url), } def seed(): configure_cloudinary() clip = ClipService() logger.info("Loading CLIP model...") import asyncio asyncio.run(clip.load_model()) pinecone_svc = PineconeService() with open(CSV_PATH, newline="", encoding="utf-8") as f: rows = list(csv.DictReader(f)) logger.info(f"Total articles to seed: {len(rows)} | concurrency={CONCURRENCY} | batch={BATCH_SIZE}") batch = [] seeded = 0 skipped = 0 errors = 0 lock = threading.Lock() def flush_batch(b: list): nonlocal seeded pinecone_svc.index.upsert(vectors=b) with lock: seeded += len(b) logger.info(f" Upserted batch — total seeded: {seeded}") with ThreadPoolExecutor(max_workers=CONCURRENCY) as executor: futures = {executor.submit(process_row, row, clip): row for row in rows} for i, future in enumerate(as_completed(futures), 1): row = futures[future] article_id = row["article_id"].strip() try: result = future.result() if result is None: with lock: skipped += 1 continue with lock: batch.append(result) if len(batch) >= BATCH_SIZE: flush_batch(batch[:]) batch.clear() except Exception as e: logger.error(f" Failed {article_id}: {e}") with lock: errors += 1 if i % 100 == 0: with lock: s, sk, e = seeded, skipped, errors logger.info(f"Progress: {i}/{len(rows)} | seeded={s} skipped={sk} errors={e}") if batch: flush_batch(batch) logger.info(f"\nDone. Seeded: {seeded}, Skipped (no image): {skipped}, Errors: {errors}") if __name__ == "__main__": seed()