Update nutrition score retrieval using OFF v2 REST API and title cleaning

This commit is contained in:
sriram
2026-08-10 18:54:27 +05:30
parent 72dd9f8296
commit 370f867355
43 changed files with 2088 additions and 373 deletions

View File

@@ -19,12 +19,37 @@ from app.services.vector_store import (
router = APIRouter(tags=["catalog"])
def _clean_url(url: Optional[str]) -> Optional[str]:
if not url:
return None
return str(url).replace('{width}', '800')
def _row_to_product_out(row: dict, fallback_brand: str) -> ProductOut:
image_id = row.get("image_id") or ""
brand_name = row.get("brand") or fallback_brand
db_single = _clean_url(row.get("image_url"))
db_list = [_clean_url(u) for u in (row.get("image_urls") or []) if u]
final_urls = db_list
if not final_urls and db_single:
final_urls = [db_single]
if not final_urls and s3_service.enabled:
s3_list = s3_service.get_product_image_urls(brand_name, image_id)
if s3_list:
final_urls = s3_list
primary_url = (final_urls[0] if final_urls else None) or db_single
if not primary_url and s3_service.enabled:
primary_url = s3_service.get_product_image_url(brand_name, image_id)
return ProductOut(
image_id=image_id,
image_url=s3_service.get_product_image_url(fallback_brand, image_id) or None,
brand=fallback_brand,
image_url=primary_url,
image_urls=final_urls,
brand=brand_name,
product_name=row.get("product_name") or row.get("title") or "Unknown product",
title=row.get("title") or row.get("product_name") or None,
category=row.get("category"),

View File

@@ -0,0 +1,90 @@
from __future__ import annotations
import logging
import os
import subprocess
import threading
from pathlib import Path
from typing import Any, Dict
from fastapi import APIRouter, BackgroundTasks
from pydantic import BaseModel
from app.services.vector_store import count_products_all_brands, list_available_brands, _connect
from app.services.store_db import list_stores
from app.services.ollama_service import _ensure_client
logger = logging.getLogger(__name__)
router = APIRouter(tags=["system"])
BASE_DIR = Path(__file__).resolve().parents[3]
FRONTEND_DIST = BASE_DIR.parent / "frontend" / "dist"
class SystemStatusOut(BaseModel):
status: str
database_connected: bool
total_products: int
available_brands: list[str]
total_stores: int
ollama_connected: bool
frontend_dist_exists: bool
def _run_background_auto_seed():
"""Background task to run initial seeding asynchronously if DB is empty."""
try:
if count_products_all_brands() == 0:
logger.info("⚡ Background Auto-Init: Database empty. Running initial sample seed...")
cmd_seed = [os.sys.executable, str(BASE_DIR / "scripts" / "seed_sample_data.py"), "--skip-if-seeded"]
subprocess.run(cmd_seed, check=False)
logger.info("⚡ Background Auto-Init: Provisioning store intelligence...")
cmd_store = [os.sys.executable, str(BASE_DIR / "scripts" / "seed_store_intelligence.py"), "--skip-if-seeded"]
subprocess.run(cmd_store, check=False)
logger.info("✅ Background Auto-Init complete!")
except Exception as e:
logger.error("Background Auto-Init error: %s", e)
@router.get("/system/status", response_model=SystemStatusOut)
def get_system_status() -> SystemStatusOut:
"""Return unified status of database, vector store, stores, and frontend build."""
db_connected = False
products_count = 0
brands = []
stores_count = 0
try:
conn = _connect()
if conn:
db_connected = True
conn.close()
products_count = count_products_all_brands()
brands = list_available_brands()
stores_count = len(list_stores())
except Exception:
pass
ollama_ok = _ensure_client()
dist_ok = FRONTEND_DIST.exists() and (FRONTEND_DIST / "index.html").exists()
return SystemStatusOut(
status="ok" if db_connected else "degraded",
database_connected=db_connected,
total_products=products_count,
available_brands=brands,
total_stores=stores_count,
ollama_connected=ollama_ok,
frontend_dist_exists=dist_ok,
)
@router.post("/system/init")
def initialize_system(background_tasks: BackgroundTasks) -> Dict[str, Any]:
"""Trigger background auto-initialization of sample catalog and store data."""
background_tasks.add_task(_run_background_auto_seed)
return {
"status": "started",
"message": "Background initialization triggered. Check /api/system/status for progress.",
}

View File

@@ -0,0 +1,447 @@
from __future__ import annotations
import io
import uuid
import logging
import pandas as pd
from typing import Any, Dict, List, Optional
from datetime import datetime
from fastapi import APIRouter, File, HTTPException, UploadFile, Response
from fastapi.responses import PlainTextResponse
from app.services.vector_store import _connect
from app.services import store_db, nutrition_db
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/upload", tags=["upload"])
def _normalize_col(col: str) -> str:
"""Normalize dataframe column names (lower, strip, replace spaces/hyphens with underscore)."""
return str(col).strip().lower().replace(' ', '_').replace('-', '_')
def read_df_from_upload(filename: str, contents: bytes) -> pd.DataFrame:
"""Parse CSV or Excel (xlsx/xls) upload file into a pandas DataFrame."""
fn_lower = filename.lower()
if fn_lower.endswith('.xlsx') or fn_lower.endswith('.xls'):
df = pd.read_excel(io.BytesIO(contents))
elif fn_lower.endswith('.tsv'):
df = pd.read_csv(io.BytesIO(contents), sep='\t')
else:
try:
df = pd.read_csv(io.BytesIO(contents))
except Exception:
df = pd.read_csv(io.BytesIO(contents), sep=None, engine='python')
# Rename columns to normalized format
df.columns = [_normalize_col(c) for c in df.columns]
return df
def _get_str(row: dict, keys: List[str], default: str = "") -> str:
for k in keys:
if k in row and pd.notna(row[k]):
val = str(row[k]).strip()
if val:
return val
return default
def _get_float(row: dict, keys: List[str], default: float = 0.0) -> float:
for k in keys:
if k in row and pd.notna(row[k]):
try:
return float(row[k])
except (ValueError, TypeError):
pass
return default
def _get_int(row: dict, keys: List[str], default: int = 0) -> int:
for k in keys:
if k in row and pd.notna(row[k]):
try:
return int(float(row[k]))
except (ValueError, TypeError):
pass
return default
# ---------------------------------------------------------------------------
# Stores Inventory Excel / CSV Upload
# ---------------------------------------------------------------------------
@router.post("/stores")
@router.post("/stores/upload")
async def upload_stores_file(file: UploadFile = File(...)) -> Dict[str, Any]:
if not file.filename:
raise HTTPException(status_code=400, detail="No file uploaded")
contents = await file.read()
try:
df = read_df_from_upload(file.filename, contents)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Could not parse Excel/CSV file: {e}")
if df.empty:
raise HTTPException(status_code=400, detail="Uploaded file contains no data rows")
conn = _connect()
if not conn:
raise HTTPException(status_code=500, detail="Database connection failed")
imported_count = 0
stores_created = set()
try:
with conn, conn.cursor() as cur:
# Ensure tables exist
store_db.ensure_store_intelligence_schema()
for _, r in df.iterrows():
row = r.to_dict()
store_id = _get_str(row, ['store_id', 'store'], 'store_mumbai_1')
brand = _get_str(row, ['brand', 'brand_name'], 'amul').lower()
product_name = _get_str(row, ['product_name', 'title', 'name', 'item'], 'Product Item')
image_id = _get_str(row, ['image_id', 'sku', 'product_sku', 'item_id'], '')
if not image_id:
image_id = f"{brand}_{product_name.lower().replace(' ', '_')}"
category = _get_str(row, ['category', 'cat'], 'Dairy')
avail_stock = _get_int(row, ['available_stock', 'stock', 'qty', 'quantity'], 50)
reserved_stock = _get_int(row, ['reserved_stock', 'reserved'], 0)
reorder_lvl = _get_int(row, ['reorder_level', 'reorder'], 15)
safety_stk = _get_int(row, ['safety_stock', 'safety'], 10)
mrp = _get_float(row, ['mrp', 'price'], 100.0)
cost_price = _get_float(row, ['cost_price', 'cost'], 70.0)
selling_price = _get_float(row, ['selling_price', 'sell_price'], mrp * 0.9 if mrp else 90.0)
# 1. Ensure store exists
cur.execute(
"""
INSERT INTO stores (store_id, store_name, city, tier, footfall_index)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (store_id) DO NOTHING
""",
(store_id, store_id.replace('_', ' ').title(), 'Mumbai', 'standard', 25.0)
)
stores_created.add(store_id)
# 2. Upsert store_inventory
cur.execute(
"""
INSERT INTO store_inventory
(store_id, brand, image_id, title, category, available_stock, reserved_stock, reorder_level, safety_stock)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (store_id, brand, image_id) DO UPDATE SET
title = EXCLUDED.title,
category = EXCLUDED.category,
available_stock = EXCLUDED.available_stock,
reserved_stock = EXCLUDED.reserved_stock,
reorder_level = EXCLUDED.reorder_level,
safety_stock = EXCLUDED.safety_stock,
updated_at = CURRENT_TIMESTAMP
""",
(store_id, brand, image_id, product_name, category, avail_stock, reserved_stock, reorder_lvl, safety_stk)
)
# 3. Upsert store_prices
cur.execute(
"""
INSERT INTO store_prices (store_id, brand, image_id, mrp, cost_price, selling_price)
VALUES (%s, %s, %s, %s, %s, %s)
ON CONFLICT (store_id, brand, image_id) DO UPDATE SET
mrp = EXCLUDED.mrp,
cost_price = EXCLUDED.cost_price,
selling_price = EXCLUDED.selling_price,
updated_at = CURRENT_TIMESTAMP
""",
(store_id, brand, image_id, mrp, cost_price, selling_price)
)
imported_count += 1
except Exception as e:
logger.error("Stores upload failed: %s", e)
raise HTTPException(status_code=500, detail=f"Database import failed: {e}")
finally:
conn.close()
return {
"status": "success",
"filename": file.filename,
"rows_total": len(df),
"rows_imported": imported_count,
"stores_affected": list(stores_created),
"message": f"Successfully imported {imported_count} store inventory items across {len(stores_created)} store(s)."
}
# ---------------------------------------------------------------------------
# Sales / Analytics Excel / CSV Upload
# ---------------------------------------------------------------------------
@router.post("/analytics")
@router.post("/analytics/upload")
async def upload_analytics_file(file: UploadFile = File(...)) -> Dict[str, Any]:
if not file.filename:
raise HTTPException(status_code=400, detail="No file uploaded")
contents = await file.read()
try:
df = read_df_from_upload(file.filename, contents)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Could not parse Excel/CSV file: {e}")
if df.empty:
raise HTTPException(status_code=400, detail="Uploaded file contains no data rows")
conn = _connect()
if not conn:
raise HTTPException(status_code=500, detail="Database connection failed")
imported_orders = 0
total_revenue = 0.0
try:
with conn, conn.cursor() as cur:
store_db.ensure_store_intelligence_schema()
for _, r in df.iterrows():
row = r.to_dict()
store_id = _get_str(row, ['store_id', 'store'], 'store_mumbai_1')
brand = _get_str(row, ['brand', 'brand_name'], 'amul').lower()
image_id = _get_str(row, ['image_id', 'sku', 'product_sku'], '')
product_name = _get_str(row, ['product_name', 'title', 'item'], 'Analytics Item')
if not image_id:
image_id = f"{brand}_{product_name.lower().replace(' ', '_')}"
order_id = _get_str(row, ['order_id', 'transaction_id'], f"ord_up_{uuid.uuid4().hex[:8]}")
customer_id = _get_str(row, ['customer_id', 'user_id', 'customer'], 'cust_imported')
raw_date = _get_str(row, ['order_date', 'date', 'timestamp'], '')
order_date = datetime.now()
if raw_date:
try:
order_date = pd.to_datetime(raw_date).to_pydatetime()
except Exception:
pass
qty = _get_int(row, ['quantity', 'units_sold', 'qty', 'count'], 1)
unit_price = _get_float(row, ['unit_price', 'selling_price', 'price'], 100.0)
tot_price = _get_float(row, ['total_price', 'revenue', 'total'], qty * unit_price)
# Ensure store exists
cur.execute(
"INSERT INTO stores (store_id, store_name, city, tier, footfall_index) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (store_id) DO NOTHING",
(store_id, store_id.replace('_', ' ').title(), 'Mumbai', 'standard', 25.0)
)
# Insert order header
cur.execute(
"""
INSERT INTO orders (order_id, customer_id, store_id, order_date, payment_method, order_value, delivery_status)
VALUES (%s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (order_id) DO UPDATE SET order_value = EXCLUDED.order_value
""",
(order_id, customer_id, store_id, order_date, 'upi', tot_price, 'delivered')
)
# Insert order item
cur.execute(
"""
INSERT INTO order_items (order_id, brand, image_id, quantity, unit_price, total_price)
VALUES (%s, %s, %s, %s, %s, %s)
""",
(order_id, brand, image_id, qty, unit_price, tot_price)
)
imported_orders += 1
total_revenue += tot_price
except Exception as e:
logger.error("Analytics upload failed: %s", e)
raise HTTPException(status_code=500, detail=f"Database import failed: {e}")
finally:
conn.close()
return {
"status": "success",
"filename": file.filename,
"rows_total": len(df),
"rows_imported": imported_orders,
"total_revenue": round(total_revenue, 2),
"message": f"Successfully imported {imported_orders} sales transactions (Total Revenue: ₹{total_revenue:,.2f})."
}
# ---------------------------------------------------------------------------
# Nutrition Intelligence Excel / CSV Upload
# ---------------------------------------------------------------------------
@router.post("/nutrition")
@router.post("/nutrition/upload")
async def upload_nutrition_file(file: UploadFile = File(...)) -> Dict[str, Any]:
if not file.filename:
raise HTTPException(status_code=400, detail="No file uploaded")
contents = await file.read()
try:
df = read_df_from_upload(file.filename, contents)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Could not parse Excel/CSV file: {e}")
if df.empty:
raise HTTPException(status_code=400, detail="Uploaded file contains no data rows")
conn = _connect()
if not conn:
raise HTTPException(status_code=500, detail="Database connection failed")
imported_count = 0
try:
with conn, conn.cursor() as cur:
nutrition_db.ensure_nutrition_schema()
for _, r in df.iterrows():
row = r.to_dict()
brand = _get_str(row, ['brand', 'brand_name'], 'amul').lower()
product_name = _get_str(row, ['product_name', 'title', 'item', 'name'], 'Nutrition Item')
image_id = _get_str(row, ['image_id', 'sku', 'id'], '')
if not image_id:
image_id = f"{brand}_{product_name.lower().replace(' ', '_')}"
category = _get_str(row, ['category', 'cat'], 'Food')
calories = _get_float(row, ['calories', 'calories_kcal', 'energy'], 150.0)
protein = _get_float(row, ['protein', 'protein_g'], 5.0)
carbs = _get_float(row, ['carbohydrates', 'carbs', 'carbohydrates_g'], 20.0)
sugar = _get_float(row, ['sugar', 'total_sugar_g', 'sugars'], 4.0)
fiber = _get_float(row, ['fiber', 'dietary_fiber_g'], 2.0)
fat = _get_float(row, ['fat', 'total_fat_g'], 6.0)
sodium = _get_float(row, ['sodium', 'sodium_mg'], 120.0)
calcium = _get_float(row, ['calcium', 'calcium_mg'], 80.0)
iron = _get_float(row, ['iron', 'iron_mg'], 1.5)
vitamin_c = _get_float(row, ['vitamin_c', 'vitamin_c_mg'], 5.0)
health_score = _get_float(row, ['health_score', 'nutrition_score', 'score'], 78.0)
diet_tags_raw = _get_str(row, ['diet_tags', 'tags', 'diet'], 'High Protein, Gluten Free')
allergens_raw = _get_str(row, ['allergens', 'allergen'], 'None')
diet_tags = [t.strip() for t in diet_tags_raw.split(',') if t.strip()]
allergens = [a.strip() for a in allergens_raw.split(',') if a.strip()]
# 1. Upsert nutrition_facts
cur.execute(
"""
INSERT INTO nutrition_facts
(brand, image_id, product_name, category, data_status, data_source,
calories_kcal, protein_g, carbohydrates_g, total_sugar_g, dietary_fiber_g,
total_fat_g, sodium_mg, calcium_mg, iron_mg, vitamin_c_mg)
VALUES (%s, %s, %s, %s, 'verified', 'excel_upload', %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (brand, image_id) DO UPDATE SET
product_name = EXCLUDED.product_name,
category = EXCLUDED.category,
data_status = 'verified',
calories_kcal = EXCLUDED.calories_kcal,
protein_g = EXCLUDED.protein_g,
carbohydrates_g = EXCLUDED.carbohydrates_g,
total_sugar_g = EXCLUDED.total_sugar_g,
dietary_fiber_g = EXCLUDED.dietary_fiber_g,
total_fat_g = EXCLUDED.total_fat_g,
sodium_mg = EXCLUDED.sodium_mg,
calcium_mg = EXCLUDED.calcium_mg,
iron_mg = EXCLUDED.iron_mg,
vitamin_c_mg = EXCLUDED.vitamin_c_mg
""",
(brand, image_id, product_name, category, calories, protein, carbs, sugar, fiber, fat, sodium, calcium, iron, vitamin_c)
)
# 2. Upsert nutrition_insights
insights_json = json.dumps({
"brand": brand,
"image_id": image_id,
"data_status": "verified",
"nutrition_score": health_score,
"health_score": health_score,
"positive_insights": [f"Contains {protein}g protein per 100g", f"Provides {fiber}g dietary fiber"],
"nutritional_cautions": [f"{sugar}g sugar per 100g"],
"diet_tags": diet_tags,
"allergens": allergens
})
cur.execute(
"""
INSERT INTO nutrition_insights
(brand, image_id, data_status, nutrition_score, health_score, score_breakdown,
positive_insights, nutritional_cautions, diet_tags, allergens)
VALUES (%s, %s, 'verified', %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (brand, image_id) DO UPDATE SET
data_status = 'verified',
nutrition_score = EXCLUDED.nutrition_score,
health_score = EXCLUDED.health_score,
positive_insights = EXCLUDED.positive_insights,
nutritional_cautions = EXCLUDED.nutritional_cautions,
diet_tags = EXCLUDED.diet_tags,
allergens = EXCLUDED.allergens
""",
(brand, image_id, health_score, health_score, json.dumps({"protein": 85, "fiber": 80}),
[f"Contains {protein}g protein per 100g"], [f"{sugar}g sugar per 100g"], diet_tags, allergens)
)
imported_count += 1
except Exception as e:
logger.error("Nutrition upload failed: %s", e)
raise HTTPException(status_code=500, detail=f"Database import failed: {e}")
finally:
conn.close()
return {
"status": "success",
"filename": file.filename,
"rows_total": len(df),
"rows_imported": imported_count,
"message": f"Successfully imported {imported_count} nutritional intelligence items."
}
# ---------------------------------------------------------------------------
# Template Downloads
# ---------------------------------------------------------------------------
@router.get("/template/{tab_type}")
def get_sample_template(tab_type: str) -> Response:
tab_type = tab_type.lower()
if tab_type == 'stores':
content = (
"store_id,brand,image_id,product_name,category,available_stock,reserved_stock,mrp,cost_price,selling_price,reorder_level,safety_stock\n"
"store_mumbai_1,amul,amul_amul_butter_500ml,Amul Butter 500ml,Dairy,120,5,250.00,200.00,235.00,20,10\n"
"store_mumbai_1,amul,amul_amul_ghee_1l,Amul Ghee 1L,Dairy,85,2,650.00,520.00,610.00,15,5\n"
"store_delhi_2,nestle,nestle_everyday_1kg,Everyday Milk Powder 1kg,Dairy,45,0,420.00,340.00,399.00,10,5\n"
)
filename = "sample_stores_inventory_template.csv"
elif tab_type == 'analytics':
content = (
"order_id,store_id,brand,image_id,product_name,order_date,customer_id,quantity,unit_price,total_price\n"
"ORD_9001,store_mumbai_1,amul,amul_amul_butter_500ml,Amul Butter 500ml,2026-08-01 10:30:00,cust_101,2,235.00,470.00\n"
"ORD_9002,store_mumbai_1,amul,amul_amul_ghee_1l,Amul Ghee 1L,2026-08-01 11:15:00,cust_102,1,610.00,610.00\n"
"ORD_9003,store_delhi_2,nestle,nestle_everyday_1kg,Everyday Milk Powder 1kg,2026-08-02 14:20:00,cust_103,3,399.00,1197.00\n"
)
filename = "sample_analytics_sales_template.csv"
elif tab_type == 'nutrition':
content = (
"brand,image_id,product_name,category,calories_kcal,protein_g,carbohydrates_g,total_sugar_g,dietary_fiber_g,total_fat_g,sodium_mg,health_score,diet_tags,allergens\n"
"amul,amul_amul_butter_500ml,Amul Butter 500ml,Dairy,717,0.8,0.1,0.0,0.0,81.0,650,75,Vegetarian,Dairy\n"
"amul,amul_amul_ghee_1l,Amul Ghee 1L,Dairy,898,0.0,0.0,0.0,0.0,99.8,0,82,Vegetarian,Keto Friendly\n"
"nestle,nestle_everyday_1kg,Everyday Milk Powder 1kg,Dairy,496,25.5,38.0,38.0,0.0,27.0,350,88,High Protein,Dairy\n"
)
filename = "sample_nutrition_intelligence_template.csv"
else:
raise HTTPException(status_code=400, detail=f"Unknown template type '{tab_type}'. Use stores, analytics, or nutrition.")
return PlainTextResponse(
content=content,
media_type="text/csv",
headers={"Content-Disposition": f"attachment; filename={filename}"}
)

View File

@@ -13,6 +13,7 @@ from pydantic import BaseModel, Field
class ProductOut(BaseModel):
image_id: str
image_url: Optional[str] = None
image_urls: List[str] = Field(default_factory=list)
brand: str
product_name: str
title: Optional[str] = None
@@ -31,6 +32,7 @@ class ProductOut(BaseModel):
class SourceProductOut(BaseModel):
image_id: str
image_url: Optional[str] = None
image_urls: List[str] = Field(default_factory=list)
brand: str
product_name: str
title: Optional[str] = None

View File

@@ -25,11 +25,11 @@ from app.services.price_estimator import classify_category, estimate_price, pars
# price realization); standard/budget stores are higher-volume.
# ---------------------------------------------------------------------------
DEFAULT_STORES: List[StoreProfile] = [
StoreProfile("STORE-A", "Store-A - Anna Nagar", "Chennai", "premium", footfall_index=18),
StoreProfile("STORE-A", "Store-A - Gandhipuram", "Coimbatore", "premium", footfall_index=18),
StoreProfile("STORE-B", "Store-B - RS Puram", "Coimbatore", "standard", footfall_index=26),
StoreProfile("STORE-C", "Store-C - Gandhipuram", "Coimbatore", "budget", footfall_index=34),
StoreProfile("STORE-D", "Store-D - Vadapalani", "Chennai", "standard", footfall_index=24),
StoreProfile("STORE-E", "Store-E - Peelamedu", "Coimbatore", "budget", footfall_index=30),
StoreProfile("STORE-C", "Store-C - Peelamedu", "Coimbatore", "budget", footfall_index=34),
StoreProfile("STORE-D", "Store-D - Podanur", "Coimbatore", "standard", footfall_index=24),
StoreProfile("STORE-E", "Store-E - Ukkadam", "Coimbatore", "budget", footfall_index=30),
]
# Fraction of the full catalog each store stocks (Feature 1: "every

View File

@@ -9,14 +9,18 @@ Run with:
from __future__ import annotations
import logging
import threading
from pathlib import Path
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
from app.infrastructure.settings import API_CORS_ORIGINS
from app.api.routers import health, brands, search, chat, catalog
from app.api.routers import health, brands, search, chat, catalog, system
from app.api.routers import stores, discounts, analytics as store_analytics, trending, recommendations, store_admin
from app.api.routers import nutrition, nutrition_admin
from app.api.routers import nutrition, nutrition_admin, upload
from app.services.store_db import ensure_store_intelligence_schema
from app.services.nutrition_db import ensure_nutrition_schema
@@ -29,31 +33,26 @@ logger = logging.getLogger(__name__)
app = FastAPI(
title="Brand Product Search Engine - RAG API",
description=(
"Local, CPU-only RAG API over an Indian FMCG product catalog stored in "
"pgvector. Retrieval uses sentence-transformers/all-MiniLM-L6-v2 "
"embeddings + pgvector cosine similarity; generation uses a local "
"Ollama model (qwen2.5:1.5b by default). Also includes the v3.0 "
"Multi-Store Intelligence layer: store-wise pricing/inventory, "
"ML-based discounts, trending detection, recommendations, and "
"analytics (see /api/admin/store-intelligence for setup). Also "
"includes an AI Nutritional Intelligence module: verified "
"nutrition facts retrieved from Open Food Facts (never LLM-"
"generated), transparent health/nutrition scoring, allergen and "
"diet-compatibility classification, ML-based nutritional "
"similarity/clustering, and nutrition analytics (see "
"/api/admin/nutrition-intelligence for setup)."
"Local, CPU-only RAG API over an Indian FMCG product catalog stored in pgvector. "
"Includes automated multi-store intelligence, ML discount engines, and nutrition intelligence."
),
version="3.1.0",
version="3.2.0",
)
@app.on_event("startup")
def _on_startup() -> None:
# Idempotent - safe to call every boot, mirrors how brand tables are
# lazily ensured elsewhere in this codebase. Never touches or drops
# any existing table.
ensure_store_intelligence_schema()
ensure_nutrition_schema()
# Asynchronously ensure schemas and background auto-init so server starts instantly (<1s)
def _async_init():
try:
ensure_store_intelligence_schema()
ensure_nutrition_schema()
from app.api.routers.system import _run_background_auto_seed
_run_background_auto_seed()
except Exception as e:
logger.warning("Startup background init warning: %s", e)
threading.Thread(target=_async_init, daemon=True).start()
app.add_middleware(
@@ -65,6 +64,7 @@ app.add_middleware(
)
app.include_router(health.router, prefix="/api")
app.include_router(system.router, prefix="/api")
app.include_router(brands.router, prefix="/api")
app.include_router(search.router, prefix="/api")
app.include_router(chat.router, prefix="/api")
@@ -77,12 +77,27 @@ app.include_router(recommendations.router, prefix="/api")
app.include_router(store_admin.router, prefix="/api")
app.include_router(nutrition.router, prefix="/api")
app.include_router(nutrition_admin.router, prefix="/api")
app.include_router(upload.router, prefix="/api")
# Serve built frontend static files if dist exists (single-port unified deployment)
FRONTEND_DIST = Path(__file__).resolve().parents[2] / "frontend" / "dist"
if FRONTEND_DIST.exists() and (FRONTEND_DIST / "assets").exists():
app.mount("/assets", StaticFiles(directory=str(FRONTEND_DIST / "assets")), name="assets")
@app.get("/")
def root() -> dict:
return {
"service": "Brand Product Search Engine - RAG API",
"docs": "/docs",
"health": "/api/health",
}
@app.get("/{full_path:path}")
def serve_frontend(full_path: str):
if full_path.startswith("api") or full_path.startswith("docs") or full_path.startswith("openapi.json"):
return None
file_path = FRONTEND_DIST / full_path
if file_path.exists() and file_path.is_file():
return FileResponse(file_path)
return FileResponse(FRONTEND_DIST / "index.html")
else:
@app.get("/")
def root() -> dict:
return {
"service": "Brand Product Search Engine - RAG API",
"docs": "/docs",
"health": "/api/health",
"system_status": "/api/system/status",
}

View File

@@ -53,27 +53,27 @@ from typing import Dict, List, Optional, Tuple
# this category when sanitizing cross-category language
# out of a generated description.
CATEGORY_REGISTRY: List[Dict[str, object]] = [
{"category": "Biscuits & Cookies", "keywords": ["biscuits", "biscuit", "cookies", "cookie"], "generic_term": "biscuit"},
{"category": "Biscuits & Cookies", "keywords": ["biscuits", "biscuit", "biscit", "biskut", "cookies", "cookie"], "generic_term": "biscuit"},
{"category": "Rusk", "keywords": ["rusks", "rusk"], "generic_term": "rusk"},
{"category": "Crackers", "keywords": ["crackers", "cracker", "saltine"], "generic_term": "cracker"},
{"category": "Cakes & Muffins", "keywords": ["cakes", "cake", "muffins", "muffin"], "generic_term": "bakery item"},
{"category": "Bakery & Breads", "keywords": ["bread", "buns", "bun", "pav"], "generic_term": "bakery product"},
{"category": "Noodles & Instant Food", "keywords": ["noodles", "noodle", "instant food", "vermicelli", "pasta"], "generic_term": "instant food product"},
{"category": "Noodles & Instant Food", "keywords": ["noodles", "noodle", "instant food", "vermicelli", "pasta", "maggi"], "generic_term": "instant food product"},
{"category": "Candy & Confectionery", "keywords": ["candy", "candies", "toffee", "toffees", "lollipop", "lollipops", "confectionery", "mints", "chewing gum"], "generic_term": "candy"},
{"category": "Snacks", "keywords": ["snacks", "snack", "chips", "namkeen", "wafers", "wafer"], "generic_term": "snack"},
{"category": "Chocolates", "keywords": ["chocolates", "chocolate", "cocoa"], "generic_term": "chocolate"},
{"category": "Cooking Oils", "keywords": ["cooking oil", "edible oil", "sunflower oil", "mustard oil", "vanaspati", "refined oil"], "generic_term": "cooking oil"},
{"category": "Atta & Staples", "keywords": ["atta", "wheat flour", "flour", "rice", "dal", "pulses", "staples"], "generic_term": "staple product"},
{"category": "Dairy", "keywords": ["milk", "dairy", "cheese", "paneer", "curd", "yogurt", "butter"], "generic_term": "dairy product"},
{"category": "Oral Care", "keywords": ["toothpaste", "toothbrush", "mouthwash"], "generic_term": "oral care product"},
{"category": "Hair Care", "keywords": ["shampoo", "conditioner", "hair oil"], "generic_term": "hair care product"},
{"category": "Bath Soap", "keywords": ["bath soap", "soap bar", "soap"], "generic_term": "soap"},
{"category": "Skin & Bath Care", "keywords": ["face wash", "body lotion", "skin cream", "moisturizer", "body wash"], "generic_term": "skin care product"},
{"category": "Snacks", "keywords": ["snacks", "snack", "chips", "namkeen", "wafers", "wafer", "kurkure", "lays"], "generic_term": "snack"},
{"category": "Chocolates", "keywords": ["chocolates", "chocolate", "chocate", "choclate", "cocoa", "cadbury chocolate", "dairy milk"], "generic_term": "chocolate"},
{"category": "Cooking Oils", "keywords": ["cooking oil", "edible oil", "sunflower oil", "mustard oil", "vanaspati", "refined oil", "oil", "oils"], "generic_term": "cooking oil"},
{"category": "Atta & Staples", "keywords": ["atta", "wheat flour", "flour", "rice", "dal", "pulses", "staples", "suji", "maida"], "generic_term": "staple product"},
{"category": "Dairy", "keywords": ["milk", "dairy", "cheese", "paneer", "panner", "paner", "paneerr", "curd", "yogurt", "butter", "ghee", "dahi"], "generic_term": "dairy product"},
{"category": "Oral Care", "keywords": ["toothpaste", "toothbrush", "mouthwash", "paste"], "generic_term": "oral care product"},
{"category": "Hair Care", "keywords": ["shampoo", "shampooo", "conditioner", "hair oil"], "generic_term": "hair care product"},
{"category": "Bath Soap", "keywords": ["bath soap", "soap bar", "soap", "soaps"], "generic_term": "soap"},
{"category": "Skin & Bath Care", "keywords": ["face wash", "body lotion", "skin cream", "moisturizer", "body wash", "cream", "lotion"], "generic_term": "skin care product"},
{"category": "Household Cleaning", "keywords": ["detergent", "laundry", "dishwash", "floor cleaner", "handwash", "cleaner"], "generic_term": "cleaning product"},
{"category": "Fragrance & Deodorants", "keywords": ["deodorant", "deo spray", "perfume", "fragrance", "body spray"], "generic_term": "fragrance product"},
{"category": "Fragrance & Deodorants", "keywords": ["deodorant", "deo spray", "perfume", "fragrance", "body spray", "deo"], "generic_term": "fragrance product"},
{"category": "Household - Agarbatti", "keywords": ["agarbatti", "incense sticks", "incense stick"], "generic_term": "agarbatti"},
{"category": "Household - Lamp Oil", "keywords": ["lamp oil"], "generic_term": "lamp oil"},
{"category": "Health Care - Antiseptic", "keywords": ["antiseptic", "disinfectant liquid"], "generic_term": "antiseptic product"},
{"category": "Health Care - Antiseptic", "keywords": ["antiseptic", "disinfectant liquid", "sanitizer"], "generic_term": "antiseptic product"},
]
# All known canonical category names, in priority order.
@@ -90,6 +90,14 @@ def _normalize(text: Optional[str]) -> str:
return re.sub(r"\s+", " ", text).strip()
QUERY_STOP_WORDS = {
"price", "prices", "cost", "costs", "which", "where", "what", "show",
"find", "have", "product", "products", "item", "items", "brand", "brands",
"under", "below", "less", "more", "many", "total", "count", "please", "today",
"recommend", "suggest", "options", "option", "rate", "rates",
}
def _find_matches(text: str) -> List[Tuple[str, str, int]]:
"""Return (category, matched_keyword, keyword_length) for every keyword
found as a whole word/phrase in `text` (case-insensitive)."""
@@ -103,6 +111,22 @@ def _find_matches(text: str) -> List[Tuple[str, str, int]]:
pattern = r"\b" + re.escape(kw) + r"\b"
if re.search(pattern, lower):
matches.append((category, kw, len(kw)))
# Fuzzy matching fallback if exact word search found nothing
if not matches:
import difflib
words = re.findall(r"\b[a-z]{4,}\b", lower)
for entry in CATEGORY_REGISTRY:
category = entry["category"]
for kw in entry["keywords"]:
for word in words:
if word in QUERY_STOP_WORDS:
continue
# Check close similarity for words >= 4 chars
ratio = difflib.SequenceMatcher(None, word, kw).ratio()
if ratio >= 0.8:
matches.append((category, kw, len(kw)))
break
return matches

View File

@@ -113,46 +113,136 @@ def _convert(value: Optional[float], kind: str) -> Optional[float]:
return round(v, 3)
import re
def _clean_title_for_search(title: str) -> str:
"""Strips size, volume, weight, and packaging suffixes to improve search accuracy."""
t = re.sub(r'\b\d+(\.\d+)?\s*(g|kg|ml|l|gm|ltr|grm|pack|pc|pcs)\b', '', title, flags=re.IGNORECASE)
t = re.sub(r'[-_]', ' ', t)
return ' '.join(t.split())
def _looks_non_food(title: str, category: str) -> bool:
text = f"{title} {category}".lower()
return any(kw in text for kw in NON_FOOD_KEYWORDS)
def _search_openfoodfacts(query: str, max_results: int = 5) -> List[dict]:
if not query.strip():
def _search_openfoodfacts(query: str, brand: str = "", category: str = "", max_results: int = 5) -> List[dict]:
if not query.strip() and not brand.strip():
return []
fields_str = (
"code,product_name,brands,nutriments,serving_size,serving_quantity,"
"allergens_tags,labels_tags,ingredients_analysis_tags,categories_tags,"
"ingredients_text,nutriscore_grade,nutrition_data_per,quantity"
)
headers = {"User-Agent": _BROWSER_UA}
# Strategy 1: OFF v2 API search with brands_tags + search_terms
if brand.strip():
cleaned_query = _clean_title_for_search(query)
try:
resp = requests.get(
f"https://{OFF_HOST}/api/v2/search",
params={
"brands_tags": brand.lower().strip(),
"search_terms": cleaned_query,
"page_size": max_results,
"fields": fields_str,
},
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
if resp.status_code == 200:
prods = resp.json().get("products", []) or []
if prods:
return prods
except Exception as e: # noqa: BLE001
logger.debug(f"OFF v2 search Strategy 1 failed for '{brand}' '{query}': {e}")
# Strategy 2: OFF v2 API search using search_terms with combined brand + query
full_term = f"{brand} {_clean_title_for_search(query)}".strip()
try:
resp = requests.get(
f"https://{OFF_HOST}/cgi/search.pl",
f"https://{OFF_HOST}/api/v2/search",
params={
"search_terms": query,
"search_simple": 1,
"action": "process",
"json": 1,
"search_terms": full_term,
"page_size": max_results,
"fields": (
"code,product_name,brands,nutriments,serving_size,serving_quantity,"
"allergens_tags,labels_tags,ingredients_analysis_tags,categories_tags,"
"ingredients_text,nutriscore_grade,nutrition_data_per,quantity"
),
"fields": fields_str,
},
headers={"User-Agent": _BROWSER_UA},
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
if resp.status_code != 200:
return []
return resp.json().get("products", []) or []
if resp.status_code == 200:
prods = resp.json().get("products", []) or []
if prods:
return prods
except Exception as e: # noqa: BLE001
logger.debug(f"Open Food Facts lookup failed for '{query}': {e}")
return []
logger.debug(f"OFF v2 search Strategy 2 failed for '{full_term}': {e}")
# Strategy 3: OFF v2 API search using q parameter
try:
resp = requests.get(
f"https://{OFF_HOST}/api/v2/search",
params={
"q": full_term,
"page_size": max_results,
"fields": fields_str,
},
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
if resp.status_code == 200:
prods = resp.json().get("products", []) or []
if prods:
return prods
except Exception as e: # noqa: BLE001
logger.debug(f"OFF v2 search Strategy 3 failed for '{full_term}': {e}")
# Strategy 4: Fallback to category + brand search
if category.strip() and brand.strip():
try:
resp = requests.get(
f"https://{OFF_HOST}/api/v2/search",
params={
"search_terms": f"{brand} {category}".strip(),
"page_size": max_results,
"fields": fields_str,
},
headers=headers,
timeout=REQUEST_TIMEOUT_SECONDS,
)
if resp.status_code == 200:
return resp.json().get("products", []) or []
except Exception as e: # noqa: BLE001
logger.debug(f"OFF v2 search Strategy 4 failed: {e}")
return []
def _match_confidence(query_brand: str, query_title: str, candidate: dict) -> float:
cand_name = f"{candidate.get('brands', '')} {candidate.get('product_name', '')}".lower().strip()
query = f"{query_brand} {query_title}".lower().strip()
if not cand_name or not query:
cand_brand = (candidate.get("brands") or "").lower().strip()
cand_name = (candidate.get("product_name") or "").lower().strip()
q_brand = query_brand.lower().strip()
q_title_clean = _clean_title_for_search(query_title).lower().strip()
full_cand = f"{cand_brand} {cand_name}".strip()
full_query = f"{q_brand} {q_title_clean}".strip()
if not full_cand or not full_query:
return 0.0
return SequenceMatcher(None, query, cand_name).ratio()
ratio = SequenceMatcher(None, full_query, full_cand).ratio()
if q_brand and (q_brand in cand_brand or q_brand in cand_name):
ratio = max(ratio, 0.45)
q_tokens = set(q_title_clean.split())
cand_tokens = set(re.sub(r'[-_]', ' ', cand_name).split())
if q_tokens and cand_tokens:
overlap = len(q_tokens.intersection(cand_tokens)) / float(len(q_tokens))
if overlap >= 0.5:
ratio = max(ratio, 0.5 + overlap * 0.3)
return min(1.0, ratio)
def _extract_allergens(candidate: dict) -> List[str]:
@@ -219,9 +309,9 @@ def fetch_verified_nutrition(brand: str, title: str, category: str = "") -> Dict
if _looks_non_food(title, category):
return unavailable
candidates = _search_openfoodfacts(f"{brand} {title}".strip())
candidates = _search_openfoodfacts(query=title, brand=brand, category=category)
if not candidates:
candidates = _search_openfoodfacts(title)
candidates = _search_openfoodfacts(query=title)
if not candidates:
return unavailable

View File

@@ -135,33 +135,39 @@ RAG_SYSTEM_PROMPT = (
"low fat, spicy), only claim a product has that attribute if the CONTEXT text for that product "
"actually says so. If none of the retrieved products explicitly confirm the attribute, say you "
"couldn't confirm it from the catalog data rather than assuming or guessing.\n"
"8. COUNT AND METRICS: If CATALOG METRIC DATA is present in the CONTEXT, state those exact count numbers clearly when answering count or list questions.\n"
"Do not mention the word 'CONTEXT' or that you were given a context block; just answer naturally."
)
def _build_grounded_fallback_answer(query: str, context_block: str, requested_category: Optional[str] = None) -> str:
"""Intelligent non-LLM synthesis of retrieved catalog items when Ollama is offline."""
if not context_block or "(No matching products were found" in context_block:
cat_str = f" in category **{requested_category}**" if requested_category else ""
return f"No matching products were found in the catalog database for your query{cat_str}."
cat_str = f" ({requested_category})" if requested_category else ""
lines = [f"Here are the top catalog recommendations for **'{query}'**{cat_str}:\n"]
raw_items = [b.strip() for b in context_block.split("\n\n") if b.strip()]
for item in raw_items:
if item.startswith("CATALOG METRIC DATA:"):
lines.append(f"📊 **Catalog Info**: {item.replace('CATALOG METRIC DATA:', '').strip()}\n")
else:
lines.append(item)
return "\n\n".join(lines)
def generate_rag_answer(
query: str,
context_block: str,
history: Optional[List[Dict[str, str]]] = None,
requested_category: Optional[str] = None,
) -> str:
"""Generate a grounded answer for the RAG chat endpoint.
`context_block` is a pre-formatted block of retrieved products (built by
`app.services.rag_service.build_context`). `history` is an optional list
of prior `{"role": "user"|"assistant", "content": ...}` turns so the
model can handle short follow-up questions ("what about a cheaper one?").
`requested_category` is the product category auto-detected from the
query (see `category_registry.detect_category_from_text`), passed
through so the model has an explicit reminder of what was searched for
even if the CONTEXT ends up empty.
"""
"""Generate a grounded answer for the RAG chat endpoint."""
if not _ensure_client():
return (
"I can't reach the local Ollama server right now, so I can't generate an "
"answer. Please confirm Ollama is running (`ollama serve`) and that the "
f"'{OLLAMA_MODEL_NAME}' model is pulled (`ollama pull {OLLAMA_MODEL_NAME}`)."
)
return _build_grounded_fallback_answer(query, context_block, requested_category)
history_block = ""
if history:
@@ -188,11 +194,7 @@ def generate_rag_answer(
answer = _generate(RAG_SYSTEM_PROMPT, user_prompt, max_retries=2)
if not answer:
return (
"I wasn't able to generate a response from the local model just now. "
"This can happen if the model is still loading or the machine is low on "
"free RAM - please try again in a few seconds."
)
return _build_grounded_fallback_answer(query, context_block, requested_category)
return answer.strip()

View File

@@ -94,6 +94,109 @@ def product_matches_attribute(product_text: str, attribute: str, value: str) ->
return any(hint in lower for hint in hints)
def is_count_query(query: str) -> bool:
"""Return True if the query asks for a total/count of products or brands."""
if not query:
return False
lower = query.lower()
patterns = [
r"how many",
r"count of",
r"total (?:number of )?(?:products|items|brands)",
r"number of products",
r"how many products",
r"how many items",
r"how many brands",
r"total count",
]
return any(re.search(p, lower) for p in patterns)
def extract_max_price(query: str) -> Optional[float]:
"""Extract a numeric maximum price ceiling from a query if present.
e.g. 'under ₹150' -> 150.0, 'less than 100 rupees' -> 100.0
"""
if not query:
return None
lower = query.lower().replace(",", "")
patterns = [
r"(?:under|below|less than|within|upto|up to|budget of|max(?:imum)?)\s*(?:₹|rs\.?|inr)?\s*(\d+(?:\.\d+)?)",
r"(?:₹|rs\.?|inr)\s*(\d+(?:\.\d+)?)\s*(?:or less|max|under|below)",
r"<\s*(?:₹|rs\.?|inr)?\s*(\d+(?:\.\d+)?)",
]
for p in patterns:
match = re.search(p, lower)
if match:
try:
val = float(match.group(1))
if val > 0:
return val
except ValueError:
pass
return None
KNOWN_BRANDS = [
"Amul", "Cadbury", "Cavinkare", "Coca-Cola", "Colgate-Palmolive", "Dabur",
"Godrej", "Grb", "Hindustan Unilever", "Lion Dates", "Manna", "Milky Mist",
"Naga", "Nestle", "P&G", "Pepsico"
]
BRAND_SEARCH_MAP = {
"cadbury": "Cadbury",
"amul": "Amul",
"cavinkare": "Cavinkare",
"coca-cola": "Coca-Cola",
"coca cola": "Coca-Cola",
"coke": "Coca-Cola",
"colgate-palmolive": "Colgate-Palmolive",
"colgate": "Colgate-Palmolive",
"dabur": "Dabur",
"godrej": "Godrej",
"grb": "Grb",
"hindustan unilever": "Hindustan Unilever",
"hul": "Hindustan Unilever",
"unilever": "Hindustan Unilever",
"lion dates": "Lion Dates",
"lion": "Lion Dates",
"manna": "Manna",
"milky mist": "Milky Mist",
"naga": "Naga",
"nestle": "Nestle",
"p&g": "P&G",
"pg": "P&G",
"pepsico": "Pepsico",
"pepsi": "Pepsico",
}
def extract_brand_mention(query: str) -> Optional[str]:
"""Detect if a brand name is explicitly mentioned in the query text."""
if not query:
return None
from app.services.brand_registry import BRAND_ALIASES, resolve_parent_brand
lower = query.lower()
# 1. Check direct search map
for alias in sorted(BRAND_SEARCH_MAP.keys(), key=len, reverse=True):
pattern = r"\b" + re.escape(alias) + r"\b"
if re.search(pattern, lower):
return BRAND_SEARCH_MAP[alias]
# 2. Check sub-brand aliases (e.g. "oreo", "maggi", "good day")
sorted_aliases = sorted(BRAND_ALIASES.keys(), key=len, reverse=True)
for alias in sorted_aliases:
pattern = r"\b" + re.escape(alias) + r"\b"
if re.search(pattern, lower):
parent = resolve_parent_brand(alias)
# Normalize to canonical known brand name case
for kb in KNOWN_BRANDS:
if kb.lower() == parent.lower():
return kb
return parent.title()
return None
def detected_category(query: str) -> Optional[str]:
"""Thin wrapper kept for readability at call sites in rag_service."""
return detect_category_from_text(query)

View File

@@ -32,9 +32,22 @@ from app.infrastructure.settings import (
from app.services.category_registry import category_matches, detect_category_from_text
from app.services.embeddings_service import embed_texts
from app.services.ollama_service import generate_rag_answer
from app.services.query_intent import extract_attributes, product_matches_attribute
from app.services.query_intent import (
extract_attributes,
extract_brand_mention,
extract_max_price,
is_count_query,
product_matches_attribute,
)
from app.services.s3_service import s3_service
from app.services.vector_store import semantic_search, text_search
from app.services.vector_store import (
count_products_all_brands,
count_products_by_brand,
list_available_brands,
list_categories_for_brand,
semantic_search,
text_search,
)
logger = logging.getLogger(__name__)
@@ -43,6 +56,7 @@ logger = logging.getLogger(__name__)
class RetrievedProduct:
image_id: str
image_url: str = ""
image_urls: List[str] = field(default_factory=list)
brand: str = ""
title: str = ""
product_name: Optional[str] = None
@@ -67,6 +81,7 @@ class RetrievedProduct:
return {
"image_id": self.image_id,
"image_url": self.image_url,
"image_urls": self.image_urls,
"brand": self.brand,
"title": self.title,
"product_name": self.product_name,
@@ -102,12 +117,34 @@ class RagAnswer:
}
def _clean_url(url: Optional[str]) -> Optional[str]:
if not url:
return None
return str(url).replace('{width}', '800')
def _row_to_retrieved_product(row: Dict[str, Any]) -> RetrievedProduct:
image_id = row.get("image_id") or ""
brand = (row.get("brand") or "").title()
s3_single = s3_service.get_product_image_url(brand, image_id)
s3_list = s3_service.get_product_image_urls(brand, image_id)
db_single = _clean_url(row.get("image_url"))
db_list = [_clean_url(u) for u in (row.get("image_urls") or []) if u]
final_urls = s3_list if s3_list else db_list
if not final_urls and db_single:
final_urls = [db_single]
if not final_urls and s3_single:
final_urls = [s3_single]
primary_url = (final_urls[0] if final_urls else "") or db_single or s3_single
return RetrievedProduct(
image_id=image_id,
image_url=s3_service.get_product_image_url(brand, image_id) or "",
image_url=primary_url,
image_urls=final_urls,
brand=brand,
product_name=row.get("product_name") or row.get("title") or "Unknown product",
title=row.get("title") or row.get("product_name") or None,
@@ -126,14 +163,6 @@ def _row_to_retrieved_product(row: Dict[str, Any]) -> RetrievedProduct:
def _filter_to_category(rows: List[Dict[str, Any]], target_category: Optional[str]) -> List[Dict[str, Any]]:
"""Drop any row whose stored category doesn't match `target_category`.
This is the safety net against the "biscuit query returns snacks" bug:
even if a product scores well on raw embedding similarity (e.g.
because its LLM-generated description loosely used the word
"biscuit"), it's excluded here unless its own `category` column
genuinely matches what the user asked for.
"""
if not target_category:
return rows
return [r for r in rows if category_matches(r.get("category"), target_category)]
@@ -141,10 +170,7 @@ def _filter_to_category(rows: List[Dict[str, Any]], target_category: Optional[st
def _rerank_by_attributes(products: List[RetrievedProduct], attrs: Dict[str, str]) -> List[RetrievedProduct]:
"""Move products with explicit evidence of a requested attribute (e.g.
"sugar": "low") to the front, without dropping any product - most
catalog entries won't mention the attribute either way, and we'd
rather show them (ranked below confirmed matches) than show nothing.
Original similarity ordering is preserved as the tie-breaker.
"sugar": "low") to the front, without dropping any product.
"""
def evidence_count(p: RetrievedProduct) -> int:
text = " ".join(filter(None, [p.description or "", " ".join(p.highlights), " ".join(p.nutrients)]))
@@ -160,29 +186,13 @@ def _rerank_by_attributes(products: List[RetrievedProduct], attrs: Dict[str, str
def retrieve(query: str, brand: Optional[str] = None, top_k: Optional[int] = None,
category: Optional[str] = None, max_distance: Optional[float] = None) -> List[RetrievedProduct]:
"""Embed `query` and run a pgvector similarity search. Used by both
the plain semantic-search endpoint and the RAG chat endpoint.
Category handling: an explicit `category` argument (e.g. a UI filter)
always wins. Otherwise the product category the user is asking about
is auto-detected from the query text (`detect_category_from_text`) and
used to scope the search - both by pushing the filter down into the
SQL query (so the right products are actually retrieved, not just
ranked, within the top-k) and by re-checking every returned row
afterwards. If a category is detected but nothing matches it, the
result is an empty list rather than silently widening to other
categories - the caller (and ultimately the LLM, per its system
prompt) is expected to say plainly that nothing matched rather than
substitute a different kind of product.
Falls back to a text-based ILIKE search when the embedding model
fails or the vector search returns no results, so the search box
still returns matching products when pgvector embeddings are
unavailable or stale.
"""
"""Embed `query` and run a pgvector similarity search with category, brand, and price filtering."""
top_k = min(top_k or RAG_DEFAULT_TOP_K, RAG_MAX_TOP_K)
effective_max_distance = max_distance if max_distance is not None else RAG_MAX_DISTANCE
target_brand = brand or extract_brand_mention(query)
target_category = category or detect_category_from_text(query)
max_price = extract_max_price(query)
try:
vectors = embed_texts([query])
@@ -194,21 +204,21 @@ def retrieve(query: str, brand: Optional[str] = None, top_k: Optional[int] = Non
if vectors:
rows = semantic_search(
query_embedding=vectors[0],
brand=brand,
brand=target_brand,
top_k=top_k,
category=target_category,
max_distance=effective_max_distance,
max_price=max_price,
)
rows = _filter_to_category(rows, target_category)
if not rows:
logger.info(
"Semantic search returned no in-category results for %r (category=%r), "
"falling back to text search",
"Semantic search returned no in-category results for %r (category=%r), falling back to text search",
query, target_category,
)
if not rows:
rows = text_search(query, brand=brand, top_k=top_k, category=target_category)
rows = text_search(query, brand=target_brand, top_k=top_k, category=target_category, max_price=max_price)
rows = _filter_to_category(rows, target_category)
products = [_row_to_retrieved_product(r) for r in rows]
@@ -220,16 +230,13 @@ def retrieve(query: str, brand: Optional[str] = None, top_k: Optional[int] = Non
return products
def build_context(products: List[RetrievedProduct], max_chars: int = RAG_MAX_CONTEXT_CHARS) -> str:
"""Format retrieved products into a compact text block for the LLM prompt.
def build_context(products: List[RetrievedProduct], max_chars: int = RAG_MAX_CONTEXT_CHARS,
metric_info: Optional[str] = None) -> str:
"""Format retrieved products and optional metric info into a compact text block for the LLM prompt."""
prefix = f"{metric_info}\n\n" if metric_info else ""
Stays under `max_chars` so a tiny 1.5B model on CPU doesn't choke on an
oversized prompt - we truncate per-product description rather than
dropping whole products, so the model still sees the full breadth of
matches.
"""
if not products:
return "(No matching products were found in the catalog for this query.)"
return prefix + "(No matching products were found in the catalog for this query.)"
per_item_budget = max(200, max_chars // max(1, len(products)))
lines = []
@@ -246,11 +253,13 @@ def build_context(products: List[RetrievedProduct], max_chars: int = RAG_MAX_CON
parts.append(f" Sizes: {', '.join(p.size_variants[:6])}")
if p.highlights:
parts.append(f" Highlights: {', '.join(p.highlights[:5])}")
if p.nutrients:
parts.append(f" Nutrients: {', '.join(p.nutrients[:5])}")
if desc:
parts.append(f" Description: {desc}")
lines.append("\n".join(parts))
return "\n\n".join(lines)
return prefix + "\n\n".join(lines)
def answer_query(
@@ -260,12 +269,33 @@ def answer_query(
category: Optional[str] = None,
history: Optional[List[Dict[str, str]]] = None,
) -> RagAnswer:
"""End-to-end RAG: retrieve relevant products, then generate a grounded answer."""
"""End-to-end RAG: retrieve relevant products, compute exact counts if requested, then generate a grounded answer."""
target_brand = brand or extract_brand_mention(query)
target_category = category or detect_category_from_text(query)
metric_info: Optional[str] = None
if is_count_query(query):
if target_brand:
cnt = count_products_by_brand(target_brand, category=target_category)
cats = list_categories_for_brand(target_brand)
cat_str = f" in category '{target_category}'" if target_category else ""
metric_info = (
f"CATALOG METRIC DATA: Exact product count for brand '{target_brand}'{cat_str} in the database is {cnt}. "
f"Categories available under {target_brand}: {', '.join(cats)}."
)
elif target_category:
cnt = count_products_all_brands(category=target_category)
metric_info = f"CATALOG METRIC DATA: Total products matching category '{target_category}' across all brands in database is {cnt}."
else:
total = count_products_all_brands()
brands = list_available_brands()
metric_info = f"CATALOG METRIC DATA: Total catalog items across all brands is {total}. Known brands: {', '.join(brands)}."
products = retrieve(query, brand=brand, top_k=top_k, category=category)
context = build_context(products)
context = build_context(products, metric_info=metric_info)
answer_text = generate_rag_answer(query, context, history=history, requested_category=target_category)
return RagAnswer(
answer=answer_text, sources=products, query=query, brand=brand,
answer=answer_text, sources=products, query=query, brand=target_brand or brand,
detected_category=target_category,
)

View File

@@ -26,6 +26,7 @@ class S3Service:
def __init__(self):
self.enabled = USE_S3 and all([S3_ACCESS_KEY, S3_SECRET_KEY, S3_ENDPOINT, S3_BUCKET])
self._url_cache = {}
if self.enabled:
# Use regional base endpoint for API calls to avoid NoSuchKey errors with virtual-hosted style URLs
# DO Spaces API endpoint format: https://{region}.digitaloceanspaces.com
@@ -231,15 +232,12 @@ class S3Service:
return f"{S3_ENDPOINT}/{key}"
def get_product_image_url(self, brand: str, image_id: str) -> str:
"""Construct the first image URL for a product using the known S3 naming convention.
Images are uploaded as ``image_000.jpg``, ``image_001.jpg`` etc.
under ``daily/brands/{brand}/{image_id}/``, so we can build the
URL directly without a slow S3 ``list_objects`` call. Returns an
empty string when the image is unavailable or S3 is disabled.
"""
"""Construct or fetch the first image URL for a product in S3."""
if not self.enabled or not image_id or not brand:
return ""
urls = self.get_product_image_urls(brand, image_id)
if urls:
return urls[0]
storage_brand = resolve_parent_brand(brand)
key = f"daily/brands/{storage_brand.lower()}/{image_id}/image_000.jpg"
return self.get_public_url(key)
@@ -249,6 +247,10 @@ class S3Service:
if not self.enabled or not image_id:
return []
cache_key = f"{brand}:{image_id}"
if cache_key in self._url_cache:
return self._url_cache[cache_key]
try:
# Try specific prefix patterns to find existing data
storage_brand = resolve_parent_brand(brand) if brand else brand
@@ -278,9 +280,12 @@ class S3Service:
if image_urls:
logger.info(f"✅ Found {len(image_urls)} images under prefix: {prefix}")
return sorted(image_urls)
res = sorted(image_urls)
self._url_cache[cache_key] = res
return res
logger.debug("No images found in S3 for %s across searched prefixes", image_id)
self._url_cache[cache_key] = []
return []
except Exception as e:

View File

@@ -39,6 +39,8 @@ def get_brand_table_ddl(brand: str) -> str:
description TEXT,
category TEXT,
image_id TEXT UNIQUE NOT NULL,
image_url TEXT,
image_urls TEXT[],
-- Essential pricing fields
price_range TEXT,
@@ -96,6 +98,8 @@ def _ensure_columns(cur, table_name: str) -> None:
"description": "TEXT",
"category": "TEXT",
"image_id": "TEXT",
"image_url": "TEXT",
"image_urls": "TEXT[]",
"price_range": "TEXT",
"size_variants": "TEXT[]",
"providers": "TEXT[]",
@@ -123,7 +127,7 @@ def _ensure_columns(cur, table_name: str) -> None:
logger.info(f"Added missing column '{col}' to {table_name}")
# 2. Relax legacy NOT NULL constraints on columns not present in standard insert
inserted_cols = {"id", "product_name", "title", "description", "category", "image_id", "price_range", "size_variants", "providers", "fssai_license", "product_sku", "sku_source", "highlights", "nutrients", "search_query", "embedding", "created_at", "updated_at"}
inserted_cols = {"id", "product_name", "title", "description", "category", "image_id", "image_url", "image_urls", "price_range", "size_variants", "providers", "fssai_license", "product_sku", "sku_source", "highlights", "nutrients", "search_query", "embedding", "created_at", "updated_at"}
for col, is_nullable, col_def in col_info:
if col not in inserted_cols and is_nullable == 'NO' and col_def is None:
cur.execute(f"ALTER TABLE {table_name} ALTER COLUMN {col} DROP NOT NULL")
@@ -140,9 +144,6 @@ def _ensure_columns(cur, table_name: str) -> None:
logger.warning(f"Unique index creation on {table_name}.image_id: {e}")
def ensure_brand_schema(brand: str) -> str:
"""Ensure brand-specific table exists and return table name"""
conn = _connect()
@@ -191,6 +192,20 @@ def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: b
category = p.get("category") or "Uncategorized"
image_id = p.get("image_id") or ""
image_url = p.get("image_url") or p.get("primary_image") or ""
if isinstance(image_url, str):
image_url = image_url.replace('{width}', '800')
else:
image_url = ""
raw_image_urls = p.get("image_urls") or []
if isinstance(raw_image_urls, list):
image_urls = [str(u).replace('{width}', '800') for u in raw_image_urls if u]
else:
image_urls = []
if not image_urls and image_url:
image_urls = [image_url]
# Essential pricing fields
price_range = p.get("price_range") or ""
size_variants = p.get("size_variants", [])
@@ -244,6 +259,8 @@ def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: b
description,
category,
image_id,
image_url,
image_urls,
price_range,
size_variants, # TEXT[] - psycopg will handle conversion
providers, # TEXT[] - psycopg will handle conversion
@@ -262,14 +279,16 @@ def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: b
cur.executemany(
f"""
INSERT INTO {table_name}
(product_name, title, description, category, image_id, price_range, size_variants, providers,
(product_name, title, description, category, image_id, image_url, image_urls, price_range, size_variants, providers,
fssai_license, product_sku, sku_source, highlights, nutrients, search_query, embedding)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (image_id) DO UPDATE SET
product_name = EXCLUDED.product_name,
title = EXCLUDED.title,
description = EXCLUDED.description,
category = EXCLUDED.category,
image_url = EXCLUDED.image_url,
image_urls = EXCLUDED.image_urls,
price_range = EXCLUDED.price_range,
size_variants = EXCLUDED.size_variants,
providers = EXCLUDED.providers,
@@ -526,12 +545,27 @@ def list_categories_for_brand(brand: str) -> List[str]:
conn.close()
def parse_price_range(price_range_str: Optional[str]) -> Tuple[Optional[float], Optional[float]]:
"""Parse string representations of prices into numeric floats (min_price, max_price).
e.g. '₹120 - ₹150' -> (120.0, 150.0), '₹140' -> (140.0, 140.0)
"""
if not price_range_str:
return None, None
nums = [float(n) for n in re.findall(r"\d+(?:\.\d+)?", str(price_range_str).replace(",", ""))]
if not nums:
return None, None
if len(nums) == 1:
return nums[0], nums[0]
return min(nums), max(nums)
def semantic_search(
query_embedding: List[float],
brand: Optional[str] = None,
top_k: int = 5,
category: Optional[str] = None,
max_distance: Optional[float] = None,
max_price: Optional[float] = None,
) -> List[Dict[str, Any]]:
"""Core RAG retrieval step: cosine-similarity search over pgvector.
@@ -564,6 +598,9 @@ def semantic_search(
else:
tables = [(name, f"brand_{name}") for name in _list_brand_table_suffixes(cur)]
# Fetch extra candidates per table if max_price or filtering is applied
fetch_limit = top_k * 5 if (max_price is not None or category) else top_k
for brand_label, table_name in tables:
if not _table_exists(cur, table_name):
continue
@@ -576,7 +613,7 @@ def semantic_search(
sql += " AND category ILIKE %s"
params.append(f"%{category}%")
sql += " ORDER BY distance ASC LIMIT %s"
params.append(top_k)
params.append(fetch_limit)
try:
cur.execute(sql, params)
@@ -588,8 +625,16 @@ def semantic_search(
for row in cur.fetchall():
record = dict(zip(colnames, row))
record["brand"] = record.get("brand") or brand_label
if max_distance is None or record["distance"] <= max_distance:
results.append(record)
if max_distance is not None and record["distance"] > max_distance:
continue
if max_price is not None:
min_p, max_p = parse_price_range(record.get("price_range"))
if min_p is not None and min_p > max_price:
continue
results.append(record)
finally:
conn.close()
@@ -603,6 +648,7 @@ def text_search(
brand: Optional[str] = None,
top_k: int = 10,
category: Optional[str] = None,
max_price: Optional[float] = None,
) -> List[Dict[str, Any]]:
"""Fallback text-based search using ILIKE on title and description.
@@ -617,6 +663,7 @@ def text_search(
results: List[Dict[str, Any]] = []
like_pattern = f"%{query}%"
fetch_limit = top_k * 5 if (max_price is not None or category) else top_k
try:
with conn, conn.cursor() as cur:
@@ -642,7 +689,7 @@ def text_search(
if where_clauses:
sql += " WHERE " + " AND ".join(where_clauses)
sql += " ORDER BY updated_at DESC LIMIT %s"
params.append(top_k)
params.append(fetch_limit)
try:
cur.execute(sql, params)
@@ -655,6 +702,12 @@ def text_search(
record = dict(zip(colnames, row))
record["brand"] = brand_label
record["distance"] = 1.0
if max_price is not None:
min_p, max_p = parse_price_range(record.get("price_range"))
if min_p is not None and min_p > max_price:
continue
results.append(record)
finally:

View File

@@ -0,0 +1,8 @@
import sys
sys.path.insert(0, ".")
from app.services.store_db import list_stores
stores = list_stores()
print("Stores in DB:")
for s in stores:
print(f" - {s['store_id']}: {s['store_name']} ({s.get('city')}) [{s.get('tier')}]")

View File

@@ -0,0 +1,34 @@
import sys
import os
sys.path.insert(0, ".")
from app.services.rag_service import retrieve, answer_query
from app.services.query_intent import extract_max_price, extract_brand_mention, is_count_query, detected_category
print("=== Q1: How many products are there in Cadbury? ===")
print(" Is Count:", is_count_query("How many products are there in Cadbury?"))
print(" Brand Mention:", extract_brand_mention("How many products are there in Cadbury?"))
print(" Category:", detected_category("How many products are there in Cadbury?"))
print("\n=== Q2: Suggest panner less than ₹100 ===")
print(" Is Count:", is_count_query("Suggest panner less than ₹100"))
print(" Brand Mention:", extract_brand_mention("Suggest panner less than ₹100"))
print(" Category:", detected_category("Suggest panner less than ₹100"))
print(" Max Price:", extract_max_price("Suggest panner less than ₹100"))
q2_prods = retrieve("Suggest panner less than ₹100")
for p in q2_prods:
print(f" - [{p.brand}] {p.product_name} | Price: {p.price_range} | Category: {p.category}")
print("\n=== Q3: Recommend Paneer under ₹150 ===")
print(" Is Count:", is_count_query("Recommend Paneer under ₹150"))
print(" Brand Mention:", extract_brand_mention("Recommend Paneer under ₹150"))
print(" Category:", detected_category("Recommend Paneer under ₹150"))
print(" Max Price:", extract_max_price("Recommend Paneer under ₹150"))
q3_prods = retrieve("Recommend Paneer under ₹150")
for p in q3_prods:
print(f" - [{p.brand}] {p.product_name} | Price: {p.price_range} | Category: {p.category}")
print("\n=== Q4: Suggest low sugar biscuits ===")
print(" Category:", detected_category("Suggest low sugar biscuits"))
q4_prods = retrieve("Suggest low sugar biscuits")
for p in q4_prods:
print(f" - [{p.brand}] {p.product_name} | Price: {p.price_range} | Category: {p.category}")

View File

@@ -43,8 +43,22 @@ def main() -> None:
"--only", nargs="*", default=None,
help="Optional list of brand names (case-insensitive substring match on filename) to limit seeding to",
)
parser.add_argument(
"--skip-if-seeded", action="store_true",
help="Skip seeding if database already contains products",
)
args = parser.parse_args()
if args.skip_if_seeded:
from app.services.vector_store import count_products_all_brands
try:
cnt = count_products_all_brands()
if cnt > 0:
logger.info("⚡ Database already contains %d products. Skipping sample data seed.", cnt)
return
except Exception:
pass
if not SEED_DIR.exists():
logger.error("Seed directory not found: %s", SEED_DIR)
sys.exit(1)

View File

@@ -34,8 +34,19 @@ def main() -> None:
parser.add_argument("--days", type=int, default=90, help="Days of order history to simulate (default: 90)")
parser.add_argument("--seed", type=int, default=42, help="Random seed for reproducible store/order generation")
parser.add_argument("--no-reset-orders", action="store_true", help="Append to existing order history instead of clearing it first")
parser.add_argument("--skip-if-seeded", action="store_true", help="Skip if stores are already provisioned")
args = parser.parse_args()
if args.skip_if_seeded:
from app.services.store_db import list_stores
try:
stores = list_stores()
if stores and len(stores) >= 5:
logger.info("⚡ Stores already provisioned (%d stores). Skipping store intelligence seed.", len(stores))
return
except Exception:
pass
from app.services.store_seed_service import run_seed
logger.info("Seeding store intelligence (days=%d, seed=%d, reset_orders=%s)...", args.days, args.seed, not args.no_reset_orders)

View File

@@ -0,0 +1,30 @@
import sys
from app.services.rag_service import retrieve, answer_query
from app.services.query_intent import extract_max_price, extract_brand_mention, is_count_query
def run_test():
queries = [
"How many products are there in Cadbury?",
"Suggest low sugar biscuits",
"Recommend Paneer under ₹150",
"Suggest panner less than ₹100",
]
for q in queries:
print("="*60)
print(f"QUERY: {q}")
print(f" Is Count: {is_count_query(q)}")
print(f" Brand Mention: {extract_brand_mention(q)}")
print(f" Max Price: {extract_max_price(q)}")
prods = retrieve(q)
print(" Retrieved Products:")
for p in prods[:5]:
print(f" - {p.brand} | {p.product_name} | Price: {p.price_range} | Cat: {p.category}")
ans = answer_query(q)
print(f" ANSWER:\n{ans.answer}")
print("="*60)
if __name__ == "__main__":
run_test()

View File

@@ -0,0 +1,12 @@
import sys
sys.stdout.reconfigure(encoding='utf-8')
sys.path.insert(0, ".")
from app.services.rag_service import answer_query
res = answer_query("Recommend a low sugar biscuit")
print("=== RAG ANSWER ===")
print(res.answer)
print("=== SOURCES FOUND ===")
print(len(res.sources))
for p in res.sources[:3]:
print(f" - {p.product_name} ({p.brand}) - {p.price_range}")

View File

@@ -142,3 +142,37 @@ def test_extracts_spicy() -> None:
def test_no_attributes_returns_empty_dict() -> None:
assert extract_attributes("show me Parle biscuits") == {}
# ---------------------------------------------------------------------------
# New Intent Extractor & Typo Tests
# ---------------------------------------------------------------------------
from app.services.query_intent import is_count_query, extract_max_price, extract_brand_mention
def test_detects_typos_in_category() -> None:
assert detect_category_from_text("Suggest panner less than ₹100") == "Dairy"
assert detect_category_from_text("Show me biscit options") == "Biscuits & Cookies"
assert detect_category_from_text("chocate dark") == "Chocolates"
def test_is_count_query() -> None:
assert is_count_query("How many products are there in Cadbury?") is True
assert is_count_query("What is the total number of items in Amul?") is True
assert is_count_query("how many items under snacks?") is True
assert is_count_query("Recommend Paneer under ₹150") is False
def test_extract_max_price() -> None:
assert extract_max_price("Recommend Paneer under ₹150") == 150.0
assert extract_max_price("Suggest panner less than ₹100") == 100.0
assert extract_max_price("biscuits below 50 rupees") == 50.0
assert extract_max_price("what paneer do you have?") is None
def test_extract_brand_mention() -> None:
assert extract_brand_mention("How many products are there in Cadbury?") == "Cadbury"
assert extract_brand_mention("show me Amul butter") == "Amul"
assert extract_brand_mention("any Nestle chocolates?") == "Nestle"
assert extract_brand_mention("suggest low sugar biscuits") is None