seeder.py

"""
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()