Add background upload enrichment and LLM-driven search intent parsing

.env.example

@@ -0,0 +1,25 @@
+# Internal service URLs for Docker Compose
+PRODUCT_SERVICE_URL=http://product-service:8001
+SEARCH_SERVICE_URL=http://search-service:8002
+
+# Optional LLM enrichment
+LLM_ENRICHMENT_ENABLED=false
+LLM_ENRICHMENT_MODEL=gpt-4o-mini
+LLM_ENRICHMENT_BASE_URL=https://api.openai.com/v1
+LLM_ENRICHMENT_API_KEY=
+LLM_ENRICHMENT_TIMEOUT=20
+LLM_ENRICHMENT_BATCH_SIZE=8
+LLM_ENRICHMENT_MAX_BATCH_CHARS=6000
+BACKGROUND_ENRICHMENT_JOB_SIZE=4
+
+# Optional OpenAI-style aliases
+OPENAI_MODEL=
+OPENAI_BASE_URL=
+OPENAI_API_KEY=
+
+# Optional LLM query parsing for search
+QUERY_PARSER_LLM_ENABLED=false
+QUERY_PARSER_LLM_MODEL=gpt-4o-mini
+QUERY_PARSER_LLM_BASE_URL=
+QUERY_PARSER_LLM_API_KEY=
+QUERY_PARSER_LLM_TIMEOUT=15

.github/workflows/code-scans.yaml

@@ -37,7 +37,7 @@ jobs:
         run: mkdir -p trivy-reports
 
       - name: Run Trivy FS Scan
-        uses: aquasecurity/trivy-action@0.24.0
+        uses: aquasecurity/trivy-action@0.35.0
         with:
           scan-type: 'fs'
           scan-ref: '.'

.gitignore

@@ -64,6 +64,6 @@ startup_error.txt
 .dockerignore
 
 
-
+/tests/*
 
 

README.md

@@ -41,6 +41,7 @@ Microservices-based marketplace platform for product ingestion, catalog manageme
 2. **Index Building**: After product changes, the Product Service triggers the Search Service indexing endpoint to rebuild semantic and keyword indices.
 3. **Natural Language Query Parsing**: Search queries are parsed for category, price bounds, and sort intent using spaCy + rule-based logic.
 4. **Hybrid Retrieval and Ranking**: The Search Service combines FAISS semantic similarity with BM25 keyword matching, applies dynamic weighting, and returns ranked results through the gateway.
+5. **LLM Enrichment**: The Product Service generate a concise short description during ingestion through an OpenAI-compatible chat API, with deterministic fallback if the model call is disabled or unavailable.
 
 The codebase is implemented with FastAPI services, a React + Vite frontend, and Docker Compose orchestration. It supports background indexing, category-aware boosting, and filtered/sorted query responses.
 
@@ -143,7 +144,41 @@ git clone <your-repo-url>
 cd syanapseMart
 ```
 
-#### 2. Launch the Full Stack
+#### 2. Create Your Environment File
+
+```bash
+cp .env.example .env
+```
+
+Edit `.env` with your preferred model endpoint. For OpenAI:
+
+```dotenv
+LLM_ENRICHMENT_ENABLED=true
+LLM_ENRICHMENT_MODEL=gpt-4o-mini
+LLM_ENRICHMENT_BASE_URL=https://api.openai.com/v1
+LLM_ENRICHMENT_API_KEY=your_api_key
+LLM_ENRICHMENT_BATCH_SIZE=8
+LLM_ENRICHMENT_MAX_BATCH_CHARS=6000
+BACKGROUND_ENRICHMENT_JOB_SIZE=4
+```
+
+For a local Ollama server with an OpenAI-compatible API:
+
+```dotenv
+LLM_ENRICHMENT_ENABLED=true
+LLM_ENRICHMENT_MODEL=llama3.2
+LLM_ENRICHMENT_BASE_URL=http://host.docker.internal:11434/v1
+LLM_ENRICHMENT_API_KEY=ollama
+LLM_ENRICHMENT_BATCH_SIZE=4
+LLM_ENRICHMENT_MAX_BATCH_CHARS=3000
+BACKGROUND_ENRICHMENT_JOB_SIZE=4
+QUERY_PARSER_LLM_ENABLED=true
+QUERY_PARSER_LLM_MODEL=llama3.2
+QUERY_PARSER_LLM_BASE_URL=http://host.docker.internal:11434/v1
+QUERY_PARSER_LLM_API_KEY=ollama
+```
+
+#### 3. Launch the Full Stack
 
 ```bash
 docker compose up --build -d
@@ -155,14 +190,14 @@ For live logs:
 docker compose up --build
 ```
 
-#### 3. Access the Application
+#### 4. Access the Application
 
-- **Frontend UI**: http://localhost
+- **Frontend UI**: http://localhost:80
 - **Gateway API**: http://localhost:8000
 - **Product Service Docs**: http://localhost:8001/docs
 - **Search Service Docs**: http://localhost:8002/docs
 
-#### 4. Verify Services
+#### 5. Verify Services
 
 ```bash
 # Gateway health
@@ -176,7 +211,7 @@ curl "http://localhost:8000/api/search?q=laptop"
 docker compose ps
 ```
 
-#### 5. Stop the Application
+#### 6. Stop the Application
 
 ```bash
 docker compose down
@@ -277,14 +312,27 @@ syanapseMart/
 
 ## Environment Variables
 
-SynapseMart works out of the box with defaults, but these env vars can be configured:
+Create a root `.env` from `.env.example` and keep service configuration there.
 
 | Variable | Service | Description | Default |
 |----------|---------|-------------|---------|
 | `PRODUCT_SERVICE_URL` | Gateway | Internal URL used by gateway for product calls | `http://product-service:8001` |
 | `SEARCH_SERVICE_URL` | Gateway | Internal URL used by gateway for search calls | `http://search-service:8002` |
 | `SEARCH_SERVICE_URL` | Product | URL used by product service to trigger indexing | `http://search-service:8002` |
 | `PRODUCT_SERVICE_URL` | Search | URL used by search service for lazy category loading | `http://product-service:8001` |
+| `LLM_ENRICHMENT_ENABLED` | Product | Enables upload-time LLM enrichment | `false` |
+| `LLM_ENRICHMENT_MODEL` | Product | OpenAI-compatible chat model name | `gpt-4o-mini` |
+| `LLM_ENRICHMENT_BASE_URL` | Product | Base URL for OpenAI-compatible `/v1` endpoint | `https://api.openai.com/v1` |
+| `LLM_ENRICHMENT_API_KEY` | Product | API key or local placeholder token | empty |
+| `LLM_ENRICHMENT_TIMEOUT` | Product | Timeout for enrichment calls in seconds | `20` |
+| `LLM_ENRICHMENT_BATCH_SIZE` | Product | Max products per LLM enrichment request | `8` |
+| `LLM_ENRICHMENT_MAX_BATCH_CHARS` | Product | Max serialized payload size per enrichment request | `6000` |
+| `BACKGROUND_ENRICHMENT_JOB_SIZE` | Product | Number of products processed per background job batch | `4` |
+| `QUERY_PARSER_LLM_ENABLED` | Search | Enables LLM-based natural language query parsing | `false` |
+| `QUERY_PARSER_LLM_MODEL` | Search | OpenAI-compatible query parser model | `gpt-4o-mini` |
+| `QUERY_PARSER_LLM_BASE_URL` | Search | Base URL for the parser model endpoint | empty |
+| `QUERY_PARSER_LLM_API_KEY` | Search | API key or local placeholder token | empty |
+| `QUERY_PARSER_LLM_TIMEOUT` | Search | Timeout for query parser calls in seconds | `15` |
 | `SENTENCE_TRANSFORMERS_HOME` | Search | Cache directory for transformer model files | `/app/model_cache` |
 | `VITE_API_URL` | UI (dev) | Frontend API base URL during local `vite` runs | `http://localhost:8000` |
 
@@ -372,4 +420,3 @@ This project is licensed under the [MIT License](LICENSE).
 - Search ranking quality depends on uploaded data quality and query phrasing.
 - NLP interpretation of price/category/sort intent may not be perfect for every query.
 - Validate outputs before using this system in production workflows.
-

docker-compose.yml

@@ -4,8 +4,8 @@ services:
     build:
       context: ./services/product
     container_name: synapse-product
-    environment:
-      - SEARCH_SERVICE_URL=http://search-service:8002
+    env_file:
+      - ./.env
     volumes:
       - product-data:/app/data
       - ./logs:/app/logs
@@ -18,8 +18,9 @@ services:
     build:
       context: ./services/search
     container_name: synapse-search
+    env_file:
+      - ./.env
     environment:
-      - PRODUCT_SERVICE_URL=http://product-service:8001
       - SENTENCE_TRANSFORMERS_HOME=/app/model_cache
     volumes:
       - ./.model_cache:/app/model_cache
@@ -35,9 +36,8 @@ services:
     container_name: synapse-gateway
     ports:
       - "8000:8000"
-    environment:
-      - PRODUCT_SERVICE_URL=http://product-service:8001
-      - SEARCH_SERVICE_URL=http://search-service:8002
+    env_file:
+      - ./.env
     depends_on:
       - product-service
       - search-service

gateway/app/api/proxy.py

@@ -34,8 +34,8 @@ async def get_categories():
         logger.error(f"Product service error: {e}")
         raise HTTPException(status_code=503, detail="Product service unavailable")
 
-@router.post("/upload")
-async def upload_proxy(request: Request):
+@router.post("/upload")
+async def upload_proxy(request: Request):
     """Forward multipart CSV upload."""
     target_url = f"{PRODUCT_SERVICE_URL}/products/upload"
     try:
@@ -45,9 +45,19 @@ async def upload_proxy(request: Request):
         body = await request.body()
         resp = await client.post(target_url, content=body, headers=headers, timeout=60.0)
         return resp.json()
-    except Exception as e:
-        logger.error(f"Upload proxy error: {e}")
-        raise HTTPException(status_code=503, detail="Upload service unavailable")
+    except Exception as e:
+        logger.error(f"Upload proxy error: {e}")
+        raise HTTPException(status_code=503, detail="Upload service unavailable")
+
+@router.get("/upload-jobs/{job_id}")
+async def upload_job_proxy(job_id: str):
+    """Fetch background upload job status."""
+    try:
+        resp = await client.get(f"{PRODUCT_SERVICE_URL}/products/jobs/{job_id}", timeout=15.0)
+        return resp.json()
+    except Exception as e:
+        logger.error(f"Upload job proxy error: {e}")
+        raise HTTPException(status_code=503, detail="Upload job service unavailable")
 
 @router.get("/search")
 async def search_proxy(request: Request):

services/product/app/api/routes.py

@@ -1,46 +1,41 @@
-from fastapi import APIRouter, Depends, UploadFile, File, HTTPException, BackgroundTasks
-from sqlalchemy.orm import Session
-from typing import List, Optional
-from ..core.database import get_db
-from ..models.product import Product
-from ..services.catalog import process_csv_upload, trigger_indexing
-
-router = APIRouter()
+from fastapi import APIRouter, Depends, UploadFile, File, HTTPException, BackgroundTasks
+from sqlalchemy.orm import Session
+from typing import List, Optional
+from ..core.database import get_db
+from ..models.product import Product
+from ..services.catalog import load_all_products_data, process_csv_upload, reindex_catalog, run_upload_job, start_upload_job
+from ..services.jobs import upload_job_store
+
+router = APIRouter()
 
 @router.post("/upload")
 async def upload_products(
     background_tasks: BackgroundTasks,
     file: UploadFile = File(...),
-    db: Session = Depends(get_db)
-):
-    """Handle product CSV uploads."""
-    try:
-        added_count, skipped_count = process_csv_upload(file, db)
-
-        # Prepare data for search indexing
-        all_products = db.query(Product).all()
-        products_data = [{
-            "id": p.id,
-            "name": p.name,
-            "description": p.description,
-            "category": p.category,
-            "price": p.price,
-            "currency": p.currency,
-            "stock_quantity": p.stock_quantity,
-            "seller_name": p.seller_name,
-            "image_url": p.image_url
-        } for p in all_products]
-
-        background_tasks.add_task(trigger_indexing, products_data)
-
-        return {
-            "message": f"Processed {added_count} products",
-            "count": added_count,
-            "skipped": skipped_count,
-            "total_products": len(all_products)
-        }
-    except Exception as e:
-        raise HTTPException(status_code=400, detail=str(e))
+    db: Session = Depends(get_db)
+):
+    """Handle product CSV uploads."""
+    try:
+        upload_result = process_csv_upload(file, db)
+        all_products_data = load_all_products_data(db)
+        background_tasks.add_task(reindex_catalog)
+
+        job = start_upload_job(upload_result["product_ids"])
+        if job:
+            background_tasks.add_task(run_upload_job, job["job_id"], upload_result["product_ids"])
+
+        return {
+            "message": (
+                f"Accepted {upload_result['added_count']} products for upload. "
+                "Search is available immediately; enrichment continues in the background."
+            ),
+            "count": upload_result["added_count"],
+            "skipped": upload_result["skipped_count"],
+            "total_products": len(all_products_data),
+            "job": job,
+        }
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=str(e))
 
 @router.get("/")
 def list_products(
@@ -62,9 +57,17 @@ def get_categories(db: Session = Depends(get_db)):
     return [c[0] for c in categories if c[0]]
 
 @router.delete("/")
-def clear_catalog(background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
-    """Wipe the product catalog."""
-    count = db.query(Product).delete()
-    db.commit()
-    background_tasks.add_task(trigger_indexing, [])
-    return {"message": f"Deleted {count} products"}
+def clear_catalog(background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
+    """Wipe the product catalog."""
+    count = db.query(Product).delete()
+    db.commit()
+    background_tasks.add_task(reindex_catalog)
+    return {"message": f"Deleted {count} products"}
+
+
+@router.get("/jobs/{job_id}")
+def get_upload_job(job_id: str):
+    job = upload_job_store.get(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Upload job not found")
+    return job

services/product/app/core/config.py

@@ -0,0 +1,27 @@
+import os
+
+
+def get_bool_env(name: str, default: bool = False) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+SEARCH_SERVICE_URL = os.getenv("SEARCH_SERVICE_URL", "http://search-service:8002")
+
+LLM_ENRICHMENT_ENABLED = get_bool_env("LLM_ENRICHMENT_ENABLED", False)
+LLM_ENRICHMENT_MODEL = os.getenv("LLM_ENRICHMENT_MODEL", os.getenv("OPENAI_MODEL", "gpt-4o-mini"))
+LLM_ENRICHMENT_BASE_URL = (
+    os.getenv("LLM_ENRICHMENT_BASE_URL", "").strip()
+    or os.getenv("OPENAI_BASE_URL", "").strip()
+    or None
+)
+LLM_ENRICHMENT_API_KEY = (
+    os.getenv("LLM_ENRICHMENT_API_KEY", "").strip()
+    or os.getenv("OPENAI_API_KEY", "").strip()
+)
+LLM_ENRICHMENT_TIMEOUT = float(os.getenv("LLM_ENRICHMENT_TIMEOUT", "20"))
+LLM_ENRICHMENT_BATCH_SIZE = max(1, int(os.getenv("LLM_ENRICHMENT_BATCH_SIZE", "8")))
+LLM_ENRICHMENT_MAX_BATCH_CHARS = max(500, int(os.getenv("LLM_ENRICHMENT_MAX_BATCH_CHARS", "6000")))
+BACKGROUND_ENRICHMENT_JOB_SIZE = max(1, int(os.getenv("BACKGROUND_ENRICHMENT_JOB_SIZE", "4")))

services/product/app/core/database.py

@@ -1,7 +1,8 @@
-from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker
-import os
+from sqlalchemy import create_engine
+from sqlalchemy import inspect, text
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+import os
 
 # Ensure data directory exists (relative to the container root)
 os.makedirs("data", exist_ok=True)
@@ -11,11 +12,24 @@
 engine = create_engine(SQLITE_URL, connect_args={"check_same_thread": False})
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
-Base = declarative_base()
-
-def get_db():
-    db = SessionLocal()
-    try:
-        yield db
+Base = declarative_base()
+
+def ensure_schema():
+    """Backfill additive columns for existing SQLite databases."""
+    inspector = inspect(engine)
+    if "products" not in inspector.get_table_names():
+        return
+
+    existing_columns = {column["name"] for column in inspector.get_columns("products")}
+    with engine.begin() as connection:
+        if "short_description" not in existing_columns:
+            connection.execute(text("ALTER TABLE products ADD COLUMN short_description VARCHAR"))
+        if "search_text" not in existing_columns:
+            connection.execute(text("ALTER TABLE products ADD COLUMN search_text VARCHAR"))
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
     finally:
         db.close()