diff --git a/app/api/routers/catalog.py b/app/api/routers/catalog.py index 50a9d2a..08ce2d8 100644 --- a/app/api/routers/catalog.py +++ b/app/api/routers/catalog.py @@ -19,7 +19,20 @@ async def _run_job(job_id: str, brand: str, max_products: int) -> None: job_store.update(job_id, "running") try: summary = await ingest_brand(brand, max_products=max_products) - job_store.update(job_id, "done", detail=f"{summary['total_products']} products ingested") + # "Ingested" has to mean "in the database". The generation stages can all + # succeed while the pgvector write fails, and reporting that as done is + # how a run that stored nothing ends up looking successful in the UI. + if summary.get("storage_error"): + job_store.update( + job_id, + "failed", + detail=( + f"Generated {summary['total_products']} product(s) but storing them " + f"failed, so none are in the catalog: {summary['storage_error']}" + ), + ) + else: + job_store.update(job_id, "done", detail=f"{summary['total_products']} products ingested") except Exception as e: # noqa: BLE001 - surface any failure to the UI logger.exception("Catalog ingestion job %s failed", job_id) job_store.update(job_id, "failed", detail=str(e)) diff --git a/app/api/routers/user_products.py b/app/api/routers/user_products.py index 57324dc..d70cb31 100644 --- a/app/api/routers/user_products.py +++ b/app/api/routers/user_products.py @@ -1,15 +1,52 @@ +""" +User workspace: add products one at a time, in a batch, or from a spreadsheet. + +Every route here has the same contract, and it is the whole point of this +module: **the response says "saved" only when the rows are in the database.** + +That used to not be true, and it produced the worst class of bug - a green +success panel over an empty catalog. Three separate things caused it: + + 1. `upsert_brand_products()` returned silently when the database was + unreachable (wrong password, USE_PGVECTOR=false, host down), so the + enrichment carried on, the seed JSON was updated, and the endpoint + answered 201. It now raises `VectorStoreUnavailable`, and these routes + translate that into a 503 that says nothing was saved. + 2. `/upload-file` returned `{"status": "success"}` regardless of outcome. A + file where every row failed came back 201 with `added_count: 0`. The + status is now derived from what actually persisted, and a file that saved + nothing is an error response, not a success one. + 3. Column headers were matched with a chain of substring tests, so two + columns could map onto the same field ("Product Name" and "Item Name" + both became `product_name`). Pandas then held duplicate columns and + `row.get(...)` returned a Series, which stringified into garbage product + names or raised "truth value of an array is ambiguous" for every row. + Mapping is now first-wins with the loser reported back to the caller. + +The seed JSON is written only after the database write is confirmed, so the two +stores cannot disagree about what exists. If the JSON write then fails, the +startup/periodic reconcile in `app.services.brand_sync` re-exports it from the +rows - the database is the source of truth, the file is the mirror. +""" +from __future__ import annotations + import io import logging import re -from typing import Any, Dict, List, Optional +from collections import OrderedDict +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Tuple + import pandas as pd from pydantic import BaseModel, Field -from fastapi import APIRouter, Depends, HTTPException, File, UploadFile +from fastapi import APIRouter, Depends, HTTPException, File, UploadFile, status +from starlette.concurrency import run_in_threadpool from app.api.deps import require_permission -from app.infrastructure.settings import SEED_CATALOG_DIR +from app.infrastructure.settings import USE_EMBEDDINGS from app.services.vector_store import ( + VectorStoreUnavailable, upsert_brand_products, resolve_parent_brand, _sanitize_name, @@ -23,6 +60,18 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/user/products", tags=["user_products"]) +# A spreadsheet this size is a mistake or an attack, not a product list. Both +# limits exist so the request fails fast with a clear message instead of pinning +# a worker for minutes and dying at the reverse proxy's timeout - which the user +# sees as "the upload did nothing", the same symptom this module is about. +MAX_UPLOAD_BYTES = 10 * 1024 * 1024 +MAX_UPLOAD_ROWS = 2000 + +# How many per-row errors to include in the response. The full count is always +# reported; this only bounds the payload. +MAX_REPORTED_ERRORS = 50 + + class AddProductRequest(BaseModel): brand: str = Field(..., description="Brand name, e.g. Lion Dates") product_name: str = Field(..., description="Product name, e.g. Lion Dates 450g") @@ -54,42 +103,327 @@ def _slugify(text: str) -> str: return re.sub(r'[^a-z0-9]+', '_', text.lower()).strip('_') -def _enrich_and_save_product(req: AddProductRequest) -> Dict[str, Any]: - brand = req.brand.strip() - brand_parent = resolve_parent_brand(brand) - brand_slug = _sanitize_name(brand_parent) +# --------------------------------------------------------------------------- +# Spreadsheet header mapping +# --------------------------------------------------------------------------- +# Headers are normalised to lowercase words ("Final Price (₹)" -> "final +# price") and then matched against the exact table first, falling back to the +# ordered keyword rules. Order is load-bearing in the fallback: "product sku" +# has to reach the sku rule before the product-name rule claims it, and "price +# range" has to reach the range rule before the price rule. +def _normalize_header(raw: Any) -> str: + text = str(raw or "").strip().lower() + text = re.sub(r"[^a-z0-9]+", " ", text) + return re.sub(r"\s+", " ", text).strip() + + +_EXACT_HEADERS: Dict[str, str] = { + # brand + "brand": "brand", "brand name": "brand", "brands": "brand", + "company": "brand", "manufacturer": "brand", "company name": "brand", + # product name + "product": "product_name", "product name": "product_name", + "product name variant": "product_name", "product variant": "product_name", + "variant": "product_name", "name": "product_name", "item": "product_name", + "item name": "product_name", "product title": "product_name", + "products": "product_name", "sku name": "product_name", + # display title (distinct from the identity used for the image_id) + "title": "title", "display title": "title", "display name": "title", + # category + "category": "category", "categories": "category", "cat": "category", + "product category": "category", "segment": "category", + # price + "price range": "price_range", "range": "price_range", "mrp range": "price_range", + "price": "final_selling_price", "final price": "final_selling_price", + "final price rs": "final_selling_price", "final selling price": "final_selling_price", + "selling price": "final_selling_price", "mrp": "final_selling_price", + "rate": "final_selling_price", "amount": "final_selling_price", + "cost": "final_selling_price", "unit price": "final_selling_price", + # identifiers + "barcode": "barcode", "barcode gtin ean": "barcode", "bar code": "barcode", + "gtin": "barcode", "ean": "barcode", "upc": "barcode", "ean13": "barcode", + "hsn": "hsn_code", "hsn code": "hsn_code", "hsn sac": "hsn_code", + "sku": "product_sku", "product sku": "product_sku", "sku code": "product_sku", + "fssai": "fssai_license", "fssai license": "fssai_license", + "fssai license number": "fssai_license", "fssai number": "fssai_license", + # text / media + "description": "description", "desc": "description", + "product description": "description", "details": "description", + "image url": "image_url", "custom image url": "image_url", "image": "image_url", + "image link": "image_url", "images": "image_url", "photo": "image_url", + "url": "image_url", "image urls": "image_url", + # list-valued + "size": "size_variants", "sizes": "size_variants", "size variants": "size_variants", + "pack size": "size_variants", "net weight": "size_variants", "weight": "size_variants", + "providers": "providers", "provider": "providers", "available at": "providers", + "platforms": "providers", "marketplaces": "providers", + "highlights": "highlights", "features": "highlights", "key features": "highlights", + "nutrients": "nutrients", "nutrition": "nutrients", "nutrition facts": "nutrients", +} + +_KEYWORD_RULES: Tuple[Tuple[str, Tuple[str, ...]], ...] = ( + ("fssai_license", ("fssai",)), + ("hsn_code", ("hsn",)), + ("barcode", ("barcode", "bar code", "gtin", "ean", "upc")), + ("product_sku", ("sku",)), + ("price_range", ("price range", "range")), + ("final_selling_price", ("final price", "selling price", "price", "mrp", "rate", "cost")), + ("image_url", ("image", "photo", "picture", "url", "link")), + ("description", ("description", "desc", "detail")), + ("category", ("category", "segment")), + ("brand", ("brand", "manufacturer", "company")), + ("size_variants", ("size", "pack", "weight", "volume", "net qty", "quantity")), + ("providers", ("provider", "platform", "marketplace", "available at")), + ("highlights", ("highlight", "feature", "benefit")), + ("nutrients", ("nutrient", "nutrition")), + ("title", ("title",)), + ("product_name", ("product", "item", "variant", "name")), +) + +# Cells that mean "blank". Spreadsheets are full of these, and a literal "nan" +# is what an empty pandas cell becomes once it has been through str(). +_BLANK_VALUES = frozenset({"", "nan", "none", "null", "na", "n/a", "-", "--", "#n/a"}) + + +def _canonical_field(normalized: str) -> Optional[str]: + exact = _EXACT_HEADERS.get(normalized) + if exact: + return exact + for canonical, keywords in _KEYWORD_RULES: + if any(keyword in normalized for keyword in keywords): + return canonical + return None + + +@dataclass +class _ColumnMapping: + """Which spreadsheet column feeds which product field.""" + columns: Dict[str, Any] = field(default_factory=dict) # field -> original column label + ignored: List[Dict[str, str]] = field(default_factory=list) # collisions, reported back + unrecognised: List[str] = field(default_factory=list) + + +def map_spreadsheet_columns(columns) -> _ColumnMapping: + """Resolve column labels to product fields, first occurrence wins. + + First-wins rather than last-wins because a spreadsheet's leftmost columns + are its primary ones; and *dropping* the loser rather than renaming it is + what keeps the DataFrame free of duplicate column labels. With duplicates, + `row[label]` yields a Series instead of a value, which is how a row of real + data used to turn into the string "0 Lion Dates 450g\\n1 ...". + """ + mapping = _ColumnMapping() + for column in columns: + normalized = _normalize_header(column) + if not normalized: + continue + canonical = _canonical_field(normalized) + if canonical is None: + mapping.unrecognised.append(str(column)) + continue + if canonical in mapping.columns: + mapping.ignored.append({ + "column": str(column), + "field": canonical, + "using_instead": str(mapping.columns[canonical]), + }) + continue + mapping.columns[canonical] = column + return mapping + + +def _text(row: Dict[str, Any], mapping: _ColumnMapping, field_name: str) -> Optional[str]: + """One cell as clean text, or None when the cell is blank in any of the + dozen ways a spreadsheet can be blank.""" + column = mapping.columns.get(field_name) + if column is None: + return None + value = row.get(column) + if value is None: + return None + try: + if pd.isna(value): + return None + except (TypeError, ValueError): # arrays/lists - not blank, keep going + pass + text = str(value).strip() + if text.lower() in _BLANK_VALUES: + return None + # Excel hands back numeric-looking identifiers as floats, so a barcode + # arrives as "8906012345001.0" and an HSN code as "2008.0". Both are then + # stored and displayed wrong. + if re.fullmatch(r"-?\d+\.0+", text): + text = text.split(".")[0] + return text + + +def _number(row: Dict[str, Any], mapping: _ColumnMapping, field_name: str) -> Optional[float]: + text = _text(row, mapping, field_name) + if text is None: + return None + # Strip currency symbols, thousands separators and stray units: "₹1,850.00" + cleaned = re.sub(r"[^0-9.\-]", "", text) + if cleaned in ("", "-", ".", "-."): + return None + try: + return float(cleaned) + except ValueError: + return None + + +def _string_list(row: Dict[str, Any], mapping: _ColumnMapping, field_name: str) -> List[str]: + text = _text(row, mapping, field_name) + if not text: + return [] + return [part.strip() for part in re.split(r"[,;|]", text) if part.strip()] + + +def row_to_request(row: Dict[str, Any], mapping: _ColumnMapping) -> AddProductRequest: + """Build a validated request from one spreadsheet row. + + Raises ValueError with a message naming the missing field, so the response + can point the user at a row rather than at the file. + """ + brand = _text(row, mapping, "brand") + product_name = _text(row, mapping, "product_name") or _text(row, mapping, "title") + if not brand: + raise ValueError("no brand name in this row") + if not product_name: + raise ValueError("no product name in this row") + if not _slugify(product_name): + raise ValueError(f"product name {product_name!r} has no letters or digits to identify it by") + + return AddProductRequest( + brand=brand, + product_name=product_name, + title=_text(row, mapping, "title"), + category=_text(row, mapping, "category"), + description=_text(row, mapping, "description"), + price_range=_text(row, mapping, "price_range"), + size_variants=_string_list(row, mapping, "size_variants"), + providers=_string_list(row, mapping, "providers"), + highlights=_string_list(row, mapping, "highlights"), + nutrients=_string_list(row, mapping, "nutrients"), + fssai_license=_text(row, mapping, "fssai_license"), + product_sku=_text(row, mapping, "product_sku"), + hsn_code=_text(row, mapping, "hsn_code"), + final_selling_price=_number(row, mapping, "final_selling_price"), + barcode=_text(row, mapping, "barcode"), + image_url=_text(row, mapping, "image_url"), + ) + + +def read_products_dataframe(filename: str, content: bytes) -> pd.DataFrame: + """Parse an uploaded CSV/TSV/Excel file into a DataFrame of strings. + + Everything is read as text on purpose. Left to infer types, pandas turns a + 13-digit barcode into a float ("8.906012345001e+12"), drops the leading + zero from HSN code "0402", and renders 185 as "185.0" - all silently, all + stored exactly that way. + """ + lowered = filename.lower() + try: + if lowered.endswith((".xlsx", ".xlsm", ".xls")): + return pd.read_excel(io.BytesIO(content), dtype=str) + if lowered.endswith((".tsv", ".tab")): + return pd.read_csv(io.BytesIO(content), sep="\t", dtype=str) + if lowered.endswith((".csv", ".txt")): + try: + return pd.read_csv(io.BytesIO(content), dtype=str) + except pd.errors.ParserError: + # Sniff the delimiter for semicolon/pipe exports, which are what + # a European Excel locale produces from "Save as CSV". + return pd.read_csv(io.BytesIO(content), sep=None, engine="python", dtype=str) + except ImportError as exc: + # openpyxl/xlrd are optional pandas extras. Missing them is a deployment + # problem, and the generic "could not parse" it used to produce sent + # people looking at their spreadsheet instead. + raise HTTPException( + status_code=status.HTTP_501_NOT_IMPLEMENTED, + detail=( + f"This server cannot read '{filename}' because an Excel reader is not " + f"installed ({exc}). Save the file as .csv and upload that, or install " + f"the missing package on the server (openpyxl for .xlsx, xlrd for .xls)." + ), + ) from exc + except Exception as exc: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Could not read '{filename}' as a spreadsheet: {exc}", + ) from exc + + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=( + f"Unsupported file type '{filename}'. Upload a .csv, .tsv, .xlsx or .xls file." + ), + ) + + +# --------------------------------------------------------------------------- +# Enrichment + persistence +# --------------------------------------------------------------------------- +@dataclass +class _PersistOutcome: + saved: List[Dict[str, Any]] = field(default_factory=list) + failures: List[Dict[str, Any]] = field(default_factory=list) + brands: "OrderedDict[str, int]" = field(default_factory=OrderedDict) + catalog_files: List[str] = field(default_factory=list) + warnings: List[str] = field(default_factory=list) + # Set once the database turns out to be unreachable. Every remaining brand + # is then failed against this message instead of re-attempting a connection + # that costs DB_CONNECT_TIMEOUT_SECONDS each time. + unavailable: Optional[str] = None + + +def _brand_sample(brand_parent: str) -> Dict[str, Any]: + """The most recently updated product for a brand, used to inherit defaults. + + `limit=1` matters: this used to be a full `SELECT *` of the brand table, + executed once per uploaded row. A 200-row file against a brand with a few + thousand products meant 200 full table reads before a single insert. + """ + try: + existing = get_products_by_brand(brand_parent, limit=1) + except Exception as exc: # noqa: BLE001 - defaults are a nicety, not the write + logger.warning("Could not read existing products for '%s': %s", brand_parent, exc) + return {} + return existing[0] if existing else {} + + +def _build_product_dict(req: AddProductRequest, brand_parent: str, + sample_existing: Dict[str, Any]) -> Dict[str, Any]: + """Fill in everything the catalog needs that the user did not supply. + + Pure apart from the optional S3 image lookup - no database access and no + embedding call, so a batch can do those once instead of once per row. + """ + brand_slug = _sanitize_name(brand_parent) product_name = req.product_name.strip() product_slug = _slugify(product_name) + if not product_slug: + raise ValueError(f"product name {product_name!r} has no letters or digits to identify it by") image_id = f"{brand_slug}_{product_slug}" - # Check existing brand products for fallback attributes (e.g. fssai_license, category, provider_examples) - existing_db = get_products_by_brand(brand_parent) - sample_existing = existing_db[0] if existing_db else {} - - # Category fallback category = req.category or sample_existing.get("category") or "Health Foods" - - # FSSAI License fallback fssai_license = req.fssai_license or sample_existing.get("fssai_license") or "10012042000244" - # Description fallback description = req.description or ( f"Introducing {product_name} from the trusted {brand_parent} brand. " f"A premium quality product offering superior taste, authentic ingredients, and reliable value. " f"Backed by {brand_parent}'s reputation for quality and consistency." ) - # Size variants fallback - size_variants = req.size_variants + size_variants = list(req.size_variants) if not size_variants: match = re.search(r'\d+\s*(?:g|kg|ml|l|pack)\b', product_name, re.I) if match: size_variants = [match.group(0)] else: - size_variants = [sample_existing.get("size_variants", ["Default"])[0]] if sample_existing.get("size_variants") else ["Standard"] + existing_sizes = sample_existing.get("size_variants") or [] + size_variants = [existing_sizes[0]] if existing_sizes else ["Standard"] - # Price range fallback price_range = req.price_range if not price_range: if req.final_selling_price: @@ -103,7 +437,7 @@ def _enrich_and_save_product(req: AddProductRequest) -> Dict[str, Any]: highlights = req.highlights or list(sample_existing.get("highlights") or ["100% Quality Assurance", "Authentic Brand Product"]) nutrients = req.nutrients or list(sample_existing.get("nutrients") or ["Energy - High", "Protein - Good Source"]) - # Image URL Resolution (S3 or web search fallback) + # Image URL Resolution (S3 or brand-sample fallback) final_image_urls = list(req.image_urls) if req.image_url and req.image_url not in final_image_urls: final_image_urls.insert(0, req.image_url) @@ -111,11 +445,15 @@ def _enrich_and_save_product(req: AddProductRequest) -> Dict[str, Any]: if not final_image_urls: # 1. Try S3 service if enabled if s3_service.enabled: - s3_urls = s3_service.get_product_image_urls(brand_parent, image_id) + try: + s3_urls = s3_service.get_product_image_urls(brand_parent, image_id) + except Exception as exc: # noqa: BLE001 - an image is not worth the row + logger.warning("S3 image lookup failed for '%s': %s", image_id, exc) + s3_urls = None if s3_urls: - final_image_urls = s3_urls + final_image_urls = list(s3_urls) - # 2. Inherit from brand sample or S3 formatted default URL + # 2. Inherit from brand sample if not final_image_urls and sample_existing.get("image_urls"): final_image_urls = list(sample_existing.get("image_urls")) @@ -125,16 +463,9 @@ def _enrich_and_save_product(req: AddProductRequest) -> Dict[str, Any]: final_image_urls = [canonical_s3] primary_image_url = final_image_urls[0] if final_image_urls else None - - # Vector embedding creation search_text = f"{brand_parent} {product_name} {category} {description} {price_range}" - try: - embedding = embed_texts([search_text])[0] - except Exception as e: - logger.warning("Embedding generation failed for '%s': %s", product_name, e) - embedding = None - product_dict = { + return { "image_id": image_id, "product_name": product_name, "title": req.title or product_name, @@ -158,45 +489,287 @@ def _enrich_and_save_product(req: AddProductRequest) -> Dict[str, Any]: "image_url": primary_image_url, "image_urls": final_image_urls, "search_query": search_text, - "embedding": embedding, + "embedding": None, } - # 1. Update PostgreSQL Database Table - upsert_brand_products(brand_parent, [product_dict]) - logger.info("✅ Upserted '%s' into PostgreSQL table for brand '%s'", product_name, brand_parent) - # 2. Update JSON Seed File - _update_json_catalog_file(brand_parent, product_dict) +def _attach_embeddings(product_dicts: List[Dict[str, Any]]) -> None: + """Embed every product in one call. - return product_dict - - -def _update_json_catalog_file(brand: str, product_dict: Dict[str, Any]) -> None: - """Append/update one product in the brand's seed catalog. - - 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. + One call per upload rather than per row: loading the sentence-transformer + is the expensive part, and `embed_texts` already batches internally. A + failure here leaves embeddings None - the product is still stored and + browsable, it just does not participate in semantic search until the next + re-embed, which is a far better outcome than losing the row. """ - upsert_products_into_catalog_file(brand, [product_dict]) + if not product_dicts: + return + if not USE_EMBEDDINGS: + # Honour the flag rather than loading a sentence-transformer anyway. The + # products still store and browse; only semantic search needs the vector. + logger.info("USE_EMBEDDINGS=false - storing %d product(s) without embeddings", + len(product_dicts)) + return + try: + vectors = embed_texts([d["search_query"] for d in product_dicts]) + except Exception as exc: # noqa: BLE001 + logger.warning("Embedding generation failed for %d product(s): %s", len(product_dicts), exc) + return + for product_dict, vector in zip(product_dicts, vectors): + product_dict["embedding"] = vector +def _persist_products(items: List[Tuple[Optional[int], AddProductRequest]]) -> _PersistOutcome: + """Enrich and store products, grouped by brand, reporting per-row outcomes. + + The order is deliberate: database first, and the seed JSON only for what the + database confirmed. Written the other way round (which is how this used to + work), a failed database write still left the product in the JSON file, so + the catalog file and the catalog disagreed and the endpoint had something to + call a success. + """ + outcome = _PersistOutcome() + if not items: + return outcome + + groups: "OrderedDict[str, List[Tuple[Optional[int], AddProductRequest]]]" = OrderedDict() + for row_number, req in items: + groups.setdefault(resolve_parent_brand(req.brand.strip()), []).append((row_number, req)) + + # Build every product dict first: one brand-sample query per brand (not per + # row) and one embedding call for the whole upload. + built: "OrderedDict[str, List[Tuple[Optional[int], AddProductRequest, Dict[str, Any]]]]" = OrderedDict() + for brand_parent, rows in groups.items(): + sample = _brand_sample(brand_parent) + prepared = [] + for row_number, req in rows: + try: + prepared.append((row_number, req, _build_product_dict(req, brand_parent, sample))) + except Exception as exc: # noqa: BLE001 - one bad row, not the file + outcome.failures.append({ + "row": row_number, + "product_name": req.product_name, + "error": str(exc), + }) + if prepared: + built[brand_parent] = prepared + + _attach_embeddings([d for prepared in built.values() for (_, _, d) in prepared]) + + for brand_parent, prepared in built.items(): + if outcome.unavailable: + outcome.failures.extend( + {"row": row_number, "product_name": req.product_name, "error": outcome.unavailable} + for row_number, req, _ in prepared + ) + continue + + product_dicts = [d for _, _, d in prepared] + saved: List[Dict[str, Any]] = [] + try: + upsert_brand_products(brand_parent, product_dicts) + saved = product_dicts + except VectorStoreUnavailable as exc: + outcome.unavailable = str(exc) + outcome.failures.extend( + {"row": row_number, "product_name": req.product_name, "error": str(exc)} + for row_number, req, _ in prepared + ) + continue + except Exception as exc: # noqa: BLE001 + # The batch failed for a data reason - one row with a value the + # column will not take. Retry singly so the other rows still land + # and the response can name the row that is actually broken. + logger.warning("Batch upsert for '%s' failed (%s); retrying rows individually", + brand_parent, exc) + for row_number, req, product_dict in prepared: + if outcome.unavailable: + outcome.failures.append({ + "row": row_number, "product_name": req.product_name, + "error": outcome.unavailable, + }) + continue + try: + upsert_brand_products(brand_parent, [product_dict]) + saved.append(product_dict) + except VectorStoreUnavailable as row_exc: + outcome.unavailable = str(row_exc) + outcome.failures.append({ + "row": row_number, "product_name": req.product_name, "error": str(row_exc), + }) + except Exception as row_exc: # noqa: BLE001 + outcome.failures.append({ + "row": row_number, "product_name": req.product_name, "error": str(row_exc), + }) + + if not saved: + continue + + outcome.brands[brand_parent] = len(saved) + outcome.saved.extend(saved) + + # Mirror into the seed catalog once per brand. A failure is a warning, + # not a row failure: the rows are committed, and the reconcile sweep in + # brand_sync re-exports the file from them. + try: + path = upsert_products_into_catalog_file(brand_parent, saved) + if path: + outcome.catalog_files.append(path.name) + except Exception as exc: # noqa: BLE001 + logger.warning("Seed catalog update failed for '%s': %s", brand_parent, exc) + outcome.warnings.append( + f"Saved {len(saved)} product(s) for '{brand_parent}' to the database, but the " + f"seed catalog file could not be updated ({exc}). It will be re-synced " + f"automatically by the brand reconcile." + ) + + return outcome + + +def _enrich_and_save_product(req: AddProductRequest) -> Dict[str, Any]: + """Store one product and return it. Raises if it was not stored.""" + outcome = _persist_products([(None, req)]) + if outcome.saved: + return outcome.saved[0] + if outcome.unavailable: + raise VectorStoreUnavailable(outcome.unavailable) + reason = outcome.failures[0]["error"] if outcome.failures else "unknown error" + raise RuntimeError(reason) + + +def _strip_embedding(product_dict: Dict[str, Any]) -> Dict[str, Any]: + return {k: v for k, v in product_dict.items() if k != "embedding"} + + +def _batch_response(outcome: _PersistOutcome, *, filename: Optional[str] = None, + rows_total: Optional[int] = None, + skipped_blank_rows: int = 0, + mapping: Optional[_ColumnMapping] = None) -> Dict[str, Any]: + """Build a response whose `status` is derived from what actually persisted.""" + added_count = len(outcome.saved) + error_count = len(outcome.failures) + + if added_count and not error_count: + outcome_status = "success" + message = ( + f"Saved {added_count} product(s) to the database" + + (f" across {len(outcome.brands)} brands" if len(outcome.brands) > 1 else "") + + (f" and updated {', '.join(outcome.catalog_files)}" if outcome.catalog_files else "") + + "." + ) + else: + outcome_status = "partial" + message = ( + f"Saved {added_count} of {added_count + error_count} product(s). " + f"{error_count} failed - see the errors list." + ) + + body: Dict[str, Any] = { + "status": outcome_status, + "message": message, + "added_count": added_count, + "error_count": error_count, + "brands": dict(outcome.brands), + "catalog_files": outcome.catalog_files, + "warnings": list(outcome.warnings), + "added_products": [_strip_embedding(p) for p in outcome.saved], + "errors": outcome.failures[:MAX_REPORTED_ERRORS], + "errors_truncated": max(0, error_count - MAX_REPORTED_ERRORS), + } + + if filename is not None: + body["filename"] = filename + if rows_total is not None: + body["rows_total"] = rows_total + body["total_rows_processed"] = added_count + error_count + body["skipped_blank_rows"] = skipped_blank_rows + if mapping is not None: + body["recognised_columns"] = {f: str(c) for f, c in mapping.columns.items()} + body["ignored_columns"] = mapping.ignored + body["unrecognised_columns"] = mapping.unrecognised + if mapping.ignored: + body["warnings"].append( + "These columns map to a field another column already filled, so they were " + "ignored: " + ", ".join( + f"'{i['column']}' (using '{i['using_instead']}' for {i['field']})" + for i in mapping.ignored + ) + ) + return body + + +def _nothing_saved_error(outcome: _PersistOutcome, body: Dict[str, Any]) -> HTTPException: + """Turn a zero-row outcome into an honest HTTP error. + + 503 when the database is the problem so the caller knows to retry rather + than to edit the file; 422 when the file is the problem. + """ + if outcome.unavailable: + return HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=( + f"Nothing was saved - the product database is not reachable. {outcome.unavailable}" + ), + ) + + first_errors = "; ".join( + f"row {f['row']}: {f['error']}" if f.get("row") else str(f["error"]) + for f in outcome.failures[:5] + ) + return HTTPException( + status_code=422, # Unprocessable Content + detail=( + f"Nothing was saved - all {body['error_count']} product(s) failed. {first_errors}" + + (" ..." if body["error_count"] > 5 else "") + ), + ) + + +# --------------------------------------------------------------------------- +# Routes +# --------------------------------------------------------------------------- @router.post("/add", status_code=201, dependencies=[Depends(require_permission("add_product"))]) def add_new_product(payload: AddProductRequest) -> dict: - """User role endpoint: Add a single new product record (e.g. Lion Dates 450g). - Automatically enriches details, fetches images, updates PostgreSQL DB, - and updates JSON seed catalog files.""" + """Add a single product (e.g. Lion Dates 450g). + + Enriches the missing details, resolves images, writes the row to the brand's + PostgreSQL table, and mirrors it into the brand's JSON seed catalog. Returns + 201 only when the row is confirmed present in the database. + """ + if not payload.brand.strip(): + raise HTTPException(status_code=400, detail="A brand name is required.") + if not _slugify(payload.product_name.strip()): + raise HTTPException( + status_code=400, + detail=( + f"Product name {payload.product_name!r} has no letters or digits, so there is " + f"nothing to identify the product by." + ), + ) + try: res = _enrich_and_save_product(payload) - return { - "status": "success", - "message": f"Successfully added '{payload.product_name}' under brand '{payload.brand}' to database and JSON catalog.", - "product": {k: v for k, v in res.items() if k != "embedding"}, - } - except Exception as e: + except VectorStoreUnavailable as exc: + logger.error("Product '%s' NOT saved - database unavailable: %s", payload.product_name, exc) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=f"'{payload.product_name}' was not saved. {exc}", + ) from exc + except Exception as exc: # noqa: BLE001 logger.exception("Failed to add product '%s'", payload.product_name) - raise HTTPException(status_code=500, detail=f"Failed to add product: {e}") + raise HTTPException( + status_code=500, + detail=f"'{payload.product_name}' was not saved: {exc}", + ) from exc + + return { + "status": "success", + "message": ( + f"Saved '{payload.product_name}' under brand '{res['brand']}' to the database " + f"and JSON catalog." + ), + "product": _strip_embedding(res), + } @router.post( @@ -205,23 +778,105 @@ def add_new_product(payload: AddProductRequest) -> dict: dependencies=[Depends(require_permission("upload_batch_products"))], ) def batch_add_products(payload: BatchAddProductsRequest) -> dict: - """User role endpoint: Batch upload multiple product records at once.""" - added = [] - errors = [] - for req in payload.products: - try: - res = _enrich_and_save_product(req) - added.append({k: v for k, v in res.items() if k != "embedding"}) - except Exception as e: - errors.append({"product_name": req.product_name, "error": str(e)}) + """Add many products in one request.""" + if not payload.products: + raise HTTPException(status_code=400, detail="No products supplied.") - return { - "status": "success", - "added_count": len(added), - "error_count": len(errors), - "added_products": added, - "errors": errors, - } + outcome = _persist_products([(i + 1, req) for i, req in enumerate(payload.products)]) + body = _batch_response(outcome) + if not outcome.saved: + raise _nothing_saved_error(outcome, body) + return body + + +def _process_upload(filename: str, content: bytes) -> Dict[str, Any]: + """Parse, enrich and store an uploaded spreadsheet. Blocking on purpose. + + Kept out of the coroutine and handed to a worker thread by the route below. + Parsing with pandas, loading the embedding model and talking to Postgres are + all synchronous and can take tens of seconds for a large file; run inline in + an `async def` they block the event loop, which stalls every other request - + including /api/health, which is what the container healthcheck reads. A + restart mid-import then looks exactly like the bug this module is about. + """ + df = read_products_dataframe(filename, content) + + if df.empty: + raise HTTPException( + status_code=400, + detail=f"'{filename}' has a header row but no data rows.", + ) + if len(df) > MAX_UPLOAD_ROWS: + raise HTTPException( + status_code=413, # Content Too Large (starlette renamed the constant; the code is stable) + detail=( + f"'{filename}' has {len(df)} rows, over the {MAX_UPLOAD_ROWS}-row limit " + f"for a single upload. Split it into smaller files." + ), + ) + + mapping = map_spreadsheet_columns(df.columns) + if "brand" not in mapping.columns or "product_name" not in mapping.columns: + recognised = ", ".join(sorted(mapping.columns)) or "none" + raise HTTPException( + status_code=400, + detail=( + f"'{filename}' must have a brand column and a product name column. " + f"Found these column headers: {', '.join(str(c) for c in df.columns)}. " + f"Recognised fields: {recognised}. Download the sample CSV for the " + f"expected headers." + ), + ) + + items: List[Tuple[Optional[int], AddProductRequest]] = [] + parse_failures: List[Dict[str, Any]] = [] + skipped_blank_rows = 0 + + for position, (_, raw_row) in enumerate(df.iterrows()): + row = raw_row.to_dict() + # +2: one for the header line, one because humans count from 1. The + # number in an error has to match the row number in the user's editor. + row_number = position + 2 + + if all(_text(row, mapping, f) is None for f in mapping.columns): + skipped_blank_rows += 1 + continue + + try: + items.append((row_number, row_to_request(row, mapping))) + except Exception as exc: # noqa: BLE001 + parse_failures.append({ + "row": row_number, + "product_name": _text(row, mapping, "product_name") or "", + "error": str(exc), + }) + + outcome = _persist_products(items) + # Parse failures come first so the response lists rows in file order. + outcome.failures = parse_failures + outcome.failures + + body = _batch_response( + outcome, + filename=filename, + rows_total=int(len(df)), + skipped_blank_rows=skipped_blank_rows, + mapping=mapping, + ) + + if not outcome.saved: + if not outcome.failures: + raise HTTPException( + status_code=422, + detail=( + f"Nothing was saved - all {skipped_blank_rows} row(s) in '{filename}' were " + f"blank in every recognised column." + ), + ) + raise _nothing_saved_error(outcome, body) + + logger.info("Upload '%s': saved %d, failed %d, skipped %d blank row(s)", + filename, body["added_count"], body["error_count"], skipped_blank_rows) + return body @router.post( @@ -230,92 +885,25 @@ def batch_add_products(payload: BatchAddProductsRequest) -> dict: dependencies=[Depends(require_permission("upload_batch_products"))], ) async def upload_products_file(file: UploadFile = File(...)) -> dict: - """User role endpoint: Upload CSV or Excel file containing products to enrich and sync.""" - filename = file.filename or "" + """Upload a CSV/Excel product spreadsheet. + + Answers 201 only when at least one row is confirmed stored in the database; + a file that stored nothing is a 422 (bad rows) or 503 (database down), and + a partial import comes back 201 with `status: "partial"` and a per-row + error list. See the module docstring for why that matters. + """ + filename = file.filename or "upload" content = await file.read() - try: - if filename.endswith(".csv"): - df = pd.read_csv(io.BytesIO(content)) - elif filename.endswith((".xlsx", ".xls")): - df = pd.read_excel(io.BytesIO(content)) - else: - raise HTTPException(status_code=400, detail="Unsupported file format. Please upload a .csv or .xlsx file.") - except Exception as e: - raise HTTPException(status_code=400, detail=f"Failed to parse file '{filename}': {e}") - - # Standardize column headers - col_map = {} - for col in df.columns: - c_clean = str(col).strip().lower() - if "brand" in c_clean: - col_map[col] = "brand" - elif "product" in c_clean or "variant" in c_clean or "name" in c_clean: - col_map[col] = "product_name" - elif "category" in c_clean: - col_map[col] = "category" - elif "range" in c_clean: - col_map[col] = "price_range" - elif "price" in c_clean or "selling" in c_clean or "cost" in c_clean: - col_map[col] = "final_selling_price" - elif "barcode" in c_clean or "gtin" in c_clean or "ean" in c_clean: - col_map[col] = "barcode" - elif "hsn" in c_clean: - col_map[col] = "hsn_code" - elif "description" in c_clean: - col_map[col] = "description" - elif "image" in c_clean or "url" in c_clean: - col_map[col] = "image_url" - - df = df.rename(columns=col_map) - - if "brand" not in df.columns or "product_name" not in df.columns: + if not content: + raise HTTPException(status_code=400, detail=f"'{filename}' is empty.") + if len(content) > MAX_UPLOAD_BYTES: raise HTTPException( - status_code=400, - detail="File must contain at least 'Brand Name' and 'Product Name' columns.", + status_code=413, # Content Too Large (starlette renamed the constant; the code is stable) + detail=( + f"'{filename}' is {len(content) / 1_048_576:.1f} MB, over the " + f"{MAX_UPLOAD_BYTES // 1_048_576} MB limit. Split it into smaller files." + ), ) - added = [] - errors = [] - - for idx, row in df.iterrows(): - b_val = str(row.get("brand") or "").strip() - p_val = str(row.get("product_name") or "").strip() - if not b_val or not p_val or b_val.lower() == "nan" or p_val.lower() == "nan": - continue - - try: - fps_raw = row.get("final_selling_price") - fps = None - if pd.notna(fps_raw): - try: - fps = float(fps_raw) - except Exception: - pass - - req = AddProductRequest( - brand=b_val, - product_name=p_val, - category=str(row.get("category")) if pd.notna(row.get("category")) else None, - price_range=str(row.get("price_range")) if pd.notna(row.get("price_range")) else None, - final_selling_price=fps, - barcode=str(row.get("barcode")) if pd.notna(row.get("barcode")) else None, - hsn_code=str(row.get("hsn_code")) if pd.notna(row.get("hsn_code")) else None, - description=str(row.get("description")) if pd.notna(row.get("description")) else None, - image_url=str(row.get("image_url")) if pd.notna(row.get("image_url")) else None, - ) - - res = _enrich_and_save_product(req) - added.append({k: v for k, v in res.items() if k != "embedding"}) - except Exception as e: - errors.append({"row": idx + 1, "product_name": p_val, "error": str(e)}) - - return { - "status": "success", - "filename": filename, - "total_rows_processed": len(added) + len(errors), - "added_count": len(added), - "error_count": len(errors), - "added_products": added, - "errors": errors, - } + return await run_in_threadpool(_process_upload, filename, content) diff --git a/app/core/catalog_engine.py b/app/core/catalog_engine.py index fd5fed4..eaf8928 100644 --- a/app/core/catalog_engine.py +++ b/app/core/catalog_engine.py @@ -940,11 +940,17 @@ class ProductCatalogEngine: if 'image_id' not in p: id_source = p.get('product_name') or p.get('title', 'unknown_product') p['image_id'] = s3_service.generate_image_id(id_source) - upsert_brand_products(brand, enhanced_products, cleanup=True) - logger.info("🧠 Stored embeddings to pgvector") + stored = upsert_brand_products(brand, enhanced_products, cleanup=True) + logger.info("🧠 Stored %s product(s) with embeddings to pgvector", stored) except Exception as e: - logger.warning(f"Vector storage skipped/failed: {e}") - + # Stage 4 is where the catalog becomes readable by the app - every + # API read goes to pgvector, not to the dict returned here. So a + # failure at this stage means the run produced nothing the user can + # see, and it has to travel back to the job status rather than being + # logged and forgotten. + logger.error("❌ Stage 4 (pgvector storage) failed for '%s': %s", brand, e) + catalog['storage_error'] = str(e) + return catalog def save_catalog(self, catalog: Dict[str, Any], filename: str = None) -> str: diff --git a/app/core/ingestion.py b/app/core/ingestion.py index ceb1a4f..d8ea958 100644 --- a/app/core/ingestion.py +++ b/app/core/ingestion.py @@ -40,11 +40,17 @@ async def ingest_brand(brand: str, max_products: int = 50) -> Dict[str, Any]: # 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) + storage_error = catalog.get("storage_error") + + # Nothing reached the database, so there is nothing to mirror out of it - + # and running the export anyway would either write an empty file or leave a + # stale one looking current. + if not storage_error: + 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, @@ -52,6 +58,7 @@ async def ingest_brand(brand: str, max_products: int = 50) -> Dict[str, Any]: "total_images": catalog.get("total_images", 0), "duration_seconds": round(duration, 2), "engine_info": catalog.get("engine_info", {}), + "storage_error": storage_error, } logger.info("Finished ingestion for brand=%s in %.2fs: %s products", brand, duration, summary["total_products"]) diff --git a/app/services/vector_store.py b/app/services/vector_store.py index a68047b..952ea64 100644 --- a/app/services/vector_store.py +++ b/app/services/vector_store.py @@ -22,6 +22,28 @@ from app.services.s3_service import s3_service logger = logging.getLogger(__name__) +class VectorStoreUnavailable(RuntimeError): + """The pgvector database could not be reached, so a write did not happen. + + Exists because the silent alternative was a production bug that took a long + time to see: upsert_brand_products() used to `return` when _connect() gave + back None - unreachable host, wrong DB_PASSWORD, USE_PGVECTOR=false - and + every caller read that as a successful write. The upload endpoints then + answered "success", the seed JSON was updated, and not one row existed in + the database. A write that cannot happen has to raise. + """ + + +class VectorStoreWriteFailed(RuntimeError): + """The INSERT ran without error but the rows are not in the table. + + Guards against the failure modes an exception cannot catch: a statement + silently rolled back, a trigger swallowing the row, or an ON CONFLICT + target that quietly matched nothing. The only trustworthy proof of a write + is reading it back. + """ + + def _sanitize_name(name: str) -> str: """Sanitize a brand name for use as a PostgreSQL table name suffix. @@ -193,22 +215,44 @@ def ensure_brand_schema(brand: str) -> str: return table_name -def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: bool = False) -> None: +def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: bool = False) -> int: """Insert products into brand-specific table - simplified with only essential fields When `cleanup=True`, any products in the table whose image_id is NOT in the provided `products` list are deleted after the upsert. This ensures the database exactly reflects the source data. The caller is responsible for providing the complete set of products for the brand when using cleanup. + + Returns the number of distinct image_ids confirmed present in the table + afterwards - the rows are read back, so a non-raising call is proof of + persistence rather than proof that a statement was merely sent. + + Raises: + VectorStoreUnavailable: the database is unreachable; nothing was written. + VectorStoreWriteFailed: the statements ran but the rows are not there. + ValueError: a product carries no image_id, which is the primary key + every other product is deduplicated on. """ + if not products: + return 0 + conn = _connect() if not conn: - return - + raise VectorStoreUnavailable( + f"Cannot save products for '{brand}': the product database is unreachable " + f"(USE_PGVECTOR={USE_PGVECTOR}, host={DB_HOST}:{DB_PORT}, db={DB_NAME}). " + f"Nothing was saved. Check DB_HOST/DB_USER/DB_PASSWORD in backend/.env " + f"and that Postgres is accepting connections." + ) + table_name = ensure_brand_schema(brand) if not table_name: - return - + conn.close() + raise VectorStoreUnavailable( + f"Cannot save products for '{brand}': the brand table could not be created " + f"or verified in database '{DB_NAME}'. Nothing was saved." + ) + rows = [] for p in products: # Extract only essential fields @@ -333,55 +377,89 @@ def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: b image_ids = [r[4] for r in rows if r[4]] - with conn, conn.cursor() as cur: - cur.executemany( - f""" - INSERT INTO {table_name} - (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, embedding) - VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) - ON CONFLICT (image_id) DO UPDATE SET - product_name = EXCLUDED.product_name, - title = EXCLUDED.title, - description = EXCLUDED.description, - category = EXCLUDED.category, - image_url = EXCLUDED.image_url, - image_urls = EXCLUDED.image_urls, - price_range = EXCLUDED.price_range, - size_variants = EXCLUDED.size_variants, - providers = EXCLUDED.providers, - fssai_license = EXCLUDED.fssai_license, - product_sku = EXCLUDED.product_sku, - sku_source = EXCLUDED.sku_source, - hsn_code = EXCLUDED.hsn_code, - final_selling_price = EXCLUDED.final_selling_price, - selling_price = EXCLUDED.selling_price, - barcode = EXCLUDED.barcode, - barcode_type = EXCLUDED.barcode_type, - highlights = EXCLUDED.highlights, - nutrients = EXCLUDED.nutrients, - search_query = EXCLUDED.search_query, - embedding = EXCLUDED.embedding, - updated_at = CURRENT_TIMESTAMP - """, - rows, + # An empty image_id is not a harmless blank: it is the conflict target, so + # two such products would overwrite each other and the second would replace + # the first instead of being added. Refuse the batch and name the rows. + if len(image_ids) != len(rows): + unnamed = [r[0] or "" for r in rows if not r[4]] + conn.close() + raise ValueError( + f"{len(unnamed)} product(s) for '{brand}' have no image_id and cannot be " + f"stored (a product needs a name with at least one letter or digit): " + f"{', '.join(unnamed[:5])}" ) - logger.info(f"✅ Upserted {len(rows)} products into {table_name}") - # Remove stale products that were deleted from the source data. - # Only runs when cleanup=True so that callers processing partial - # product sets (e.g. multiple seed files contributing to the same - # brand table) don't accidentally orphan each other's data. - if cleanup and image_ids: - cur.execute( - f"DELETE FROM {table_name} WHERE image_id != ALL(%s::text[])", - (image_ids,), + expected_ids = sorted(set(image_ids)) + + try: + with conn, conn.cursor() as cur: + cur.executemany( + f""" + INSERT INTO {table_name} + (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, embedding) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (image_id) DO UPDATE SET + product_name = EXCLUDED.product_name, + title = EXCLUDED.title, + description = EXCLUDED.description, + category = EXCLUDED.category, + image_url = EXCLUDED.image_url, + image_urls = EXCLUDED.image_urls, + price_range = EXCLUDED.price_range, + size_variants = EXCLUDED.size_variants, + providers = EXCLUDED.providers, + fssai_license = EXCLUDED.fssai_license, + product_sku = EXCLUDED.product_sku, + sku_source = EXCLUDED.sku_source, + hsn_code = EXCLUDED.hsn_code, + final_selling_price = EXCLUDED.final_selling_price, + selling_price = EXCLUDED.selling_price, + barcode = EXCLUDED.barcode, + barcode_type = EXCLUDED.barcode_type, + highlights = EXCLUDED.highlights, + nutrients = EXCLUDED.nutrients, + search_query = EXCLUDED.search_query, + embedding = EXCLUDED.embedding, + updated_at = CURRENT_TIMESTAMP + """, + rows, ) - deleted = cur.rowcount - if deleted: - logger.info(f"🗑️ Removed {deleted} stale product(s) from {table_name}") - conn.close() + # Remove stale products that were deleted from the source data. + # Only runs when cleanup=True so that callers processing partial + # product sets (e.g. multiple seed files contributing to the same + # brand table) don't accidentally orphan each other's data. + if cleanup and image_ids: + cur.execute( + f"DELETE FROM {table_name} WHERE image_id != ALL(%s::text[])", + (image_ids,), + ) + deleted = cur.rowcount + if deleted: + logger.info(f"🗑️ Removed {deleted} stale product(s) from {table_name}") + + # Read the rows back. This is the line that turns "we sent an + # INSERT" into "the data is in the table", and it is the only + # signal the API layer is allowed to report success on. + cur.execute( + f"SELECT COUNT(DISTINCT image_id) FROM {table_name} WHERE image_id = ANY(%s::text[])", + (expected_ids,), + ) + row = cur.fetchone() + persisted = int(row[0]) if row else 0 + + if persisted < len(expected_ids): + raise VectorStoreWriteFailed( + f"Wrote {len(rows)} product(s) for '{brand}' to {table_name} but only " + f"{persisted} of {len(expected_ids)} are readable back afterwards. " + f"The data was not saved - treat this as a failed import." + ) + + logger.info("✅ Upserted %d product(s) into %s (%d verified in table)", + len(rows), table_name, persisted) + finally: + 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' @@ -393,6 +471,8 @@ def upsert_brand_products(brand: str, products: List[Dict[str, Any]], cleanup: b except Exception: # noqa: BLE001 - cache invalidation must never fail a write pass + return persisted + 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""" diff --git a/requirements.txt b/requirements.txt index 6801e76..6647884 100644 --- a/requirements.txt +++ b/requirements.txt @@ -71,6 +71,14 @@ boto3>=1.34.162 # --- missing from this file - declared explicitly now. --- scikit-learn>=1.5.2 pandas>=2.2.2 +# pandas' Excel readers are optional extras it does not install itself, and the +# upload endpoints (/api/user/products/upload-file, /api/upload/*) accept .xlsx +# and .xls. Undeclared, they happened to be present in some environments and +# absent in others - so an Excel upload that worked locally failed in the +# container with "Missing optional dependency 'openpyxl'". openpyxl reads +# .xlsx/.xlsm; xlrd is only for the legacy .xls format. +openpyxl>=3.1.5 +xlrd>=2.0.1 numpy>=1.26.4 scipy>=1.13.1 joblib>=1.4.2 diff --git a/tests/test_user_products_upload.py b/tests/test_user_products_upload.py new file mode 100644 index 0000000..b35198c --- /dev/null +++ b/tests/test_user_products_upload.py @@ -0,0 +1,370 @@ +""" +Tests for the user product upload path. + +The bug these exist for: uploading a spreadsheet returned a success message +while nothing reached the database. So the assertions here are deliberately not +"did it answer 2xx" - they are "did it answer 2xx *and* hand the rows to the +store", and, for every failure mode, "did it refuse to call that a success". + +Hermetic, like the rest of the suite: the store, the embedding model and the +seed-catalog writer are all substituted, so nothing here touches a real +database, downloads a model, or writes into data/seed_catalogs/. +""" +from __future__ import annotations + +import io + +import pytest + +from app.api.routers import user_products +from app.services.vector_store import VectorStoreUnavailable + +UPLOAD_URL = "/api/user/products/upload-file" + +# The headers the frontend's "Download Sample CSV" button produces. +TEMPLATE_CSV = ( + "Brand Name,Product Name / Variant,Category,Price Range,Final Price (₹)," + "Barcode (GTIN/EAN),HSN Code,Custom Image URL,Description\n" + "Lion Dates,Lion Dates 450g,Health Foods,₹160-220,185.00,20086040,2008,,Premium dates\n" + "Naga,Naga Maida 2kg,Flour & Grains,₹90-110,98.00,8906012345001,1101,,Refined wheat flour\n" +) + + +@pytest.fixture +def store(monkeypatch): + """Capture what the endpoint hands to the database instead of writing it. + + Also stubs the two slow collaborators. `embed_texts` would download and load + a sentence-transformer; `get_products_by_brand` would open a real connection + to whatever DB_HOST points at. + """ + calls = [] + + def fake_upsert(brand, products, cleanup=False): + calls.append((brand, products)) + return len(products) + + monkeypatch.setattr(user_products, "upsert_brand_products", fake_upsert) + monkeypatch.setattr(user_products, "embed_texts", lambda texts: [[0.0] * 384 for _ in texts]) + monkeypatch.setattr(user_products, "get_products_by_brand", lambda *a, **k: []) + monkeypatch.setattr(user_products, "upsert_products_into_catalog_file", lambda brand, products: None) + return calls + + +def _upload(client, headers, content: str | bytes, name: str = "products.csv"): + data = content.encode("utf-8") if isinstance(content, str) else content + return client.post(UPLOAD_URL, files={"file": (name, data)}, headers=headers) + + +# --------------------------------------------------------------------------- +# The reported bug: success without persistence +# --------------------------------------------------------------------------- +def test_upload_is_not_a_success_when_the_database_is_unreachable(client, user_headers, monkeypatch, store): + """The exact production symptom. An unreachable database must not answer 201.""" + def unavailable(brand, products, cleanup=False): + raise VectorStoreUnavailable("the product database is unreachable (host=db:5432)") + + monkeypatch.setattr(user_products, "upsert_brand_products", unavailable) + + resp = _upload(client, user_headers, TEMPLATE_CSV) + + assert resp.status_code == 503, resp.text + detail = resp.json()["detail"] + assert "Nothing was saved" in detail + assert "unreachable" in detail + + +def test_single_add_is_not_a_success_when_the_database_is_unreachable(client, user_headers, monkeypatch, store): + def unavailable(brand, products, cleanup=False): + raise VectorStoreUnavailable("the product database is unreachable (host=db:5432)") + + monkeypatch.setattr(user_products, "upsert_brand_products", unavailable) + + resp = client.post( + "/api/user/products/add", + json={"brand": "Lion Dates", "product_name": "Lion Dates 450g"}, + headers=user_headers, + ) + + assert resp.status_code == 503, resp.text + assert "not saved" in resp.json()["detail"] + + +def test_upload_reaches_the_database_and_reports_what_it_stored(client, user_headers, store): + resp = _upload(client, user_headers, TEMPLATE_CSV) + + assert resp.status_code == 201, resp.text + body = resp.json() + assert body["status"] == "success" + assert body["added_count"] == 2 + assert body["error_count"] == 0 + assert body["rows_total"] == 2 + + stored = {p["product_name"]: p for _, products in store for p in products} + assert set(stored) == {"Lion Dates 450g", "Naga Maida 2kg"} + + dates = stored["Lion Dates 450g"] + assert dates["category"] == "Health Foods" + assert dates["final_selling_price"] == 185.0 + assert dates["price_range"] == "₹160-220" + # Not "20086040.0" - an identifier read as a float and stringified is a + # different identifier. + assert dates["barcode"] == "20086040" + assert dates["hsn_code"] == "2008" + assert stored["Naga Maida 2kg"]["barcode"] == "8906012345001" + + +def test_a_real_xlsx_workbook_imports_with_its_identifiers_intact(client, user_headers, store): + """The reported case was an .xlsx upload, and Excel is where identifiers rot. + + A 13-digit barcode in a spreadsheet cell is a number to Excel, so it arrives + as a float and stringifies to "8906012345001.0"; an HSN code of 0402 loses + its leading zero. Both are then stored - silently - as a different value + than the one in the file. + """ + openpyxl = pytest.importorskip("openpyxl", reason="declared in requirements.txt for .xlsx uploads") + + workbook = openpyxl.Workbook() + sheet = workbook.active + sheet.append(["Brand Name", "Product Name", "Final Price (₹)", "Barcode (GTIN/EAN)", "HSN Code"]) + sheet.append(["Lion Dates", "Lion Dates 450g", 185, 8906012345001, "0402"]) + + buffer = io.BytesIO() + workbook.save(buffer) + + resp = _upload(client, user_headers, buffer.getvalue(), name="products.xlsx") + + assert resp.status_code == 201, resp.text + assert resp.json()["added_count"] == 1 + + stored = [p for _, products in store for p in products][0] + assert stored["barcode"] == "8906012345001" + assert stored["hsn_code"] == "0402" + assert stored["final_selling_price"] == 185.0 + + +def test_partial_import_is_reported_as_partial_not_success(client, user_headers, monkeypatch, store): + """One brand failing must neither sink the other nor be called a success.""" + def selective(brand, products, cleanup=False): + if "naga" in brand.lower(): + raise RuntimeError("column overflow on selling_price") + return len(products) + + monkeypatch.setattr(user_products, "upsert_brand_products", selective) + + resp = _upload(client, user_headers, TEMPLATE_CSV) + + assert resp.status_code == 201, resp.text + body = resp.json() + assert body["status"] == "partial" + assert body["added_count"] == 1 + assert body["error_count"] == 1 + assert body["errors"][0]["row"] == 3 # header is row 1, Naga is row 3 + assert "column overflow" in body["errors"][0]["error"] + + +# --------------------------------------------------------------------------- +# Header handling +# --------------------------------------------------------------------------- +def test_template_headers_all_map_to_their_field(): + mapping = user_products.map_spreadsheet_columns([ + "Brand Name", "Product Name / Variant", "Category", "Price Range", + "Final Price (₹)", "Barcode (GTIN/EAN)", "HSN Code", "Custom Image URL", + "Description", + ]) + + assert mapping.columns == { + "brand": "Brand Name", + "product_name": "Product Name / Variant", + "category": "Category", + "price_range": "Price Range", + "final_selling_price": "Final Price (₹)", + "barcode": "Barcode (GTIN/EAN)", + "hsn_code": "HSN Code", + "image_url": "Custom Image URL", + "description": "Description", + } + assert mapping.ignored == [] + + +def test_a_product_sku_column_is_not_mistaken_for_the_product_name(): + """'Product SKU' contains 'product'. Matched loosely, it used to become a + second product_name column, and duplicate columns are what turned a row + into a stringified Series.""" + mapping = user_products.map_spreadsheet_columns(["Brand", "Product Name", "Product SKU"]) + + assert mapping.columns["product_name"] == "Product Name" + assert mapping.columns["product_sku"] == "Product SKU" + + +def test_colliding_columns_are_reported_and_do_not_corrupt_rows(client, user_headers, store): + """Two columns for one field: keep the first, say so, keep importing.""" + csv = ( + "Brand,Product Name,Item Name,Final Price\n" + "Lion Dates,Lion Dates 450g,Ignore This One,185\n" + ) + + resp = _upload(client, user_headers, csv) + + assert resp.status_code == 201, resp.text + body = resp.json() + assert body["added_count"] == 1 + assert body["ignored_columns"] == [ + {"column": "Item Name", "field": "product_name", "using_instead": "Product Name"} + ] + assert any("ignored" in w for w in body["warnings"]) + + stored = [p for _, products in store for p in products] + assert stored[0]["product_name"] == "Lion Dates 450g" + + +def test_missing_required_columns_is_a_400_naming_the_headers(client, user_headers, store): + resp = _upload(client, user_headers, "Foo,Bar\n1,2\n") + + assert resp.status_code == 400 + detail = resp.json()["detail"] + assert "brand column" in detail + assert "Foo" in detail + + +# --------------------------------------------------------------------------- +# Row-level handling +# --------------------------------------------------------------------------- +def test_rows_missing_a_brand_are_errors_not_silent_skips(client, user_headers, store): + csv = ( + "Brand,Product Name\n" + "Lion Dates,Lion Dates 450g\n" + ",Orphan Product\n" + ) + + resp = _upload(client, user_headers, csv) + + body = resp.json() + assert resp.status_code == 201, resp.text + assert body["added_count"] == 1 + assert body["error_count"] == 1 + assert body["errors"][0]["row"] == 3 + assert "brand" in body["errors"][0]["error"] + + +def test_fully_blank_rows_are_skipped_and_counted(client, user_headers, store): + csv = ( + "Brand,Product Name\n" + "Lion Dates,Lion Dates 450g\n" + ",\n" + ",\n" + ) + + resp = _upload(client, user_headers, csv) + + body = resp.json() + assert body["added_count"] == 1 + assert body["error_count"] == 0 + assert body["skipped_blank_rows"] == 2 + + +def test_a_file_where_every_row_fails_is_never_a_success(client, user_headers, store): + csv = ( + "Brand,Product Name\n" + ",Orphan One\n" + ",Orphan Two\n" + ) + + resp = _upload(client, user_headers, csv) + + assert resp.status_code == 422, resp.text + assert "Nothing was saved" in resp.json()["detail"] + assert store == [] + + +def test_empty_and_headers_only_files_are_rejected(client, user_headers, store): + assert _upload(client, user_headers, b"").status_code == 400 + assert _upload(client, user_headers, "Brand,Product Name\n").status_code == 400 + + +def test_unsupported_file_type_is_rejected(client, user_headers, store): + resp = _upload(client, user_headers, b"%PDF-1.4", name="products.pdf") + + assert resp.status_code == 400 + assert "Unsupported file type" in resp.json()["detail"] + + +def test_oversized_row_count_is_rejected_before_any_write(client, user_headers, store): + rows = "".join(f"Lion Dates,Product {i}\n" for i in range(user_products.MAX_UPLOAD_ROWS + 1)) + + resp = _upload(client, user_headers, "Brand,Product Name\n" + rows) + + assert resp.status_code == 413 + assert store == [] + + +# --------------------------------------------------------------------------- +# Batching: the collaborators must be called once per upload, not once per row +# --------------------------------------------------------------------------- +def test_one_embedding_call_and_one_write_per_brand(client, user_headers, monkeypatch, store): + embed_calls = [] + sample_calls = [] + # Pinned rather than inherited: the assertion below is about batching, and + # it should not silently pass because a .env happened to disable embeddings. + monkeypatch.setattr(user_products, "USE_EMBEDDINGS", True) + monkeypatch.setattr(user_products, "embed_texts", + lambda texts: embed_calls.append(len(texts)) or [[0.0] * 384 for _ in texts]) + monkeypatch.setattr(user_products, "get_products_by_brand", + lambda brand, **k: sample_calls.append(brand) or []) + + csv = "Brand,Product Name\n" + "".join( + f"Lion Dates,Lion Dates {i}g\n" for i in range(10) + ) + + resp = _upload(client, user_headers, csv) + + assert resp.json()["added_count"] == 10 + assert embed_calls == [10], "embeddings must be generated in one batched call" + assert len(store) == 1, "one write per brand, not one per row" + assert len(sample_calls) == 1, "one brand-sample read per brand, not one per row" + + +def test_products_still_save_when_embeddings_are_disabled(client, user_headers, monkeypatch, store): + """USE_EMBEDDINGS=false must skip the model, not the row.""" + def fail_if_called(texts): + raise AssertionError("embed_texts must not be called when USE_EMBEDDINGS is false") + + monkeypatch.setattr(user_products, "USE_EMBEDDINGS", False) + monkeypatch.setattr(user_products, "embed_texts", fail_if_called) + + resp = _upload(client, user_headers, TEMPLATE_CSV) + + assert resp.status_code == 201, resp.text + assert resp.json()["added_count"] == 2 + assert all(p["embedding"] is None for _, products in store for p in products) + + +def test_a_failed_embedding_does_not_lose_the_product(client, user_headers, monkeypatch, store): + """Semantic search is a feature of the row; it is not the row.""" + monkeypatch.setattr(user_products, "USE_EMBEDDINGS", True) + monkeypatch.setattr(user_products, "embed_texts", + lambda texts: (_ for _ in ()).throw(RuntimeError("model not downloaded"))) + + resp = _upload(client, user_headers, TEMPLATE_CSV) + + assert resp.status_code == 201, resp.text + assert resp.json()["added_count"] == 2 + + +def test_the_seed_catalog_is_only_written_for_rows_the_database_took( + client, user_headers, monkeypatch, store +): + """Dual persistence must not become divergent persistence.""" + catalog_writes = [] + monkeypatch.setattr(user_products, "upsert_products_into_catalog_file", + lambda brand, products: catalog_writes.append((brand, len(products)))) + + def unavailable(brand, products, cleanup=False): + raise VectorStoreUnavailable("database unreachable") + + monkeypatch.setattr(user_products, "upsert_brand_products", unavailable) + + resp = _upload(client, user_headers, TEMPLATE_CSV) + + assert resp.status_code == 503 + assert catalog_writes == [], "the JSON catalog must not gain products the database refused"