bahawal.baloch
/
listing-radar
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'main' into AMB_DEV

AMB_DEV
ahmedmujtaba-gif 2026-05-08 20:29:46 +05:00
parent 56f52a4edc 0fe92f182f
commit 6ed79621b0
6 changed files with 458 additions and 311 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -5,5 +5,5 @@
**pycache** **pycache**
*agent/** *agent/**
**downloaded_images** **downloaded_images**
__pycache__/ **model_export**
*.pyc **cpython**

212
dev_backend/vector_db_router/models.py
View File

@ -1,104 +1,108 @@
from qdrant_client import AsyncQdrantClient, models from qdrant_client import AsyncQdrantClient, models
from qdrant_client.models import PointStruct from qdrant_client.models import PointStruct
from typing import Dict, Any from typing import Dict, Any
import logging
class CollectionHandler: logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
def __init__(self, collection_name: str, vector: Any, vector_size: int, payload: Dict, log = logging.getLogger(__name__)
id: int=None,
link: str=None, class CollectionHandler:
asin: str=None, def __init__(self, collection_name: str, vector: Any, vector_size: int, payload: Dict,
category: str=None, id: int=None,
brand: str=None, link: str=None,
client: AsyncQdrantClient=None asin: str=None,
): category: str=None,
self.collection_name = collection_name brand: str=None,
self.vector = vector client: AsyncQdrantClient=None
self.id = id ):
self.vector_size = vector_size self.collection_name = collection_name
self.payload = payload self.vector = vector
self.link = link self.id = id
self.asin = asin self.vector_size = vector_size
self.category = category self.payload = payload
self.brand = brand self.link = link
self.client = client if client else AsyncQdrantClient("localhost", port=6333) self.asin = asin
self.category = category
async def create_collection(self): self.brand = brand
try: self.client = client if client else AsyncQdrantClient("localhost", port=6333)
if await self.client.collection_exists(self.collection_name):
return {"message": "Collection already exists"} async def create_collection(self):
try:
await self.client.create_collection( if await self.client.collection_exists(self.collection_name):
collection_name=self.collection_name, return {"message": "Collection already exists"}
vectors_config=models.VectorParams(size=self.vector_size, distance=models.Distance.EUCLID),
optimizers_config=models.OptimizersConfigDiff(indexing_threshold=20000) await self.client.create_collection(
) collection_name=self.collection_name,
vectors_config=models.VectorParams(size=self.vector_size, distance=models.Distance.EUCLID),
# Creating payload indexes as per project logic optimizers_config=models.OptimizersConfigDiff(indexing_threshold=20000)
)
await self.client.create_payload_index(
collection_name=self.collection_name, # Creating payload indexes as per project logic
field_name="link",
field_schema=models.PayloadSchemaType.KEYWORD await self.client.create_payload_index(
) collection_name=self.collection_name,
await self.client.create_payload_index( field_name="link",
collection_name=self.collection_name, field_schema=models.PayloadSchemaType.KEYWORD
field_name="title", )
field_schema=models.PayloadSchemaType.KEYWORD await self.client.create_payload_index(
) collection_name=self.collection_name,
field_name="title",
await self.client.create_payload_index( field_schema=models.PayloadSchemaType.KEYWORD
collection_name=self.collection_name, )
field_name="brand",
field_schema=models.PayloadSchemaType.KEYWORD await self.client.create_payload_index(
) collection_name=self.collection_name,
await self.client.create_payload_index( field_name="brand",
collection_name=self.collection_name, field_schema=models.PayloadSchemaType.KEYWORD
field_name="asin", )
field_schema=models.PayloadSchemaType.KEYWORD await self.client.create_payload_index(
) collection_name=self.collection_name,
field_name="asin",
field_schema=models.PayloadSchemaType.KEYWORD
return {"message": f"Collection {self.collection_name} created successfully"} )
except Exception as e:
return {"message": str(e)}
return {"message": f"Collection {self.collection_name} created successfully"}
async def insertion(self): except Exception as e:
try: return {"message": str(e)}
await self.client.upsert(
collection_name=self.collection_name, async def insertion(self):
points=[ try:
PointStruct(id=self.id, vector=self.vector, payload=self.payload) await self.client.upsert(
] collection_name=self.collection_name,
) points=[
return True PointStruct(id=self.id, vector=self.vector, payload=self.payload)
except Exception as e: ]
# Note: In a real app we'd use a logger here )
print(f"Insertion failed for ID {self.id}: {e}") return True
return False except Exception as e:
# Note: In a real app we'd use a logger here
async def upsert_point(self): print(f"Insertion failed for ID {self.id}: {e}")
return await self.insertion() return False
async def search(self, query_vector): async def upsert_point(self):
try: return await self.insertion()
result = await self.client.search(
collection_name=self.collection_name, async def search(self, query_vector, score_threshold: float = 0.3, limit: int = 10):
query_vector=query_vector, try:
limit=10 result = await self.client.query_points(
) collection_name=self.collection_name,
return result query=query_vector,
except Exception as e: score_threshold=score_threshold,
print("Search failed: ", e) limit=limit
return None )
return result.points
async def update_collection(self): except Exception as e:
"""Update is implemented as an upsert of the point data.""" log.error(f"Search failed: {e}")
return await self.upsert_point() return None
async def delete_collection(self): async def update_collection(self):
try: """Update is implemented as an upsert of the point data."""
await self.client.delete_collection(collection_name=self.collection_name) return await self.upsert_point()
return {"message": f"Collection {self.collection_name} deleted successfully"}
except Exception as e: async def delete_collection(self):
return {"message": str(e)} try:
await self.client.delete_collection(collection_name=self.collection_name)
return {"message": f"Collection {self.collection_name} deleted successfully"}
except Exception as e:
return {"message": str(e)}

90
dev_backend/vector_db_router/plugins.py Normal file
View File

@ -0,0 +1,90 @@
import os
import requests
from urllib.parse import urlparse
from pathlib import Path
from PIL import Image
def download_image(url: str, filename: str = None) -> str:
"""
Download an image from URL and save it in data/temp/ folder.
Args:
url (str): Image URL
filename (str, optional): Custom filename. If None, extracted from URL.
Returns:
str: Full path to the downloaded image
"""
try:
# Get project root directory (where your main script is)
root_dir = Path(os.path.dirname(os.path.abspath(__file__))).parent
# Create data/temp folder structure
temp_dir = root_dir / "data" / "temp"
temp_dir.mkdir(parents=True, exist_ok=True)
# Generate filename if not provided
if not filename:
parsed_url = urlparse(url)
filename = os.path.basename(parsed_url.path)
if not filename or "." not in filename:
# Fallback filename
ext = filename.split('.')[-1] if '.' in filename else 'jpg'
filename = f"image_{hash(url) % 100000}.{ext}"
# Ensure filename has extension
if '.' not in filename:
filename += ".jpg"
file_path = temp_dir / filename
# Download the image
response = requests.get(url, stream=True, timeout=30)
response.raise_for_status()
# Save image
with open(file_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
print(f"✅ Image downloaded: {file_path}")
return str(file_path)
except Exception as e:
print(f"❌ Failed to download image: {e}")
raise
def read_image(image_path: str) -> Image.Image:
"""
Read an image from the given path and return a PIL Image object.
Args:
image_path (str): Path to the image file
Returns:
PIL.Image.Image: Loaded image
Raises:
FileNotFoundError: If image doesn't exist
Exception: For other image loading errors
"""
try:
if not os.path.exists(image_path):
raise FileNotFoundError(f"Image not found at path: {image_path}")
# Open the image
image = Image.open(image_path)
# Convert to RGB (important for DINOv2 and most models)
if image.mode != "RGB":
image = image.convert("RGB")
print(f"✅ Image loaded successfully: {image_path} | Size: {image.size}")
return image
except FileNotFoundError as e:
print(f"❌ File not found: {e}")
raise
except Exception as e:
print(f"❌ Failed to read image: {e}")
raise

46
dev_backend/vector_db_router/serializers.py
View File

@ -1,22 +1,24 @@
from pydantic import BaseModel from pydantic import BaseModel
from typing import Dict, List, Any from typing import Dict, List, Any
class CreateCollectionSerializer(BaseModel): class CreateCollectionSerializer(BaseModel):
collection_name: str collection_name: str
vector: List[float] vector: List[float]
vector_size: int vector_size: int
payload: Dict[str, Any] payload: Dict[str, Any]
id: int id: int
class QueryCollectionSerializer(BaseModel): class QueryCollectionSerializer(BaseModel):
collection_name: str collection_name: str
query_vector: List[float] url: str
score_threshold: float = 0.3 # Euclidean distance — lower = more similar. 0.3 = very tight match
class UpdateCollectionSerializer(BaseModel): limit: int = 10
collection_name: str
vector: List[float] class UpdateCollectionSerializer(BaseModel):
payload: Dict[str, Any] collection_name: str
id: int vector: List[float]
payload: Dict[str, Any]
class DeleteCollectionSerializer(BaseModel): id: int
collection_name: str
class DeleteCollectionSerializer(BaseModel):
collection_name: str

414
dev_backend/vector_db_router/views.py
View File

@ -1,183 +1,233 @@
from db_setup import get_qdrant_client from db_setup import get_qdrant_client
from typing import Annotated from typing import Annotated
from fastapi import Depends, HTTPException, APIRouter from fastapi import Depends, HTTPException, APIRouter
from qdrant_client import AsyncQdrantClient from qdrant_client import AsyncQdrantClient
from fastapi.responses import JSONResponse from .plugins import download_image,read_image
from .serializers import ( from fastapi.responses import JSONResponse
CreateCollectionSerializer, from .serializers import (
QueryCollectionSerializer, CreateCollectionSerializer,
UpdateCollectionSerializer, QueryCollectionSerializer,
DeleteCollectionSerializer UpdateCollectionSerializer,
) DeleteCollectionSerializer
from model_export.dino_image_matching import get_vectors )
from .models import CollectionHandler from model_export.dino_image_matching import get_vectors,get_embedding
import os from .models import CollectionHandler
app_router = APIRouter() import os
import logging app_router = APIRouter()
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') import logging
log = logging.getLogger(__name__) logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
log = logging.getLogger(__name__)
import pandas as pd
import pandas as pd
@app_router.get("/get_vectors")
async def get_vectors_endpoint( @app_router.get("/get_vectors")
q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)], async def get_vectors_endpoint(
image_path:str=os.getenv("DATASET") q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)],
): image_path:str=os.getenv("DATASET")
try: ):
# Construct path relative to this file try:
base_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # Construct path relative to this file
excel_path = os.path.join(base_root, "model_export", "listing_data.xlsx") base_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
df = pd.read_excel(excel_path) excel_path = os.path.join(base_root, "model_export", "listing_data.xlsx")
asin_list = df['ASIN'].dropna().astype(str).tolist() df = pd.read_excel(excel_path)
log.info(f"Generating vectors and ingesting {len(asin_list)} ASINs into Qdrant from {excel_path}") asin_list = df['ASIN'].dropna().astype(str).tolist()
log.info(f"Generating vectors and ingesting {len(asin_list)} ASINs into Qdrant from {excel_path}")
# 1. Initialize/Create collection "Product"
# DINOv2 vitb14 vector size is 768 # 1. Initialize/Create collection "Product"
init_handler = CollectionHandler( # DINOv2 vitb14 vector size is 768
collection_name="Product", init_handler = CollectionHandler(
vector=[], collection_name="Product",
vector_size=768, vector=[],
payload={}, vector_size=768,
client=q payload={},
) client=q
await init_handler.create_collection() )
await init_handler.create_collection()
result_lst = []
for index, row in df.iterrows(): result_lst = []
asin = str(row['ASIN']) for index, row in df.iterrows():
if pd.isna(row['ASIN']): asin = str(row['ASIN'])
continue if pd.isna(row['ASIN']):
continue
title = str(row['Title'])
brand = str(row['Brand']) title = str(row['Title'])
link = str(row['Image']) brand = str(row['Brand'])
link = str(row['Image'])
# Call get_vectors
vector = get_vectors(image_input=image_path, item=asin) # Call get_vectors
vector = get_vectors(image_input=image_path, item=asin)
if vector is None:
log.warning(f"Skipping {asin} due to missing image/vector") if vector is None:
continue log.warning(f"Skipping {asin} due to missing image/vector")
continue
payload = {
"asin": asin, payload = {
"title": title, "asin": asin,
"brand": brand, "title": title,
"link": link "brand": brand,
} "link": link
}
# 2. Ingest into Qdrant using CollectionHandler
# Use the injected client 'q' and convert index to int # 2. Ingest into Qdrant using CollectionHandler
handler = CollectionHandler( # Use the injected client 'q' and convert index to int
collection_name="Product", handler = CollectionHandler(
vector=vector, collection_name="Product",
vector_size=768, vector=vector,
payload=payload, vector_size=768,
id=int(index), payload=payload,
link=link, id=int(index),
asin=asin, link=link,
brand=brand, asin=asin,
client=q brand=brand,
) client=q
success = await handler.upsert_point() )
success = await handler.upsert_point()
if success:
result_lst.append({ if success:
"item": asin, result_lst.append({
"status": "ingested", "item": asin,
"payload": payload "status": "ingested",
}) "payload": payload
log.info(f"Vector ingested for {asin} (ID: {index})") })
else: log.info(f"Vector ingested for {asin} (ID: {index})")
log.error(f"Failed to ingest vector for {asin}") else:
log.error(f"Failed to ingest vector for {asin}")
return JSONResponse({"status": "success", "message": f"Ingested {len(result_lst)} items into Product collection", "result": result_lst})
except Exception as e: return JSONResponse({"status": "success", "message": f"Ingested {len(result_lst)} items into Product collection", "result": result_lst})
raise HTTPException(status_code=500, detail=str(e)) except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app_router.post("/create")
async def create_collection_endpoint( @app_router.post("/create")
q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)], async def create_collection_endpoint(
body: CreateCollectionSerializer = None q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)],
): body: CreateCollectionSerializer = None
try: ):
if body is None: try:
raise HTTPException(status_code=400, detail="Collection name is required") if body is None:
raise HTTPException(status_code=400, detail="Collection name is required")
print("collection_name: ", body.collection_name)
print("collection_name: ", body.collection_name)
handler = CollectionHandler(
collection_name=body.collection_name, handler = CollectionHandler(
vector=body.vector, collection_name=body.collection_name,
vector_size=body.vector_size, vector=body.vector,
payload=body.payload, vector_size=body.vector_size,
id=body.id payload=body.payload,
) id=body.id
)
# 1. Create collection
result = await handler.create_collection() # 1. Create collection
result = await handler.create_collection()
# 2. Automatically call upsert_point
await handler.upsert_point() # 2. Automatically call upsert_point
await handler.upsert_point()
return JSONResponse(result)
return JSONResponse(result)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app_router.get("/query")
async def query_collection_endpoint( @app_router.get("/query")
q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)], async def query_collection_endpoint(
body: QueryCollectionSerializer q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)],
): body: QueryCollectionSerializer
try: ):
handler = CollectionHandler( try:
collection_name=body.collection_name, result = []
vector=body.query_vector, if isinstance(body.url, str):
vector_size=len(body.query_vector), # Handle semicolon-separated URLs by taking the first one
payload={}, target_url = body.url.split(';')[0].strip() if ';' in body.url else body.url
id=0 log.info(f"Querying collection {body.collection_name} with URL: {target_url}")
) downloaded_image_path = download_image(target_url)
result = await handler.search(body.query_vector) query_vector = get_embedding(downloaded_image_path)
return JSONResponse({"results": str(result)}) # get_embedding already returns a flat list of 768 floats
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) handler = CollectionHandler(
collection_name=body.collection_name,
@app_router.put("/update") vector=query_vector,
async def update_collection_endpoint( vector_size=len(query_vector),
q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)], payload={},
body: UpdateCollectionSerializer id=0,
): client=q
try: )
handler = CollectionHandler( search_result = await handler.search(
collection_name=body.collection_name, query_vector,
vector=body.vector, score_threshold=body.score_threshold,
vector_size=len(body.vector), limit=body.limit
payload=body.payload, )
id=body.id if search_result:
) result = [
result = await handler.update_collection() {"id": p.id, "score": p.score, "payload": p.payload}
return JSONResponse({"status": "success", "result": result}) for p in search_result
except Exception as e: ]
raise HTTPException(status_code=500, detail=str(e)) else:
result = [] # No match within threshold
@app_router.delete("/delete")
async def delete_collection_endpoint( elif isinstance(body.url, list):
q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)], result = []
body: DeleteCollectionSerializer for url in body.url:
): downloaded_image_path = download_image(url)
try: query_vector = get_embedding(downloaded_image_path)
handler = CollectionHandler( # get_embedding already returns a flat list of 768 floats
collection_name=body.collection_name,
vector=[], handler = CollectionHandler(
vector_size=0, collection_name=body.collection_name,
payload={}, vector=query_vector,
id=0 vector_size=len(query_vector),
) payload={},
result = await handler.delete_collection() id=0,
return JSONResponse(result) client=q
except Exception as e: )
raise HTTPException(status_code=500, detail=str(e)) search_result = await handler.search(
query_vector,
score_threshold=body.score_threshold,
limit=body.limit
)
if search_result:
result.append([
{"id": p.id, "score": p.score, "payload": p.payload}
for p in search_result
])
return JSONResponse({"results": result})
except Exception as e:
log.error(f"Query failed: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app_router.put("/update")
async def update_collection_endpoint(
q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)],
body: UpdateCollectionSerializer
):
try:
handler = CollectionHandler(
collection_name=body.collection_name,
vector=body.vector,
vector_size=len(body.vector),
payload=body.payload,
id=body.id
)
result = await handler.update_collection()
return JSONResponse({"status": "success", "result": result})
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app_router.delete("/delete")
async def delete_collection_endpoint(
q: Annotated[AsyncQdrantClient, Depends(get_qdrant_client)],
body: DeleteCollectionSerializer
):
try:
handler = CollectionHandler(
collection_name=body.collection_name,
vector=[],
vector_size=0,
payload={},
id=0
)
result = await handler.delete_collection()
return JSONResponse(result)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))

3
model_export/dino_image_matching.py
View File

@ -46,7 +46,8 @@ def get_embedding(image_path):
# Normalize embedding (important for cosine similarity) # Normalize embedding (important for cosine similarity)
embedding = F.normalize(embedding, p=2, dim=1) embedding = F.normalize(embedding, p=2, dim=1)
return embedding.cpu() # Return flat list (squeeze batch dim)
return embedding.squeeze(0).cpu().tolist()
def get_vectors(image_input, item): def get_vectors(image_input, item):
try: try: