track product with LFS

.gitattributes

@@ -0,0 +1 @@
+data/product_data.csv filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,8 @@
+env/
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+*.DS_Store
+data/embeddings.npy
+data/image_urls.pkl

data/product_data.csv

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5fd4ad76f4783518e02888845949d8eb4ecffe69860266d3fc576d2d27cadca4
+size 30552758

main.py

@@ -0,0 +1,80 @@
+import uvicorn
+import numpy as np
+import pandas as pd
+import pickle
+from fastapi import FastAPI, File, UploadFile
+from fastapi.responses import JSONResponse
+from PIL import Image
+from io import BytesIO
+from model.feature_extractor import FeatureExtractor
+from utils.faiss_index import FaissIndex
+
+import os
+os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"
+
+app = FastAPI()
+
+# Load model and data
+embeddings = np.load("data/embeddings.npy")
+with open("data/image_urls.pkl", "rb") as f:
+    image_urls = pickle.load(f)
+product_data = pd.read_csv("data/product_data.csv")
+
+fe = FeatureExtractor()
+index = FaissIndex(dim=embeddings.shape[1])
+index.build(embeddings, image_urls)
+
+@app.post("/recommend")
+async def recommend(file: UploadFile = File(...), threshold: float = 0.8, k: int = 100):
+    try:
+        image = Image.open(BytesIO(await file.read())).convert("RGB")
+        user_emb = fe.extract(image)
+        results = index.search(user_emb, threshold=threshold, k=k)
+
+        if not results:
+            return JSONResponse({"message": "No similar products found"}, status_code=404)
+
+        input_url = results[0][0]
+        input_row = product_data[product_data['IMAGE'] == input_url]
+
+        input_group_id = input_row['GROUP_ID'].values[0] if not input_row.empty else None
+        input_product_name = input_row['PRODUCT_NAME'].values[0] if not input_row.empty else None
+
+        # Filtering logic
+        filtered = []
+        for url, sim in results:
+            row = product_data[product_data['IMAGE'] == url]
+            group_id = row['GROUP_ID'].values[0] if not row.empty else None
+            product_name = row['PRODUCT_NAME'].values[0] if not row.empty else None
+
+            if (input_group_id is None or input_group_id == 0):
+                if product_name != input_product_name:
+                    filtered.append((url, sim))
+            else:
+                if group_id != input_group_id:
+                    filtered.append((url, sim))
+
+        # De-duplicate by product name
+        seen = set()
+        final = []
+        for url, sim in filtered:
+            row = product_data[product_data['IMAGE'] == url]
+            product_name = row['PRODUCT_NAME'].values[0] if not row.empty else None
+            if product_name and product_name not in seen:
+                seen.add(product_name)
+                brand_name = row['BRAND_NAME'].values[0] if 'BRAND_NAME' in row else "Unknown"
+                final.append({
+                    "brand_name": brand_name,
+                    "product_name": product_name,
+                    "image_url": url,
+                    "similarity_score": float(f"{sim:.4f}")
+                })
+
+        return {"recommendations": final[:15]}
+
+    except Exception as e:
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

model/feature_extractor.py

@@ -0,0 +1,38 @@
+import torch
+import torchvision.models as models
+from torchvision import transforms
+from PIL import Image
+import numpy as np
+
+class FeatureExtractor:
+    def __init__(self):
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        # Load pretrained ResNet50 without the final classification layer
+        resnet = models.resnet50(pretrained=True)
+        # Remove the final fully connected layer (fc)
+        self.model = torch.nn.Sequential(*list(resnet.children())[:-1])
+        self.model.eval().to(self.device)
+
+        # Standard ImageNet preprocessing
+        self.transform = transforms.Compose([
+            transforms.Resize(256),
+            transforms.CenterCrop(224),
+            transforms.ToTensor(),
+            transforms.Normalize(
+                mean=[0.485, 0.456, 0.406],
+                std=[0.229, 0.224, 0.225]
+            ),
+        ])
+
+    def extract(self, image: Image.Image):
+        image = self.transform(image).unsqueeze(0).to(self.device)
+        with torch.no_grad():
+            features = self.model(image)
+        features = features.squeeze().cpu().numpy()
+        features = features.reshape(-1)  # flatten (2048,)
+
+        # Normalize to unit vector (important for cosine similarity)
+        norm = np.linalg.norm(features)
+        if norm > 0:
+            features = features / norm
+        return features

precompute_embeddings.py

@@ -0,0 +1,31 @@
+import pandas as pd
+import numpy as np
+import pickle
+from tqdm import tqdm
+from model.feature_extractor import FeatureExtractor
+from utils.image_utils import load_image_from_url
+
+def main():
+    df = pd.read_csv("data/product_data.csv")
+    fe = FeatureExtractor()
+    embeddings = []
+    valid_urls = []
+
+    # tqdm wraps the iterable and shows progress bar with description
+    for url in tqdm(df['IMAGE_URL'], desc="Extracting embeddings"):
+        img = load_image_from_url(url)
+        if img is not None:
+            emb = fe.extract(img)
+            embeddings.append(emb)
+            valid_urls.append(url)
+
+    embeddings = np.array(embeddings)
+    np.save("data/embeddings.npy", embeddings)
+
+    with open("data/image_urls.pkl", "wb") as f:
+        pickle.dump(valid_urls, f)
+
+    print(f"Saved {len(valid_urls)} embeddings and URLs.")
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,12 @@
+streamlit
+torch
+torchvision
+timm
+faiss-cpu
+pandas
+Pillow
+requests
+tqdm
+numpy
+fastapi
+uvicorn

streamlit_app.py

@@ -0,0 +1,164 @@
+import os
+os.environ["KMP_DUPLICATE_LIB_OK"] = "TRUE"
+
+import streamlit as st
+from model.feature_extractor import FeatureExtractor
+from utils.faiss_index import FaissIndex
+from PIL import Image
+import pandas as pd
+import numpy as np
+import pickle
+import streamlit.components.v1 as components
+
+st.set_page_config(page_title="🛍️ Product Recommender", layout="wide")
+
+@st.cache_resource
+def load_resources():
+    embeddings = np.load("data/embeddings.npy")
+    with open("data/image_urls.pkl", "rb") as f:
+        image_urls = pickle.load(f)
+    product_data = pd.read_csv("data/product_data.csv")
+    fe = FeatureExtractor()
+    index = FaissIndex(dim=embeddings.shape[1])
+    index.build(embeddings, image_urls)
+    return fe, index, image_urls, product_data
+
+fe, index, image_urls, product_data = load_resources()
+
+st.title("🛍️ Product Image Recommender")
+
+uploaded_file = st.file_uploader("Upload a product image", type=["jpg", "jpeg", "png"])
+
+if uploaded_file:
+    user_img = Image.open(uploaded_file).convert("RGB")
+    st.image(user_img, caption="Uploaded Image", width=250)
+
+    user_emb = fe.extract(user_img)
+    results = index.search(user_emb, threshold=0.8, k=100)
+
+    if len(results) > 0:
+        input_image_url = results[0][0]
+
+        # Get GROUP_ID of uploaded image
+        input_group_id_series = product_data.loc[product_data['IMAGE'] == input_image_url, 'GROUP_ID']
+        input_group_id = input_group_id_series.values[0] if not input_group_id_series.empty else None
+
+        # Get PRODUCT_NAME of uploaded image
+        input_product_name_series = product_data.loc[product_data['IMAGE'] == input_image_url, 'PRODUCT_NAME']
+        input_product_name = input_product_name_series.values[0] if not input_product_name_series.empty else None
+
+        # st.markdown(f"**GROUP_ID of uploaded image:** `{input_group_id}`")
+
+        filtered_results = []
+        for url, sim in results:
+            group_id_series = product_data.loc[product_data['IMAGE'] == url, 'GROUP_ID']
+            group_id = group_id_series.values[0] if not group_id_series.empty else None
+
+            product_series = product_data.loc[product_data['IMAGE'] == url, 'PRODUCT_NAME']
+            product_name = product_series.values[0] if not product_series.empty else None
+
+            # Rule: if GROUP_ID is None or 0, exclude same product name
+            if (input_group_id is None or input_group_id == 0):
+                if product_name != input_product_name:
+                    filtered_results.append((url, sim))
+            else:
+                if group_id != input_group_id:
+                    filtered_results.append((url, sim))
+
+        seen_products = set()
+        deduped_results = []
+        for url, sim in filtered_results:
+            product_series = product_data.loc[product_data['IMAGE'] == url, 'PRODUCT_NAME']
+            product_name = product_series.values[0] if not product_series.empty else None
+            if product_name and product_name not in seen_products:
+                seen_products.add(product_name)
+                deduped_results.append((url, sim))
+
+        top_results = deduped_results[:15]
+
+        cards_html = ""
+        for url, sim in top_results:
+            brand = product_data.loc[product_data['IMAGE'] == url, 'BRAND_NAME'].values
+            product = product_data.loc[product_data['IMAGE'] == url, 'PRODUCT_NAME'].values
+            brand_name = brand[0] if len(brand) > 0 else "Unknown Brand"
+            product_name = product[0] if len(product) > 0 else "Unknown Product"
+            cards_html += f"""
+                <div class="card">
+                    <img src="{url}" alt="Product Image"/>
+                    <div class="info">
+                        <h4>{brand_name}</h4>
+                        <p>{product_name}</p>
+                        <span>Similarity: {sim:.2f}</span>
+                    </div>
+                </div>
+            """
+
+        full_html = f"""
+        <style>
+            .carousel-wrapper {{
+                overflow-x: auto;
+                overflow-y: visible;  /* allow vertical overflow if any */
+                white-space: nowrap;
+                padding: 20px 16px 40px 16px;
+                height: auto;
+                scroll-behavior: smooth;
+            }}
+            .carousel {{
+                display: flex;
+                gap: 10px;
+                align-items: stretch;  /* all cards same height */
+            }}
+            .card {{
+                flex: 0 0 auto;
+                width: 280px;  /* 1.5x wider */
+                /* no fixed height */
+                border: 1px solid #ddd;
+                border-radius: 14px;
+                padding: 14px;
+                background: #fff;
+                box-shadow: 0 4px 12px rgba(0,0,0,0.1);
+                text-align: center;
+                box-sizing: border-box;
+                transition: transform 0.2s ease-in-out;
+                font-family: "Segoe UI", sans-serif;
+            }}
+            .card:hover {{
+                transform: scale(1.04);
+                box-shadow: 0 6px 16px rgba(0,0,0,0.12);
+            }}
+            .card img {{
+                width: 100%;
+                height: 300px;  /* 1.5x taller */
+                object-fit: cover;
+                border-radius: 8px;
+            }}
+            .info h4 {{
+                font-size: 20px;
+                margin: 12px 0 6px;
+                color: #222;
+                white-space: normal;
+            }}
+            .info p {{
+                font-size: 16px;
+                margin: 0 0 8px;
+                color: #555;
+                white-space: normal;
+            }}
+            .info span {{
+                font-size: 13px;
+                color: #888;
+            }}
+        </style>
+
+        <div class="carousel-wrapper">
+            <div class="carousel">
+                {cards_html}
+            </div>
+        </div>
+        """
+
+        st.subheader("🔍 Recommended Products")
+        components.html(full_html, height=600, scrolling=False)
+
+    else:
+        st.info("✨ No visually similar items found — this might be a one-of-a-kind product!")

utils/faiss_index.py

@@ -0,0 +1,26 @@
+import faiss
+import numpy as np
+
+class FaissIndex:
+    def __init__(self, dim):
+        self.index = faiss.IndexFlatIP(dim)
+        self.image_map = []
+
+    def build(self, embeddings, image_ids):
+        norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
+        normalized_embeddings = embeddings / (norms + 1e-10)
+        self.index.add(normalized_embeddings.astype('float32'))
+        self.image_map = image_ids
+
+    def search(self, query_vector, threshold=0.8, k=50):
+        query_norm = np.linalg.norm(query_vector)
+        if query_norm > 0:
+            query_vector = query_vector / query_norm
+        query = np.array([query_vector]).astype('float32')
+        similarities, indices = self.index.search(query, k)
+
+        results = []
+        for i, sim in zip(indices[0], similarities[0]):
+            if sim >= threshold:
+                results.append((self.image_map[i], sim))
+        return sorted(results, key=lambda x: -x[1])

utils/image_utils.py

@@ -0,0 +1,13 @@
+import requests
+from io import BytesIO
+from PIL import Image
+
+def load_image_from_url(url):
+    try:
+        response = requests.get(url, timeout=10)
+        response.raise_for_status()
+        image = Image.open(BytesIO(response.content)).convert('RGB')
+        return image
+    except Exception as e:
+        print(f"Failed to load image from {url}: {e}")
+        return None