148 lines
5.8 KiB
Python
148 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Retroactively fix cross-category language in already-generated product
|
|
descriptions, and recompute embeddings from the cleaned text.
|
|
|
|
Why this is needed
|
|
-------------------
|
|
This is a one-time (or occasional) *data* fix to go with the *code* fix in
|
|
`app/services/category_registry.py` / `app/core/catalog_engine.py`. New
|
|
catalog generations already sanitize descriptions before embedding them,
|
|
but products ingested BEFORE that change (including the bundled
|
|
`data/seed_catalogs/*.json` samples) may still have descriptions like:
|
|
|
|
"ITC Bingo Korean Style is a crispy, savory biscuit that adds a
|
|
touch of spice..."
|
|
|
|
for a product whose actual category is `Snacks`, not `Biscuits & Cookies`.
|
|
That leftover word is exactly what let a query like "recommend biscuits
|
|
with low sugar" surface Bingo snack products - the description text (and
|
|
therefore its embedding) said "biscuit" even though the product isn't one.
|
|
|
|
What this script does
|
|
----------------------
|
|
For every product in every `data/seed_catalogs/*.json` file:
|
|
1. Re-run `sanitize_category_language(description, category)` to strip
|
|
out any other category's keywords.
|
|
2. If the description changed, recompute its embedding the same way
|
|
`catalog_engine.py` does at ingestion time.
|
|
3. Write the updated JSON back to disk.
|
|
|
|
Pass `--apply-to-db` to also push the corrected rows straight into
|
|
pgvector (equivalent to re-running `seed_sample_data.py` afterwards).
|
|
|
|
Usage:
|
|
python scripts/resanitize_catalog.py # dry-run, prints a diff summary
|
|
python scripts/resanitize_catalog.py --write # updates the JSON files in place
|
|
python scripts/resanitize_catalog.py --write --apply-to-db
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from app.services.category_registry import sanitize_category_language # 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 _process_file(path: Path, write: bool) -> List[Dict[str, Any]]:
|
|
"""Returns the list of products whose description changed (for reporting)."""
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
products = data.get("products", [])
|
|
changed: List[Dict[str, Any]] = []
|
|
needs_reembed: List[int] = []
|
|
|
|
for i, p in enumerate(products):
|
|
original_desc = p.get("description") or ""
|
|
category = p.get("category") or "General"
|
|
cleaned_desc = sanitize_category_language(original_desc, category)
|
|
if cleaned_desc != original_desc:
|
|
changed.append({
|
|
"title": p.get("title") or p.get("product_name"),
|
|
"category": category,
|
|
"before": original_desc[:160],
|
|
"after": cleaned_desc[:160],
|
|
})
|
|
p["description"] = cleaned_desc
|
|
needs_reembed.append(i)
|
|
|
|
if needs_reembed and write:
|
|
# Local import: only pay the torch/sentence-transformers startup
|
|
# cost when there's actually something to re-embed.
|
|
from app.services.embeddings_service import embed_texts
|
|
|
|
texts = []
|
|
for i in needs_reembed:
|
|
p = products[i]
|
|
title = p.get("title") or ""
|
|
cat = p.get("category") or ""
|
|
desc = p.get("description") or ""
|
|
texts.append(f"{title} [CAT={cat}] :: {desc}")
|
|
vectors = embed_texts(texts)
|
|
for i, vec in zip(needs_reembed, vectors):
|
|
products[i]["embedding"] = vec
|
|
logger.info("Recomputed %d embedding(s) for %s", len(needs_reembed), path.name)
|
|
|
|
if write and changed:
|
|
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
logger.info("Wrote %d description fix(es) to %s", len(changed), path.name)
|
|
|
|
return changed
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--write", action="store_true", help="Write changes to the JSON files (default: dry-run report only)")
|
|
parser.add_argument("--apply-to-db", action="store_true", help="Also upsert the corrected rows into pgvector (implies --write)")
|
|
args = parser.parse_args()
|
|
write = args.write or args.apply_to_db
|
|
|
|
if not SEED_DIR.exists():
|
|
logger.error("Seed directory not found: %s", SEED_DIR)
|
|
sys.exit(1)
|
|
|
|
total_changed = 0
|
|
for path in sorted(SEED_DIR.glob("*.json")):
|
|
changed = _process_file(path, write=write)
|
|
total_changed += len(changed)
|
|
for c in changed:
|
|
logger.info(" [%s] %r\n before: %s...\n after: %s...",
|
|
c["category"], c["title"], c["before"], c["after"])
|
|
|
|
if not write:
|
|
logger.info(
|
|
"%d description(s) would be changed across %d file(s). "
|
|
"Re-run with --write to apply (and --apply-to-db to also update pgvector).",
|
|
total_changed, len(list(SEED_DIR.glob('*.json'))),
|
|
)
|
|
return
|
|
|
|
logger.info("%d description(s) fixed.", total_changed)
|
|
|
|
if args.apply_to_db:
|
|
from app.services.vector_store import ensure_brand_schema, upsert_brand_products
|
|
|
|
for path in sorted(SEED_DIR.glob("*.json")):
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
brand = data.get("brand")
|
|
products = data.get("products", [])
|
|
if not brand or not products:
|
|
continue
|
|
ensure_brand_schema(brand)
|
|
upsert_brand_products(brand, products, cleanup=True)
|
|
logger.info("Upserted %d product(s) for brand %r into pgvector", len(products), brand)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|