#!/usr/bin/env python3 """Builds supabase/seed/product_index.csv from the Open Beauty Facts export (docs/dispatch/AH). curl -O https://static.openbeautyfacts.org/data/openbeautyfacts-products.jsonl.gz python3 scripts/build_product_index.py openbeautyfacts-products.jsonl.gz [--stats stats.json] Stdlib only. What it does, in order: 1. Reads every OBF row. Keeps face and skin care that maps to one of the app's categories (cleanser, toner, essence, serum, eye, moisturizer, face_oil, sunscreen, spot) and drops hair, body, hands, makeup, fragrance, oral, shaving, masks, wipes and food. The classifier reads the product name, the generic name and the category tags; OBF tags are sparse (about 4 in 10 rows carry any), so names do most of the work. Rows with no brand or no name are dropped. 2. Parses actives from the ingredient list with scripts/inci_actives.py (the app's core/rules/inci_actives.dart is the same parser). actives_source is `inci` when the list was readable, else `none`: unknown, never "no actives". 3. Dedups with the same normalisation the app uses to match reads (scripts/shelf_identity.py, ShelfIdentity in Dart): one row per normalised brand and name-token set. The row kept is the one with a readable ingredient list, then the most complete, then the most recently edited. 4. Drops OBF rows that are the same product as a curated catalog row (data/catalog.csv): the curated row is in the index already (the migration copies catalog_products) and wins. 5. Cleans what is shown (ticket 0027): HTML entities decoded, junk before the first word dropped (">>BRINGGREEN"), a brand stored as a product line mapped to its maker (BRAND_ALIASES), known brands in their own casing (BRAND_DISPLAY), and all-caps brands and names title-cased. Search is case-insensitive, so casing is display only. 6. A sunscreen with no readable ingredient list gets actives ["sunscreen"] and actives_source `category`: a sunscreen carries UV filters by definition. No other category implies actives. The CSV has OBF data only, which is ODbL: keep it in product_index, `source = 'obf'`, with the OBF code, and never merge it into the curated catalog. scripts/load_product_index.sh loads it. """ import argparse import collections import csv import gzip import html import json import re import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import inci_actives # noqa: E402 import shelf_identity # noqa: E402 REPO = Path(__file__).resolve().parent.parent OUT = REPO / "supabase" / "seed" / "product_index.csv" CATALOG = REPO / "data" / "catalog.csv" MAX_INGREDIENTS = 4000 fold = shelf_identity.fold def rx(*words): return re.compile("|".join(words)) # Checked against the name, generic name and category tags, folded. EXCLUDE = rx( r"\bhair\b", r"cheveu", r"\bhaar", r"shampo", r"conditioner", r"apres-?shampo", r"capillaire", r"coloration", r"hair-dyes", r"hair-care", r"\bdeo", r"antiperspirant", r"anti-perspirant", r"toothpaste", r"dentifrice", r"tandpasta", r"zahnpasta", r"mouthwash", r"bain de bouche", r"\btooth", r"dental", r"eau de parfum", r"eau de toilette", r"perfume", r"cologne", r"en:perfumes", r"mascara", r"lipstick", r"\blips?\b", r"lip ?balm", r"levres", r"lippen", r"gloss", r"foundation", r"fond de teint", r"concealer", r"eyeshadow", r"eye shadow", r"eyeliner", r"\bkohl", r"\bblush", r"bronzer", r"highlighter", r"\bprimer\b", r"\bbb cream", r"\bcc cream", r"powder", r"poudre", r"\bnail", r"vernis", r"ongles", r"cuticle", r"makeup(?![- ]remover)", r"make-up(?![- ]remover)", r"maquillage", r"\bshav", r"rasage", r"aftershave", r"after shave", r"after-shave", r"\bbeard", r"barbe", r"wipes", r"lingette", r"doekjes", r"intim", r"\bmask", r"masque", r"masker", r"maske\b", r"after ?sun", r"after-sun", r"apres[- ]soleil", r"self[- ]tan", r"autobronz", r"\bhand", r"\bmains?\b", r"\bfoot\b", r"\bfeet\b", r"\bpieds?\b", r"en:plant-based", r"en:dietary", r"en:dairies", r"en:fats", r"en:beverages", r"en:condoms", r"en:foods", r"supplement", r"\bdiaper", r"\bcouche", r"tattoo", r"\bbaby\b", r"\bbebe\b", r"gumm", r"chewable", r"throat", r"tablet", r"capsule", r"physiologique", r"\blash", r"\bcils\b", r"\bbrow", r"wimper", r"kirpik", ) # Body words exclude a row only when nothing says face: "face and body" creams stay. BODY = rx(r"\bbody", r"\bcorps\b", r"korper", r"lichaam", r"bodylotion", r"shower", r"douche", r"\bbath", r"\bbain\b", r"dusch", r"\bbad\b", r"\bsoap", r"savon", r"seife", r"\bzeep", r"\blegs?\b") FACE = rx(r"\bface\b", r"\bfacial", r"visage", r"gezicht", r"gesicht", r"rostro", r"\bviso\b", r"\bcara\b", r"skin ?care", r"en:face", r"facial-creams", r"day-creams", r"night-creams", r"face-lotions", r"anti-aging-face-care", r"anti-wrinkles-creams", r"huidverzorging", r"gezichts") SUNSCREEN = rx(r"\bspf ?\d", r"sunscreen", r"sun ?screen", r"sun (?:cream|lotion|fluid|milk|spray|stick|gel)", r"sunblock", r"solaire", r"\becran\b", r"zonnebrand", r"zonnecreme", r"sonnencreme", r"sonnenschutz", r"sonnenmilch", r"protector solar", r"protecao solar", r"fotoprotector", r"\buv ?(?:fluid|defen|protect)", r"en:sunscreen", r"en:in-sun-protections") CLEANSER = rx(r"cleans", r"face wash", r"facial wash", r"\bwash\b", r"micell", r"mizell", r"nettoyant", r"demaquill", r"make-?up remover", r"reinigung", r"reiniging", r"limpiador", r"limpeza", r"gel lavant", r"mousse nettoyante", r"lavant", r"purif(?:ying|iant) (?:gel|foam)", r"gel purifiant", r"gel moussant", r"\bfoam", r"\bscrub", r"gommage", r"peeling", r"en:cleansers", r"en:cleansing-waters", r"en:micellar-water", r"gezichtsreiniging") EYE = rx(r"\beyes?\b", r"\byeux\b", r"oogcreme", r"ooggel", r"augen", r"under[- ]eye", r"en:eye-creams") SPOT = rx(r"(?= 2} with CATALOG.open(newline="", encoding="utf-8") as f: for r in csv.DictReader(f): spelled.setdefault(shelf_identity.normalise_brand(r["brand"]), r["brand"]) for b in SUB_BRANDS: spelled[shelf_identity.normalise_brand(b)] = b skip = PARENT_BRANDS | GENERIC_BRANDS | TOO_GENERIC_TO_MATCH return {k: display_brand(v) for k, v in spelled.items() if k and k not in skip} def repair_brand(brand, name, known): """(brand, name) with a parent company or a generic phrase replaced by the brand the name starts with, taken off the front of the name. Unchanged when the brand is fine or the name does not start with a known brand. The name can come back empty ("sensitive skin | Aveeno"): the caller drops that row, a brand with no product.""" b = shelf_identity.normalise_brand(brand) if b not in PARENT_BRANDS and b not in GENERIC_BRANDS: return brand, name words = name.split() best = None for k in range(min(len(words), 5), 0, -1): head = shelf_identity.normalise_brand(" ".join(words[:k])) if head in known: best = (known[head], " ".join(words[k:]).strip(" -,:")) break return best if best else (brand, name) # --- Display cleanup (ticket 0027) --------------------------------------------------------------- # OBF rows that store a product line, not the maker, as the brand. Keyed by normalised brand. BRAND_ALIASES = {"1025": "Round Lab"} # Round Lab's 1025 Dokdo line # Brands whose own casing title-casing would break. Keyed by normalised brand. BRAND_DISPLAY = { "cerave": "CeraVe", "cosrx": "COSRX", "eltamd": "EltaMD", "skin1004": "SKIN1004", "larocheposay": "La Roche-Posay", "paulaschoice": "Paula's Choice", "ordinary": "The Ordinary", "elf": "e.l.f.", "roundlab": "Round Lab", # Short enough to be kept in capitals as an acronym otherwise. "olay": "Olay", "nuxe": "Nuxe", "mixa": "Mixa", "hipp": "HiPP", } # Kept in capitals when title-casing an all-caps name. ACRONYMS = {"spf", "uv", "uva", "uvb", "pa", "aha", "bha", "pha", "ha", "am", "pm", "sos", "bb", "cc", "dd", "xl", "ii", "iii", "iv", "lsf", "fps", "ip", "ph", "dna", "pdrn", "q10", "b3", "b5", "b12", "c", "e", "npa", "ac", "ai", "dx", "ds", "k", "h", "r", "mela"} # Words with their own casing. SPECIAL_CASE = {"ph": "pH"} # A brand this short in capitals is usually an acronym (NYX, SVR, QV): left as it is. SHORT_BRAND = 4 _JUNK_PREFIX = re.compile(r"^[^\w'\"(\[]+") _SPACES = re.compile(r"\s+") def clean_text(s): """Entities decoded (twice, for double-encoded rows), non-breaking spaces, junk before the first word dropped, whitespace collapsed.""" t = html.unescape(html.unescape(s or "")).replace("\u00a0", " ") t = _JUNK_PREFIX.sub("", t) return _SPACES.sub(" ", t).strip() def all_caps(s): letters = [c for c in s if c.isalpha()] return len(letters) >= 2 and all(c.isupper() for c in letters) def _title_word(w): core = re.sub(r"[^\w]", "", w).lower() if core in SPECIAL_CASE: return w.lower().replace(core, SPECIAL_CASE[core]) if core in ACRONYMS or any(ch.isdigit() for ch in w): return w out, up = [], True for i, ch in enumerate(w.lower()): out.append(ch.upper() if up and ch.isalpha() else ch) if ch.isalpha(): up = False # A new word after a hyphen, slash or bracket; after the French elision l' or d'. if ch in "-/([" or (ch in "'\u2019" and i == 1 and w[0].lower() in "ld"): up = True return "".join(out) def title_case(s): return " ".join(_title_word(w) for w in s.split(" ")) def display_brand(brand): known = BRAND_DISPLAY.get(shelf_identity.normalise_brand(brand)) if known: return known if all_caps(brand) and sum(c.isalpha() for c in brand) > SHORT_BRAND: return title_case(brand) return brand _BRAND_IN_NAME = [(re.compile(r"(? rank(groups[key]): groups[key] = row else: groups[key] = row rows, changes = cleanup(list(groups.values()), curated) stats["brand_repaired"] = len(changes["repaired"]) stats["dropped_after_repair"] = len(changes["removed"]) stats["kept"] = len(rows) for r in rows: stats["kept_by_category"][r["category"]] += 1 stats["kept_by_actives_source"][r["actives_source"]] += 1 stats["top_brands"][r["brand"]] += 1 if r["actives"] != "[]": stats["kept_with_actives"] += 1 if html.unescape(r["raw_brand"] + r["raw_name"]) != r["raw_brand"] + r["raw_name"]: stats["cleaned_entities"] += 1 if _JUNK_PREFIX.match(html.unescape(r["raw_brand"])) or _JUNK_PREFIX.match(html.unescape(r["raw_name"])): stats["cleaned_junk_prefix"] += 1 if shelf_identity.normalise_brand(r["clean_brand"]) != shelf_identity.normalise_brand(clean_text(r["raw_brand"])): stats["brand_aliased"] += 1 stats["recased_brand"] += r["brand"] != r["clean_brand"] stats["recased_name"] += r["name"] != r["clean_name"] if r["actives_source"] == "category": stats["sunscreen_from_category"] += 1 if r["category"] == "sunscreen" and r["actives_source"] == "inci" and "sunscreen" not in r["actives"]: stats["sunscreen_inci_without_filter"] += 1 stats["top_brands"] = collections.Counter(dict(stats["top_brands"].most_common(30))) stats["short_or_numeric_brands"] = sorted({r["brand"] for r in rows if len(r["brand"]) < 3 or r["brand"].isdigit()}) return rows, stats def cleanup(rows, curated): """The AL item 3 pass over finished rows: drops what is not skincare, repairs brands, then drops what the repair made a duplicate of another row or of a curated product. Returns the rows, sorted, and what changed: {"removed": [(why, row)], "repaired": [(row before, row after)]}.""" known = known_brands(rows) changes = {"removed": [], "repaired": []} kept = [] for r in rows: why = not_skincare(r["brand"], r["name"], r.get("ingredients_text", "")) if why: changes["removed"].append((why, r)) continue brand, name = repair_brand(r["brand"], r["name"], known) if (brand, name) != (r["brand"], r["name"]): if not name: changes["removed"].append(("brand only, no product name", r)) continue fixed = {**r, "brand": brand, "name": display_name(name)} changes["repaired"].append((r, fixed)) r = fixed kept.append(r) seen = {} out = [] for r in sorted(kept, key=lambda r: (r["actives_source"] != "inci", r["source_code"])): key = identity_key(r["brand"], r["name"]) if key in curated: changes["removed"].append(("same as a curated product after repair", r)) elif key in seen: changes["removed"].append(("duplicate after repair", r)) else: seen[key] = r out.append(r) out.sort(key=lambda r: (fold(r["brand"]), fold(r["name"]), r["source_code"])) return out, changes def refilter(path): """Applies [cleanup] to an already built product_index.csv, so the filter can change without a new OBF download mixing new rows into the before and after.""" with Path(path).open(newline="", encoding="utf-8") as f: rows = list(csv.DictReader(f)) out, changes = cleanup(rows, curated_keys()) return rows, out, changes COLUMNS = ["source", "source_code", "brand", "name", "category", "actives", "actives_source", "ingredients_text"] def write(rows): OUT.parent.mkdir(parents=True, exist_ok=True) with OUT.open("w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=COLUMNS, extrasaction="ignore", lineterminator="\n") w.writeheader() w.writerows(rows) def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("export", nargs="?", help="openbeautyfacts-products.jsonl(.gz)") ap.add_argument("--stats", help="write the counts as JSON here too") ap.add_argument("--refilter", action="store_true", help="re-apply the cleanup to the committed product_index.csv instead of building from an export") ap.add_argument("--changes", help="with --refilter: write every removed and repaired row here as JSON") args = ap.parse_args() if args.refilter: before, rows, changes = refilter(OUT) write(rows) summary = {"before": len(before), "after": len(rows), "removed_by_reason": dict(collections.Counter(why for why, _ in changes["removed"]).most_common()), "repaired": len(changes["repaired"])} print(json.dumps(summary, indent=1)) if args.changes: Path(args.changes).write_text(json.dumps({ "summary": summary, "removed": [{"why": why, "source_code": r["source_code"], "brand": r["brand"], "name": r["name"], "category": r["category"]} for why, r in changes["removed"]], "repaired": [{"source_code": a["source_code"], "before": [a["brand"], a["name"]], "after": [b["brand"], b["name"]]} for a, b in changes["repaired"]], }, indent=1, ensure_ascii=False)) return if not args.export: ap.error("an export file, or --refilter") rows, stats = build(args.export) write(rows) printable = {k: (dict(v.most_common()) if isinstance(v, collections.Counter) else v) for k, v in stats.items()} print(json.dumps(printable, indent=1)) if args.stats: Path(args.stats).write_text(json.dumps(printable, indent=1)) print(f"wrote {OUT.relative_to(REPO)}: {len(rows)} rows, {OUT.stat().st_size / 1e6:.1f} MB", file=sys.stderr) if __name__ == "__main__": main()