diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..a2f4b3e --- /dev/null +++ b/.gitattributes @@ -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 diff --git a/app/api/routers/brands.py b/app/api/routers/brands.py index 895dcf2..fddeb55 100644 --- a/app/api/routers/brands.py +++ b/app/api/routers/brands.py @@ -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)) diff --git a/app/api/routers/system.py b/app/api/routers/system.py index 5b5a1bc..8df2187 100644 --- a/app/api/routers/system.py +++ b/app/api/routers/system.py @@ -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.", + } diff --git a/app/api/routers/user_products.py b/app/api/routers/user_products.py index fbf52f7..3b19017 100644 --- a/app/api/routers/user_products.py +++ b/app/api/routers/user_products.py @@ -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"))]) diff --git a/app/api/schemas.py b/app/api/schemas.py index 8cf0365..1395229 100644 --- a/app/api/schemas.py +++ b/app/api/schemas.py @@ -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] diff --git a/app/core/ingestion.py b/app/core/ingestion.py index e7bff4c..ceb1a4f 100644 --- a/app/core/ingestion.py +++ b/app/core/ingestion.py @@ -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), diff --git a/app/main.py b/app/main.py index f4c500c..e764438 100644 --- a/app/main.py +++ b/app/main.py @@ -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) diff --git a/app/services/brand_registry.py b/app/services/brand_registry.py index 2afaf77..4e16595 100644 --- a/app/services/brand_registry.py +++ b/app/services/brand_registry.py @@ -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"(? 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 diff --git a/app/services/brand_sync.py b/app/services/brand_sync.py new file mode 100644 index 0000000..3c3e84a --- /dev/null +++ b/app/services/brand_sync.py @@ -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 diff --git a/app/services/query_intent.py b/app/services/query_intent.py index 8f091bf..c04733e 100644 --- a/app/services/query_intent.py +++ b/app/services/query_intent.py @@ -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() diff --git a/app/services/vector_store.py b/app/services/vector_store.py index 68af150..070b1fd 100644 --- a/app/services/vector_store.py +++ b/app/services/vector_store.py @@ -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). diff --git a/scripts/seed_sample_data.py b/scripts/seed_sample_data.py index da0fb69..81702e6 100644 --- a/scripts/seed_sample_data.py +++ b/scripts/seed_sample_data.py @@ -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.") diff --git a/tests/test_brand_registry.py b/tests/test_brand_registry.py new file mode 100644 index 0000000..78f80a2 --- /dev/null +++ b/tests/test_brand_registry.py @@ -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"