backend apis updation

This commit is contained in:
sriram
2026-08-13 15:54:29 +05:30
parent b8d93fbbf2
commit 241fd237f8
13 changed files with 973 additions and 105 deletions

22
.gitattributes vendored Normal file
View File

@@ -0,0 +1,22 @@
# Line endings, pinned so they do not depend on each developer's Git config.
#
# Git for Windows ships core.autocrlf=true in its system gitconfig, which
# stores LF but checks out CRLF - the source of the "LF will be replaced by
# CRLF" warnings on `git add`. Declaring the policy here overrides that for
# everyone, so Windows, Linux and the Docker build all agree.
#
# The repository is already entirely LF (`git ls-files --eol` shows i/lf
# across the board), so eol=lf pins the working tree to what the index
# already holds and rewrites no committed content.
* text=auto eol=lf
# Binaries must never be EOL-converted - a substituted byte corrupts them.
# Git already auto-detects these; saying so explicitly means a future edit
# to the wildcard above cannot silently start mangling them.
*.joblib binary
*.db binary
# Windows-only scripts genuinely need CRLF. None are tracked in this repo
# today (start_app.bat lives above it), but the rule belongs with the policy.
*.bat text eol=crlf
*.ps1 text eol=crlf

View File

@@ -1,12 +1,22 @@
from __future__ import annotations
import re
from typing import Optional
from fastapi import APIRouter, HTTPException, Query
from app.api.schemas import AllProductsOut, BrandsOut, CategoriesOut, ProductListOut, ProductOut
from app.api.schemas import (
AllProductsOut,
BrandCardOut,
BrandCardsOut,
BrandsOut,
CategoriesOut,
ProductListOut,
ProductOut,
)
from app.services.s3_service import s3_service
from app.services.vector_store import (
get_brand_overview,
list_available_brands,
list_categories_for_brand,
get_products_by_brand,
@@ -106,6 +116,53 @@ def get_brands() -> BrandsOut:
return BrandsOut(brands=list_available_brands())
def _initials(name: str) -> str:
"""Monogram for the card's image fallback - there are no logo assets."""
words = [w for w in re.split(r"[^A-Za-z0-9]+", name) if w]
if not words:
return "?"
if len(words) == 1:
return words[0][:2].upper()
return (words[0][0] + words[1][0]).upper()
# Declared before /brands/{brand}/... so "overview" is never read as a brand
# name. The path-segment counts differ, so this is belt-and-braces.
@router.get("/brands/overview", response_model=BrandCardsOut)
def get_brand_cards(
refresh: bool = Query(False, description="Bypass the short-lived overview cache"),
) -> BrandCardsOut:
"""Per-brand summaries for the home page card grid.
Additive: GET /brands keeps returning a plain list of names, which the
sidebar and the admin/user pages rely on.
"""
rows = get_brand_overview(force_refresh=refresh)
cards = []
for row in rows:
name = row["display_name"]
image_url = _clean_url(row.get("sample_image_url"))
if not image_url and s3_service.enabled and row.get("sample_image_id"):
image_url = s3_service.get_product_image_url(name, row["sample_image_id"])
cards.append(BrandCardOut(
name=name,
slug=row["suffix"],
product_count=row["product_count"],
category_count=row["category_count"],
categories=list(row.get("categories") or []),
image_url=image_url,
initials=_initials(name),
))
return BrandCardsOut(
total_brands=len(cards),
total_products=sum(c.product_count for c in cards),
brands=cards,
)
@router.get("/brands/{brand}/categories", response_model=CategoriesOut)
def get_brand_categories(brand: str) -> CategoriesOut:
return CategoriesOut(brand=brand, categories=list_categories_for_brand(brand))

View File

@@ -48,6 +48,26 @@ def _run_background_auto_seed():
logger.error("Background Auto-Init error: %s", e)
def _run_background_brand_sync() -> None:
"""Reconcile brand tables against data/seed_catalogs/ in both directions.
Complements the auto-seed above rather than replacing it: that one only
fires against a completely empty database, so without this a brand table
created after first boot never gets a seed file, and a seed file added
after first boot is never loaded.
"""
try:
from app.services.brand_sync import reconcile_brand_catalogs
summary = reconcile_brand_catalogs()
logger.info("🔁 Brand catalog reconcile: %s", summary)
except Exception as e:
logger.error("Brand catalog reconcile error: %s", e)
class BrandSyncRequest(BaseModel):
dry_run: bool = False
@router.get("/system/status", response_model=SystemStatusOut)
def get_system_status() -> SystemStatusOut:
"""Return unified status of database, vector store, stores, and frontend build."""
@@ -89,3 +109,23 @@ def initialize_system(background_tasks: BackgroundTasks) -> Dict[str, Any]:
"status": "started",
"message": "Background initialization triggered. Check /api/system/status for progress.",
}
@router.post("/system/brand-sync", dependencies=[Depends(require_admin)])
def sync_brand_catalogs(payload: BrandSyncRequest, background_tasks: BackgroundTasks) -> Dict[str, Any]:
"""Reconcile brand tables with their seed catalog files.
`dry_run` answers inline - it is a handful of count queries and writes
nothing, so it is safe to poke at. A real run is backgrounded because
exporting a large brand serialises thousands of 384-float embeddings.
"""
from app.services.brand_sync import reconcile_brand_catalogs
if payload.dry_run:
return {"status": "ok", "summary": reconcile_brand_catalogs(dry_run=True)}
background_tasks.add_task(_run_background_brand_sync)
return {
"status": "started",
"message": "Brand catalog reconcile triggered. Check /api/system/status for progress.",
}

View File

@@ -1,8 +1,6 @@
import io
import json
import logging
import re
from pathlib import Path
from typing import Any, Dict, List, Optional
import pandas as pd
from pydantic import BaseModel, Field
@@ -16,14 +14,13 @@ from app.services.vector_store import (
_sanitize_name,
get_products_by_brand,
)
from app.services.brand_sync import upsert_products_into_catalog_file
from app.services.embeddings_service import embed_texts
from app.services.s3_service import s3_service
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/user/products", tags=["user_products"])
SEED_DIR = Path(__file__).resolve().parents[3] / "data" / "seed_catalogs"
class AddProductRequest(BaseModel):
brand: str = Field(..., description="Brand name, e.g. Lion Dates")
@@ -174,50 +171,14 @@ def _enrich_and_save_product(req: AddProductRequest) -> Dict[str, Any]:
def _update_json_catalog_file(brand: str, product_dict: Dict[str, Any]) -> None:
SEED_DIR.mkdir(parents=True, exist_ok=True)
"""Append/update one product in the brand's seed catalog.
# Determine seed file name (e.g. brand_catalog_lion_dates.json)
brand_slug = _sanitize_name(resolve_parent_brand(brand))
file_path = SEED_DIR / f"brand_catalog_{brand_slug}.json"
# Strip embedding before saving to JSON file for clean JSON size
clean_dict = {k: v for k, v in product_dict.items() if k != "embedding"}
if file_path.exists():
try:
data = json.loads(file_path.read_text(encoding="utf-8-sig"))
except Exception as e:
logger.warning("Could not read existing catalog JSON %s: %s", file_path.name, e)
data = {"brand": brand, "products": []}
else:
data = {
"brand": brand.lower(),
"search_query": f"{brand} products catalog",
"generation_timestamp": str(Path(__file__).resolve()),
"total_products": 0,
"total_images": 0,
"products": [],
}
products_list = data.get("products", [])
# Replace existing or append new product
updated = False
for i, p in enumerate(products_list):
if p.get("image_id") == clean_dict["image_id"] or p.get("product_name") == clean_dict["product_name"]:
products_list[i] = clean_dict
updated = True
break
if not updated:
products_list.append(clean_dict)
data["products"] = products_list
data["total_products"] = len(products_list)
data["total_images"] = sum(len(p.get("image_urls") or []) for p in products_list)
file_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
logger.info("✅ Updated JSON seed file '%s' (total products: %d)", file_path.name, data["total_products"])
Delegates to brand_sync so this shares the file-resolution rules with the
startup reconcile. That also fixes a mis-targeting bug this used to have:
picking the file by sanitised slug wrote P&G products to a new
brand_catalog_p_g.json instead of the real brand_catalog_p_and_g.json.
"""
upsert_products_into_catalog_file(brand, [product_dict])
@router.post("/add", status_code=201, dependencies=[Depends(require_permission("add_product"))])

View File

@@ -79,6 +79,27 @@ class BrandsOut(BaseModel):
brands: List[str]
class BrandCardOut(BaseModel):
"""A brand as shown on the home page card grid.
`name` is the same string GET /api/brands returns, so the frontend can
pass it straight back to /api/brands/{brand}/products.
"""
name: str
slug: str
product_count: int
category_count: int
categories: List[str] = Field(default_factory=list)
image_url: Optional[str] = None
initials: str
class BrandCardsOut(BaseModel):
total_brands: int
total_products: int
brands: List[BrandCardOut]
class CategoriesOut(BaseModel):
brand: str
categories: List[str]

View File

@@ -32,6 +32,20 @@ async def ingest_brand(brand: str, max_products: int = 50) -> Dict[str, Any]:
catalog = await catalog_engine.generate_catalog(brand=brand, max_products=max_products)
duration = time.time() - start
# Mirror the freshly ingested rows into data/seed_catalogs/ so a brand that
# arrived through the pipeline is reproducible from disk like the bundled
# ones. Reading back from pgvector rather than reusing catalog["products"]
# keeps the file honest about what was actually stored.
#
# Placed here rather than in the API router because this function is also
# the CLI's entry point (cli/ingest_brand.py), and a failure to write the
# file must not turn a successful ingest into a failed job.
try:
from app.services.brand_sync import export_brand_to_seed_file
export_brand_to_seed_file(brand)
except Exception: # noqa: BLE001 - DB rows are already committed
logger.warning("Seed-catalog export failed for %s (DB rows intact)", brand, exc_info=True)
summary = {
"brand": brand,
"total_products": catalog.get("total_products", 0),

View File

@@ -47,8 +47,13 @@ async def lifespan(_app: FastAPI):
try:
ensure_store_intelligence_schema()
ensure_nutrition_schema()
from app.api.routers.system import _run_background_auto_seed
from app.api.routers.system import _run_background_auto_seed, _run_background_brand_sync
_run_background_auto_seed()
# Runs after the auto-seed so a cold boot has already loaded the
# bundled catalogs and this finds nothing to do. On a warm boot the
# auto-seed no-ops and this is what picks up a brand table or seed
# file that appeared since last time.
_run_background_brand_sync()
except Exception as e:
logger.warning("Startup background init warning: %s", e)

View File

@@ -1,3 +1,6 @@
import re
from functools import lru_cache
BRAND_ALIASES = {
# Cadbury family
"cadbury gems": "cadbury",
@@ -253,6 +256,21 @@ BRAND_ALIASES = {
DEFAULT_ALIASES = BRAND_ALIASES
def _contains_word(haystack: str, needle: str) -> bool:
"""True when `needle` occurs in `haystack` as a whole word.
Plain `in` would treat any fragment as a match, so a short brand name
could be swallowed by an unrelated alias that merely contains those
letters - e.g. "sun" inside "hul sunsilk". Anchoring both ends on a
word boundary keeps the genuine multi-word hits ("tata" inside
"hul tata tea") while dropping the fragment ones.
"""
if not needle:
return False
return re.search(r"(?<!\w)" + re.escape(needle) + r"(?!\w)", haystack) is not None
@lru_cache(maxsize=1024)
def resolve_parent_brand(brand: str) -> str:
"""Return the parent (canonical) brand for storage purposes.
@@ -260,15 +278,19 @@ def resolve_parent_brand(brand: str) -> str:
returns the parent brand name so that sub-brands share the same
database table, S3 folder, and JSON file as their parent.
Falls back to fuzzy substring matching, then returns the input
unchanged if no alias is known.
Falls back to whole-word matching in either direction, then returns
the input unchanged if no alias is known.
Cached because `_table_name()` in vector_store calls this on every
query and the fallback loop scans all ~230 aliases. BRAND_ALIASES is
a module constant that is never mutated, so the result is stable.
"""
key = brand.lower().strip()
direct = BRAND_ALIASES.get(key)
if direct:
return direct
for alias, parent in BRAND_ALIASES.items():
if alias in key or key in alias:
if _contains_word(key, alias) or _contains_word(alias, key):
return parent
return brand

433
app/services/brand_sync.py Normal file
View File

@@ -0,0 +1,433 @@
"""
Keeps the `brand_*` Postgres tables and `data/seed_catalogs/*.json` in step.
A brand can enter the system from either side: `POST /api/user/products/add`
and `POST /api/catalog/generate` write rows, while a catalog file may be
dropped in by hand or shipped in the image. Neither side used to produce the
other, so a brand ingested at runtime had no seed file, and a seed file added
after first boot was never loaded (the startup auto-seed only runs against a
completely empty database).
This module is the single place that knows the correspondence between the two,
and `reconcile_brand_catalogs()` repairs it in both directions.
The correspondence is *not* the filename. `brand_catalog_p_and_g.json` holds
`"brand": "p&g"`, which sanitises to table `brand_p_g`; `brand_catalog_tata.json`
resolves through BRAND_ALIASES into `brand_hindustan_unilever`. Everything here
therefore indexes files by their `brand` field, exactly as
`scripts/seed_sample_data.py` does. Keying on filenames instead would split
P&G across two files and let Tata clobber Hindustan Unilever.
"""
from __future__ import annotations
import json
import logging
import os
from collections import defaultdict
from decimal import Decimal
from pathlib import Path
from typing import Any, Dict, List, Optional
from app.services.brand_registry import resolve_parent_brand
from app.services.vector_store import (
_connect,
_list_brand_table_suffixes,
_sanitize_name,
display_name_for_suffix,
ensure_brand_schema,
get_products_by_brand,
invalidate_brand_overview_cache,
upsert_brand_products,
)
logger = logging.getLogger(__name__)
# app/services/brand_sync.py -> parents[2] is backend/
SEED_DIR = Path(__file__).resolve().parents[2] / "data" / "seed_catalogs"
# The column list written by upsert_brand_products, minus `embedding` (handled
# separately because it is huge and optional). Keeping these in sync is what
# makes an exported file round-trip back through the seeder without losing
# hsn/price/barcode/sku data - the failing mode of scripts/export_seed_data.py.
EXPORT_COLUMNS = (
"product_name", "title", "description", "category", "image_id",
"image_url", "image_urls", "price_range", "size_variants", "providers",
"fssai_license", "product_sku", "sku_source", "hsn_code",
"final_selling_price", "selling_price", "barcode", "barcode_type",
"highlights", "nutrients", "search_query",
)
def brand_slug(brand: str) -> str:
"""The table suffix a brand name resolves to (e.g. 'ITC' -> 'itc')."""
return _sanitize_name(resolve_parent_brand(brand))
def _read_catalog(path: Path) -> Optional[Dict[str, Any]]:
"""Parse a seed catalog, returning None for anything that isn't one.
Files without a brand or products (notably `hsn_gst_master.json`) are not
catalogs, and a corrupt file must not take down a startup reconcile.
"""
try:
data = json.loads(path.read_text(encoding="utf-8-sig"))
except Exception as e: # noqa: BLE001 - one bad file cannot break the sweep
logger.warning("Could not read seed catalog %s: %s", path.name, e)
return None
if not isinstance(data, dict):
return None
products = data.get("products")
if not isinstance(products, list) or not products:
return None
brand = data.get("brand") or (products[0].get("brand_name") if isinstance(products[0], dict) else None)
if not brand:
return None
data["brand"] = brand
return data
def index_seed_files(seed_dir: Path = SEED_DIR) -> Dict[str, List[Path]]:
"""Map each brand table suffix to the seed files that feed it.
More than one file can feed a suffix - tata.json and hindustan_unilever.json
both land in brand_hindustan_unilever - which is why the value is a list.
"""
index: Dict[str, List[Path]] = defaultdict(list)
if not seed_dir.exists():
return {}
for path in sorted(seed_dir.glob("*.json")):
data = _read_catalog(path)
if data is None:
continue
index[brand_slug(data["brand"])].append(path)
return dict(index)
# Resolving a brand to its file means parsing every catalog, and /batch-add
# calls that once per product. The mapping only changes when a file is created
# or removed, so it is cached and invalidated explicitly on write.
_TARGET_CACHE: Dict[str, Path] = {}
def invalidate_seed_index_cache() -> None:
_TARGET_CACHE.clear()
def canonical_seed_file(brand: str, index: Optional[Dict[str, List[Path]]] = None) -> Path:
"""The file to write for `brand`.
Prefers a file whose own brand field sanitises to the same slug, so P&G
products append to brand_catalog_p_and_g.json rather than creating an
orphan brand_catalog_p_g.json, and Hindustan Unilever products never get
written into brand_catalog_tata.json.
"""
slug = brand_slug(brand)
if index is None:
cached = _TARGET_CACHE.get(slug)
if cached is not None:
return cached
index = index_seed_files()
resolved = SEED_DIR / f"brand_catalog_{slug}.json"
candidates = index.get(slug, [])
for path in candidates:
data = _read_catalog(path)
if data and _sanitize_name(data["brand"]) == slug:
resolved = path
break
else:
if candidates:
resolved = candidates[0]
_TARGET_CACHE[slug] = resolved
return resolved
def upsert_products_into_catalog_file(brand: str, products: List[Dict[str, Any]]) -> Optional[Path]:
"""Merge `products` into the brand's seed catalog, creating it if needed.
Products are matched on image_id or product_name, so re-adding an existing
product updates it in place instead of duplicating. Embeddings are stripped
(they are ~384 floats each and the file is meant to stay readable), and the
write is atomic so a crash cannot leave a half-written catalog that then
fails to parse on the next boot.
"""
if not products:
return None
SEED_DIR.mkdir(parents=True, exist_ok=True)
file_path = canonical_seed_file(brand)
if file_path.exists():
try:
data = json.loads(file_path.read_text(encoding="utf-8-sig"))
except Exception as e: # noqa: BLE001
logger.warning("Could not read existing catalog JSON %s: %s", file_path.name, e)
data = {"brand": brand, "products": []}
else:
data = {
"brand": brand.lower(),
"search_query": f"{brand} products catalog",
"generation_timestamp": str(Path(__file__).resolve()),
"total_products": 0,
"total_images": 0,
"products": [],
}
products_list = data.get("products") or []
by_image_id = {
p.get("image_id"): i for i, p in enumerate(products_list) if p.get("image_id")
}
by_name = {
p.get("product_name"): i for i, p in enumerate(products_list) if p.get("product_name")
}
for product in products:
clean = {k: v for k, v in product.items() if k != "embedding"}
idx = by_image_id.get(clean.get("image_id"))
if idx is None:
idx = by_name.get(clean.get("product_name"))
if idx is None:
products_list.append(clean)
if clean.get("image_id"):
by_image_id[clean["image_id"]] = len(products_list) - 1
if clean.get("product_name"):
by_name[clean["product_name"]] = len(products_list) - 1
else:
products_list[idx] = clean
data["products"] = products_list
data["total_products"] = len(products_list)
data["total_images"] = sum(len(p.get("image_urls") or []) for p in products_list)
payload = json.dumps(data, indent=2, ensure_ascii=False)
tmp_path = file_path.with_suffix(".json.tmp")
tmp_path.write_text(payload, encoding="utf-8")
os.replace(tmp_path, file_path)
logger.info("✅ Updated JSON seed file '%s' (total products: %d)",
file_path.name, data["total_products"])
return file_path
def _jsonable(value: Any) -> Any:
"""Coerce a psycopg row value into something json.dumps accepts."""
if isinstance(value, Decimal):
return float(value)
if isinstance(value, (list, tuple)):
return [_jsonable(v) for v in value]
return value
def export_brand_to_seed_file(brand: str, include_embeddings: bool = True) -> Optional[Path]:
"""Write a brand's database rows out to its seed catalog, field-complete.
Deliberately not scripts/export_seed_data.py, whose _read_products emits
only ~10 keys - running that would silently strip hsn_code, prices,
barcodes, SKUs and FSSAI numbers out of the existing catalogs.
Embeddings are carried through when present because that is what lets
seed_sample_data.py re-seed without invoking the embedding model.
"""
rows = get_products_by_brand(brand)
if not rows:
logger.info("Nothing to export for brand '%s' (no rows)", brand)
return None
display = display_name_for_suffix(brand_slug(brand))
products: List[Dict[str, Any]] = []
for row in rows:
product: Dict[str, Any] = {"brand": display, "brand_name": display}
for col in EXPORT_COLUMNS:
if col in row:
product[col] = _jsonable(row[col])
if include_embeddings and row.get("embedding") is not None:
try:
product["embedding"] = [float(x) for x in row["embedding"]]
except (TypeError, ValueError):
pass
products.append(product)
path = upsert_products_into_catalog_file(display, products)
if path:
logger.info("📤 Exported %d product(s) for '%s' -> %s", len(products), display, path.name)
return path
def load_seed_catalogs(seed_dir: Path = SEED_DIR,
only: Optional[List[str]] = None) -> Dict[str, List[Dict[str, Any]]]:
"""Read seed catalogs and group their products by resolved parent brand.
Grouping matters: several files can feed one table, and the seeder's
stale-row cleanup deletes anything not in the batch it is given. Merging
first is what stops tata.json and hindustan_unilever.json erasing each
other.
"""
if not seed_dir.exists():
logger.error("Seed directory not found: %s", seed_dir)
return {}
files = sorted(seed_dir.glob("*.json"))
if only:
wanted = [w.lower() for w in only]
files = [f for f in files if any(w in f.name.lower() for w in wanted)]
brand_products: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
for path in files:
data = _read_catalog(path)
if data is None:
logger.warning("Skipping %s - no brand/products found", path.name)
continue
resolved = resolve_parent_brand(data["brand"])
brand_products[resolved].extend(data["products"])
logger.info("Read %d products from %s -> resolved brand '%s'",
len(data["products"]), path.name, resolved)
return dict(brand_products)
def seed_brands(brand_products: Dict[str, List[Dict[str, Any]]], cleanup: bool = True) -> int:
"""Upsert grouped products into their brand tables. Returns the row count."""
total = 0
for resolved_brand, all_products in brand_products.items():
logger.info("Seeding %d product(s) for brand '%s'", len(all_products), resolved_brand)
table = ensure_brand_schema(resolved_brand)
if not table:
logger.error("Could not create/verify table for brand '%s' - is pgvector reachable?",
resolved_brand)
continue
upsert_brand_products(resolved_brand, all_products, cleanup=cleanup)
logger.info("Seeded %d products for brand '%s' (table=%s)",
len(all_products), resolved_brand, table)
total += len(all_products)
return total
def _db_brand_counts() -> Dict[str, int]:
"""{table suffix: row count} for every brand_* table, on one connection."""
conn = _connect()
if not conn:
return {}
counts: Dict[str, int] = {}
try:
with conn, conn.cursor() as cur:
for suffix in sorted(set(_list_brand_table_suffixes(cur))):
try:
cur.execute(f"SELECT COUNT(*) FROM brand_{suffix}")
row = cur.fetchone()
counts[suffix] = int(row[0]) if row else 0
except Exception as e: # noqa: BLE001
logger.warning("Could not count brand_%s: %s", suffix, e)
except Exception as e: # noqa: BLE001
logger.error("Failed to enumerate brand tables: %s", e)
finally:
conn.close()
return counts
def _detect_collisions(seed_dir: Path = SEED_DIR) -> List[Dict[str, Any]]:
"""Distinct parent brands that sanitise onto one table (e.g. 'P&G' vs 'P G').
Grouped by *resolved parent*, not by the raw brand field: tata.json and
hindustan_unilever.json share a table because BRAND_ALIASES deliberately
merges them, which is not a collision. A genuine one is two unrelated
parents whose names differ only in characters _sanitize_name strips.
Reported rather than repaired - renaming a live table is a one-way door,
and nothing collides today. This is here so it surfaces the day it does.
"""
by_slug: Dict[str, set] = defaultdict(set)
if not seed_dir.exists():
return []
for path in sorted(seed_dir.glob("*.json")):
data = _read_catalog(path)
if data is None:
continue
parent = resolve_parent_brand(data["brand"]).strip().lower()
by_slug[_sanitize_name(parent)].add(parent)
return [
{"slug": slug, "brands": sorted(names)}
for slug, names in sorted(by_slug.items()) if len(names) > 1
]
def reconcile_brand_catalogs(dry_run: bool = False) -> Dict[str, Any]:
"""Repair the table <-> seed-file correspondence in both directions.
Idempotent and non-destructive:
* A populated table with no seed file gets one exported.
* A seed file whose table is empty or absent gets seeded, with
cleanup disabled - the table holds nothing this batch could be a
partial view of, so there is no stale row to remove and no way to
delete data by passing an incomplete set.
* A file that already maps to a populated table is left alone. That
rule is what stops brand_catalog_tata.json being overwritten with
Hindustan Unilever's merged rows.
"""
# Files may have appeared on disk since the last resolve (that is half of
# what this function exists to handle), so start from a cold index.
invalidate_seed_index_cache()
db_counts = _db_brand_counts()
file_index = index_seed_files()
collisions = _detect_collisions()
to_export = sorted(
suffix for suffix, count in db_counts.items()
if count > 0 and suffix not in file_index
)
to_seed = sorted(
slug for slug in file_index
if db_counts.get(slug, 0) == 0
)
summary: Dict[str, Any] = {
"tables": len(db_counts),
"files": len(file_index),
"exported": [],
"seeded": [],
"collisions": collisions,
"skipped": [],
"dry_run": dry_run,
}
if dry_run:
summary["exported"] = [display_name_for_suffix(s) for s in to_export]
summary["seeded"] = to_seed
return summary
for suffix in to_export:
display = display_name_for_suffix(suffix)
try:
path = export_brand_to_seed_file(display)
if path:
summary["exported"].append(display)
except Exception as e: # noqa: BLE001 - one brand must not stop the sweep
logger.error("Export failed for brand '%s': %s", display, e)
summary["skipped"].append({"brand": display, "reason": str(e)})
for slug in to_seed:
paths = file_index.get(slug, [])
try:
grouped = load_seed_catalogs(only=[p.name for p in paths])
if not grouped:
continue
seed_brands(grouped, cleanup=False)
summary["seeded"].append(slug)
except Exception as e: # noqa: BLE001
logger.error("Seeding failed for slug '%s': %s", slug, e)
summary["skipped"].append({"brand": slug, "reason": str(e)})
if summary["exported"] or summary["seeded"]:
invalidate_brand_overview_cache()
try:
from app.services.query_intent import invalidate_brand_mention_cache
invalidate_brand_mention_cache()
except Exception: # noqa: BLE001
pass
if collisions:
logger.warning("Brand slug collisions detected: %s", collisions)
return summary

View File

@@ -15,7 +15,8 @@ and free on CPU-only hardware. Two things are extracted:
from __future__ import annotations
import re
from typing import Dict, Optional
import time
from typing import Any, Dict, Optional
from app.services.category_registry import detect_category_from_text # noqa: F401 (re-exported)
@@ -170,18 +171,70 @@ BRAND_SEARCH_MAP = {
}
BRAND_MENTION_TTL_SECONDS = 300
# Populated lazily by _brand_index(); holds the static tables above merged with
# whatever brands actually exist in the vector store right now.
_MENTION_CACHE: Dict[str, Any] = {"at": 0.0, "known": None, "map": None}
def invalidate_brand_mention_cache() -> None:
"""Force the next _brand_index() call to re-read the live brand list.
Called after a brand is added or the seed catalogs are reconciled, so a
newly ingested brand becomes mentionable without waiting out the TTL.
"""
_MENTION_CACHE.update(at=0.0, known=None, map=None)
def _brand_index() -> tuple[list[str], Dict[str, str]]:
"""The static brand tables, plus any live brand they don't already cover.
KNOWN_BRANDS and BRAND_SEARCH_MAP carry hand-tuned synonyms ("coke",
"hul", "pepsi") that cannot be derived from the database, so they are the
base and always win - the live list only *adds* brands, via setdefault.
That is what lets a brand ingested at runtime be recognised in chat and
search without anyone editing this file.
Cached for BRAND_MENTION_TTL_SECONDS because this runs on every query and
list_available_brands() opens a fresh connection. If the database is
unreachable the static behaviour stands rather than failing the query.
"""
now = time.monotonic()
if _MENTION_CACHE["map"] is not None and now - _MENTION_CACHE["at"] < BRAND_MENTION_TTL_SECONDS:
return _MENTION_CACHE["known"], _MENTION_CACHE["map"]
known = list(KNOWN_BRANDS)
mapping = dict(BRAND_SEARCH_MAP)
try:
from app.services.vector_store import list_available_brands
covered = {k.lower() for k in known}
for brand in list_available_brands():
if brand.lower() not in covered:
known.append(brand)
covered.add(brand.lower())
mapping.setdefault(brand.lower(), brand)
except Exception: # noqa: BLE001 - a DB blip must not break intent parsing
pass
_MENTION_CACHE.update(at=now, known=known, map=mapping)
return known, mapping
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()
known_brands, brand_search_map = _brand_index()
# 1. Check direct search map
for alias in sorted(BRAND_SEARCH_MAP.keys(), key=len, reverse=True):
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]
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)
@@ -190,7 +243,7 @@ def extract_brand_mention(query: str) -> Optional[str]:
if re.search(pattern, lower):
parent = resolve_parent_brand(alias)
# Normalize to canonical known brand name case
for kb in KNOWN_BRANDS:
for kb in known_brands:
if kb.lower() == parent.lower():
return kb
return parent.title()

View File

@@ -4,7 +4,9 @@ from typing import List, Optional, Dict, Any
import json
import logging
import os
import re
import time
import psycopg
from app.infrastructure.settings import DATABASE_URL, USE_PGVECTOR, DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD
@@ -367,9 +369,19 @@ def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: b
deleted = cur.rowcount
if deleted:
logger.info(f"🗑️ Removed {deleted} stale product(s) from {table_name}")
conn.close()
# Every write path into the catalog funnels through here, so this is the
# one place that has to invalidate the derived views: the brand cards'
# counts, and the brand list that chat/search intent parsing scopes on.
invalidate_brand_overview_cache()
try:
from app.services.query_intent import invalidate_brand_mention_cache
invalidate_brand_mention_cache()
except Exception: # noqa: BLE001 - cache invalidation must never fail a write
pass
def get_existing_product_image_id(brand: str, product_name: str) -> Optional[str]:
"""Check if a product with this name exists in the brand table and return its image_id"""
@@ -412,20 +424,45 @@ def _build_sanitized_brand_map() -> Dict[str, str]:
return seen
# Acronyms that .title() would mangle ("itc" -> "Itc"). Only needed for brands
# whose parent value in BRAND_ALIASES is lowercase and not a real word.
BRAND_DISPLAY_OVERRIDES = {
"itc": "ITC",
"grb": "GRB",
"hul": "HUL",
}
def display_name_for_suffix(suffix: str, brand_map: Optional[Dict[str, str]] = None) -> str:
"""Turn a brand table suffix back into the name shown in the UI.
Shared by list_available_brands() and get_brand_overview() so the sidebar
list and the brand cards can never disagree - the frontend passes these
strings straight back to /api/brands/{brand}/products, so they must round
trip through resolve_parent_brand + _sanitize_name to the same table.
"""
if brand_map is None:
brand_map = _build_sanitized_brand_map()
override = BRAND_DISPLAY_OVERRIDES.get(suffix)
if override:
return override
return brand_map.get(suffix) or suffix.replace('_', ' ').title()
def list_available_brands() -> List[str]:
"""List all available brands that have tables in the database"""
conn = _connect()
if not conn:
return []
brand_map = _build_sanitized_brand_map()
brands = []
with conn, conn.cursor() as cur:
try:
# Query information_schema for tables starting with brand_
cur.execute("""
SELECT table_name
FROM information_schema.tables
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public' AND table_name LIKE 'brand_%'
""")
tables = cur.fetchall()
@@ -434,15 +471,143 @@ def list_available_brands() -> List[str]:
table_name = table[0]
suffix = table_name[len("brand_"):].lower() if table_name.startswith("brand_") else table_name.lower()
# Prefer the original display name from the brand map if available
brand_name = brand_map.get(suffix, suffix.replace('_', ' ').title())
brands.append(brand_name)
brands.append(display_name_for_suffix(suffix, brand_map))
except Exception as e:
logger.error(f"Failed to list brands: {e}")
conn.close()
return sorted(list(set(brands)))
BRAND_OVERVIEW_TTL_SECONDS = int(os.getenv("BRAND_OVERVIEW_TTL_SECONDS", "30"))
_OVERVIEW_CACHE: Dict[str, Any] = {"at": 0.0, "data": None}
def invalidate_brand_overview_cache() -> None:
"""Drop the cached brand overview so the next read reflects a fresh write."""
_OVERVIEW_CACHE.update(at=0.0, data=None)
def get_brand_overview(force_refresh: bool = False) -> List[Dict[str, Any]]:
"""Per-brand summary rows backing the brand cards on the home page.
Everything runs on a single connection: _connect() opens a new one each
call, so doing this per brand would cost dozens of handshakes on every
page load. Results are cached for BRAND_OVERVIEW_TTL_SECONDS and dropped
explicitly by upsert_brand_products(), which covers every write path.
"""
now = time.monotonic()
if (
not force_refresh
and _OVERVIEW_CACHE["data"] is not None
and now - _OVERVIEW_CACHE["at"] < BRAND_OVERVIEW_TTL_SECONDS
):
return _OVERVIEW_CACHE["data"]
conn = _connect()
if not conn:
return []
brand_map = _build_sanitized_brand_map()
overview: List[Dict[str, Any]] = []
try:
with conn, conn.cursor() as cur:
# Brand tables predating the current DDL are missing columns until a
# write runs _ensure_columns on them, so build each query from the
# columns that actually exist. The rest of this module survives that
# by selecting *; naming columns here would 500 the whole endpoint.
cur.execute(
"""
SELECT table_name, column_name FROM information_schema.columns
WHERE table_schema = 'public' AND table_name LIKE 'brand_%'
"""
)
table_columns: Dict[str, set] = {}
for table_name, column_name in cur.fetchall():
table_columns.setdefault(table_name, set()).add(column_name)
for suffix in sorted(set(_list_brand_table_suffixes(cur))):
table_name = f"brand_{suffix}"
columns = table_columns.get(table_name, set())
has_category = "category" in columns
try:
if has_category:
cur.execute(
f"""
SELECT COUNT(*),
COUNT(DISTINCT category)
FILTER (WHERE category IS NOT NULL AND category <> '')
FROM {table_name}
"""
)
row = cur.fetchone()
product_count = int(row[0]) if row else 0
category_count = int(row[1]) if row and row[1] is not None else 0
else:
cur.execute(f"SELECT COUNT(*) FROM {table_name}")
row = cur.fetchone()
product_count = int(row[0]) if row else 0
category_count = 0
if product_count == 0:
# An empty table is a leftover, not a brand to advertise.
continue
categories: List[str] = []
if has_category:
cur.execute(
f"""
SELECT category FROM {table_name}
WHERE category IS NOT NULL AND category <> ''
GROUP BY category ORDER BY COUNT(*) DESC LIMIT 4
"""
)
categories = [r[0] for r in cur.fetchall() if r[0]]
img = None
image_cols = [c for c in ("image_id", "image_url", "image_urls") if c in columns]
if "image_url" in columns or "image_urls" in columns:
where = " OR ".join(
f"({c} IS NOT NULL)" for c in ("image_url", "image_urls") if c in columns
)
order = " ORDER BY updated_at DESC" if "updated_at" in columns else ""
cur.execute(
f"SELECT {', '.join(image_cols)} FROM {table_name} "
f"WHERE {where}{order} LIMIT 1"
)
row = cur.fetchone()
img = dict(zip(image_cols, row)) if row else None
except Exception as e:
logger.warning("Brand overview skipped %s: %s", table_name, e)
continue
sample_image_id = (img or {}).get("image_id")
sample_image_url = (img or {}).get("image_url") or None
if not sample_image_url and img and img.get("image_urls"):
urls = [u for u in img["image_urls"] if u]
sample_image_url = urls[0] if urls else None
overview.append({
"suffix": suffix,
"display_name": display_name_for_suffix(suffix, brand_map),
"product_count": product_count,
"category_count": category_count,
"categories": categories,
"sample_image_id": sample_image_id,
"sample_image_url": sample_image_url,
})
except Exception as e:
logger.error(f"Failed to build brand overview: {e}")
finally:
conn.close()
overview.sort(key=lambda b: b["display_name"].lower())
_OVERVIEW_CACHE.update(at=now, data=overview)
return overview
def get_products_by_brand(brand: str, limit: Optional[int] = None, offset: int = 0,
category: Optional[str] = None) -> List[Dict[str, Any]]:
"""Fetch products for a specific brand from its table (plain listing, no ranking).

View File

@@ -20,22 +20,17 @@ Usage:
from __future__ import annotations
import argparse
import json
import logging
import sys
from pathlib import Path
from collections import defaultdict
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app.services.vector_store import ensure_brand_schema, upsert_brand_products, _connect, _sanitize_name # noqa: E402
from app.services.brand_registry import resolve_parent_brand # noqa: E402
from app.services.brand_sync import SEED_DIR, load_seed_catalogs, seed_brands # noqa: E402
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
SEED_DIR = Path(__file__).resolve().parents[1] / "data" / "seed_catalogs"
def main() -> None:
parser = argparse.ArgumentParser(description="Seed pgvector with bundled sample catalogs")
@@ -63,34 +58,11 @@ def main() -> None:
logger.error("Seed directory not found: %s", SEED_DIR)
sys.exit(1)
files = sorted(SEED_DIR.glob("*.json"))
if args.only:
wanted = [w.lower() for w in args.only]
files = [f for f in files if any(w in f.name.lower() for w in wanted)]
if not files:
logger.warning("No matching seed files found in %s", SEED_DIR)
return
# Aggregate products by resolved (canonical) brand so that multiple
# Products are aggregated by resolved (canonical) brand so that multiple
# seed files contributing to the same brand table are upserted as a
# single batch. This ensures the stale-product cleanup (cleanup=True)
# doesn't orphan products from a sibling file.
brand_products: dict[str, list[dict]] = defaultdict(list)
file_brand_map: dict[str, str] = {}
for f in files:
data = json.loads(f.read_text(encoding="utf-8-sig"))
products = data.get("products", [])
brand = data.get("brand") or (products[0].get("brand_name") if products else None)
if not brand or not products:
logger.warning("Skipping %s - no brand/products found", f.name)
continue
resolved = resolve_parent_brand(brand)
brand_products[resolved].extend(products)
file_brand_map[f.name] = resolved
logger.info("Read %d products from %s -> resolved brand '%s'", len(products), f.name, resolved)
brand_products = load_seed_catalogs(only=args.only)
if not brand_products:
logger.warning("No valid seed data found in %s", SEED_DIR)
@@ -99,17 +71,7 @@ def main() -> None:
# Do not drop existing brand tables to preserve user database state.
# Stale table cleanup disabled per environment requirements.
total = 0
for resolved_brand, all_products in brand_products.items():
logger.info("Seeding %d product(s) for brand '%s'", len(all_products), resolved_brand)
table = ensure_brand_schema(resolved_brand)
if not table:
logger.error("Could not create/verify table for brand '%s' - is pgvector reachable?", resolved_brand)
continue
upsert_brand_products(resolved_brand, all_products, cleanup=True)
logger.info("Seeded %d products for brand '%s' (table=%s)", len(all_products), resolved_brand, table)
total += len(all_products)
total = seed_brands(brand_products, cleanup=True)
print(f"\nDone. Seeded {total} products across {len(brand_products)} brand(s) into pgvector.")
print("Start the API with `uvicorn app.main:app --reload` and try a search/chat query.")

View File

@@ -0,0 +1,113 @@
"""
Guard rails for brand -> parent-brand resolution.
`resolve_parent_brand` decides which `brand_*` table, which S3 folder, and
which seed catalog file a product lands in. An edit to BRAND_ALIASES that
quietly changes an existing brand's answer would relocate live rows, so the
first test pins the answer for every realistic input: all alias keys, all
parent values, and the `brand` field of every bundled seed catalog.
No database, no network - the registry is a plain dict.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from app.services.brand_registry import BRAND_ALIASES, resolve_parent_brand
from app.services.vector_store import _sanitize_name
SEED_DIR = Path(__file__).resolve().parents[1] / "data" / "seed_catalogs"
def _seed_brand_fields() -> list[str]:
"""The `brand` value of every seed catalog (skipping non-catalog files)."""
brands = []
for path in sorted(SEED_DIR.glob("*.json")):
try:
data = json.loads(path.read_text(encoding="utf-8-sig"))
except Exception:
continue
brand = data.get("brand")
if brand:
brands.append(brand)
return brands
# Every seed catalog and the table it must continue to feed. `tata` mapping to
# hindustan_unilever is not a typo: alias "hul tata tea" claims it, which is
# where all 121 of that file's products already live.
EXPECTED_SEED_TABLES = {
"amul": "brand_amul",
"cadbury": "brand_cadbury",
"cavinkare": "brand_cavinkare",
"coca-cola": "brand_coca_cola",
"colgate-palmolive": "brand_colgate_palmolive",
"dabur": "brand_dabur",
"godrej": "brand_godrej",
"grb": "brand_grb",
"hindustan unilever": "brand_hindustan_unilever",
"lion dates": "brand_lion_dates",
"Manna": "brand_manna",
"milky mist": "brand_milky_mist",
"naga": "brand_naga",
"Nestle": "brand_nestle",
"p&g": "brand_p_g",
"pepsico": "brand_pepsico",
"tata": "brand_hindustan_unilever",
}
@pytest.mark.parametrize("alias,parent", sorted(BRAND_ALIASES.items()))
def test_every_alias_resolves_to_its_declared_parent(alias: str, parent: str) -> None:
assert resolve_parent_brand(alias) == parent
# Every parent brand resolves to itself except "sunfeast", which the alias
# "itc sunfeast" claims for ITC. That collapse predates the word-boundary
# change and is pinned here so it stays a deliberate quirk rather than drift.
_NON_SELF_PARENTS = {"sunfeast": "itc"}
@pytest.mark.parametrize("parent", sorted(set(BRAND_ALIASES.values())))
def test_parent_brands_resolve_to_their_pinned_answer(parent: str) -> None:
assert resolve_parent_brand(parent) == _NON_SELF_PARENTS.get(parent, parent)
def test_seed_catalogs_keep_their_current_tables() -> None:
"""The load-bearing assertion: no seed file changes destination table."""
actual = {
brand: f"brand_{_sanitize_name(resolve_parent_brand(brand))}"
for brand in _seed_brand_fields()
}
assert actual == EXPECTED_SEED_TABLES
def test_known_sub_brands_still_route_to_their_family() -> None:
"""Word-boundary matching must not break legitimate sub-brand routing."""
assert resolve_parent_brand("Dove") == "hindustan unilever"
assert resolve_parent_brand("Silk") == "cadbury"
assert resolve_parent_brand("Bingo") == "itc"
assert resolve_parent_brand("Sunfeast") == "itc"
def test_new_brand_is_not_swallowed_by_a_fragment_match() -> None:
"""
"sun" appears inside "hul sunsilk" but is not a word there, so a new brand
called Sun must stay Sun. Under the old substring fallback it silently
became "hindustan unilever" - the bug this guards.
"""
assert resolve_parent_brand("Sun") == "Sun"
assert resolve_parent_brand("Bing") == "Bing"
def test_unknown_brand_passes_through_unchanged() -> None:
assert resolve_parent_brand("Totally Made Up Brand") == "Totally Made Up Brand"
def test_new_brand_round_trips_to_its_own_table() -> None:
"""ITC is the worked example: it must own brand_itc, not merge elsewhere."""
assert resolve_parent_brand("ITC") == "itc"
assert _sanitize_name(resolve_parent_brand("ITC")) == "itc"