#!/usr/bin/env python3 """ Builds `brands.db`: an offline, bundle-able SQLite file of the most popular brand-name foods sold in Canada and the US (protein shakes, bars and powders always in), from two open sources: * USDA FoodData Central, Branded Foods (CC0 1.0, public domain). Read from `foods.copy.csv`, the vetted output of the server importer (`server/scripts/import-usda-branded.ts build `), so the same rules apply on the phone and the server (newest row per barcode, calorie check, US/CA only). The release zip itself is read only for each food's USDA category. * Open Food Facts (database ODbL 1.0, contents DbCL 1.0). Use the JSONL export, pre-filtered to US/Canada lines while it downloads (13 GB, never stored whole): curl -sSfL https://static.openfoodfacts.org/data/openfoodfacts-products.jsonl.gz | gzip -dc \ | LC_ALL=C grep -F -e '"en:united-states"' -e '"en:canada"' | gzip -1 > off-us-ca.jsonl.gz The Parquet export (Hugging Face, needs duckdb) is also read, but downloaded at ~0.2 MB/s here. The nightly .csv.gz is read too but, checked 2026-10-03, it leaves the nutrition columns empty for many US products whose numbers the API and JSONL have (5 of 5 sampled), so it undercounts. Wired into the app on gf/v2-brands-wire (2026-10-03): the app reads it through app/src/features/foods/brandsDb.ts. Report and decisions: overnight/2026-10-03-food-data/branded.md. Run (Python 3.10+; standard library, plus duckdb for the Parquet input). SQLite is only ever written in a scratch folder (never on the Documents mount); the finished file is checked (PRAGMA integrity_check) and then written into the repo as a gzip copy plus its sha256: python3 app/scripts/build-brands/build_brands.py \ --usda-copy ../.data/usda/foods.copy.csv \ --usda-zip ../.data/usda/FoodData_Central_branded_food_csv_2026-04-30.zip \ --off ../.data/off/off-us-ca.jsonl.gz \ --top 450000 --fts --out "$(mktemp -d)/brands.db" \ --gz app/assets/brands.db.gz \ [--cache ../.data/off/parsed-sources.pickle] [--measure 100000,200000,300000,400000] [--stats stats.json] `--gz` also writes app/assets/brands.db.sha256 (the sha256 of the unzipped file). The app's postinstall step (app/scripts/unpack-brands.js) unzips the .gz to app/assets/brands.db, which git ignores, whenever that file is missing or its sha256 differs. Copy the same .gz and this script to website/src/static/data/ for the free download (ODbL 4.6); the website's release build refuses a copy that differs from the app's. What it does, in order: 1. Reads both sources into one shape: name, brand, barcode(s), per-100 g kcal/protein/carbs/fat (all four required) plus fibre, sugar, sat fat, sodium (mg) when given, serving grams + label, countries, Open Food Facts' scan count, category words. 2. Drops rows whose numbers can't be right for what the name says (`implausible`: a protein bar or bread at 0/0/0/0, a "protein" food or a nut butter with 0 g protein). Then merges by barcode (13 digits, or 8 for EAN-8, the app's `canonicalBarcode` rule). Where both sources have a barcode, USDA's manufacturer numbers win; Open Food Facts adds its scan count and fills in what USDA left empty (fibre, sugar, sat fat, sodium, serving, brand). A row filled in that way says so (`from_off` = 2), so the app credits both sources on it. 3. Normalises brands ("Kirkland" = "Kirkland Signature", "Fairlife, LLC" = "fairlife", USDA's "Fa!rlife" = Fairlife: a "!" between letters is an "i") and folds rows that are the same product in different pack sizes: same brand, same name (a name that is only the brand, like "Pepsi" by Pepsi, included), kcal within 3% and protein within 5%. Every barcode of a folded row still points at the kept row. 4. Ranks: Open Food Facts' per-country scan rank for Canada/the US (popularity_tags, last three years), then its raw scan count, then the brand's total (so USDA-only items of popular brands come before unknown ones), then completeness. Every protein shake/powder/bar/drink passing the checks is kept regardless of rank. 5. Writes the top N into `brand_foods` (id = rank, so the word index returns the most scanned first) + `brand_barcodes`, the contentless word index `brand_fts`, and a `meta` table carrying the attribution and licence notices. The shape is schema.sql; numbers are kept as whole tenths and `from_off` says which source each row came from. Because any Open Food Facts row makes the whole file a Derivative Database under ODbL 4.4, the FILE is ODbL 1.0 (USDA rows are public domain, so mixing them in loses nothing). The per-row `from_off` says where that row came from: 0 = USDA only, 1 = Open Food Facts' numbers, 2 = USDA's numbers with details Open Food Facts filled in (`FROM_USDA`, `FROM_OFF`, `FROM_USDA_AND_OFF`). """ from __future__ import annotations import argparse import csv import gzip import io import json import os import pickle import re import sqlite3 import sys import tempfile import time import unicodedata import zipfile from collections import Counter, defaultdict csv.field_size_limit(sys.maxsize) KJ_PER_KCAL = 4.184 SALT_PER_SODIUM = 2.5 MAX_KCAL_100 = 900.0 USDA_LICENCE = "CC0-1.0" OFF_LICENCE = "ODbL-1.0" ATTRIBUTION = ( "Contains data from Open Food Facts (https://world.openfoodfacts.org), available under the " "Open Database License (https://opendatacommons.org/licenses/odbl/1-0/); individual contents " "under the Database Contents License. Contains USDA FoodData Central Branded Foods " "(https://fdc.nal.usda.gov), public domain (CC0 1.0)." ) FILE_LICENCE = ( "This database is made available under the Open Database License: " "https://opendatacommons.org/licenses/odbl/1-0/. Any rights in individual contents of the " "database are licensed under the Database Contents License: " "https://opendatacommons.org/licenses/dbcl/1-0/" ) # --- protein products: always kept ---------------------------------------------------------------- PROTEIN_OFF_TAGS = ( "en:protein-shakes", "en:protein-powders", "en:protein-bars", "en:protein-drinks", "en:whey-proteins", "en:protein-supplements", "en:bodybuilding-supplements", "en:protein-puddings", "en:high-protein-yogurts", "en:meal-replacements", ) PROTEIN_USDA_CATEGORIES = re.compile(r"protein|sport drinks|energy, protein|meal replacement", re.I) PROTEIN_NAME = re.compile( r"\b(protein|whey|isolate|casein|mass gainer|gainer|core power|premier protein|muscle milk)\b", re.I ) # --- barcodes (mirror of app/src/features/barcode/code.ts) --------------------------------------- def gtin_ok(code: str) -> bool: if not re.fullmatch(r"\d{8,14}", code): return False digits = [int(c) for c in code] check = digits.pop() s = sum(d * (3 if i % 2 == 0 else 1) for i, d in enumerate(reversed(digits))) return (10 - s % 10) % 10 == check def expand_upce(code: str) -> str | None: if not re.fullmatch(r"[01]\d{7}", code): return None n, (d1, d2, d3, d4, d5, d6), c = code[0], code[1:7], code[7] if d6 in "012": body = f"{d1}{d2}{d6}0000{d3}{d4}{d5}" elif d6 == "3": body = f"{d1}{d2}{d3}00000{d4}{d5}" elif d6 == "4": body = f"{d1}{d2}{d3}{d4}00000{d5}" else: body = f"{d1}{d2}{d3}{d4}{d5}0000{d6}" return f"{n}{body}{c}" def canonical_barcode(raw: str) -> str | None: code = re.sub(r"[\s-]", "", raw or "") if not code.isdigit(): return None if len(code) == 14 and code[0] == "0": # GTIN-14 with an empty indicator digit code = code[1:] if len(code) == 13: return code if gtin_ok(code) else None if len(code) == 12: return "0" + code if gtin_ok(code) else None if len(code) == 8: a = expand_upce(code) if a and gtin_ok(a): return "0" + a return code if gtin_ok(code) else None return None def barcode_int(code: str) -> int: """13-digit codes as their number; EAN-8 lifted above every 13-digit code so they never clash.""" return int(code) if len(code) == 13 else 10**13 + int(code) # --- search words: an EXACT port of app/src/features/foods/searchWords.ts ---------------------- # The app reads typed words with searchTokens/wordForms; the word index must hold the words # indexWords makes, or a search the generic list answers would miss here. test_build_brands.py # checks every function below against the TypeScript one, on the shared examples in # search-words-examples.json and (when the app's node_modules are there) on random strings. # Change searchWords.ts and this together. # Straight, curly (’ ‘), modifier-letter (ʼ), backtick and acute: iOS types ’ by default. APOSTROPHES = re.compile("['\u2019\u2018\u02bc`\u00b4]") _COMBINING = re.compile("[\u0300-\u036f]") _NOT_WORD = re.compile(r"[^a-z0-9]+") _AMP_JOIN = re.compile(r"([a-z0-9])&(?=[a-z0-9])") _HYPHEN_CHAIN = re.compile(r"[a-z0-9]+(?:-[a-z0-9]+)+") # re.ASCII: JavaScript's \b only knows [A-Za-z0-9_], Python's would also count "ß" or "é" as letters. _DOT_CHAIN = re.compile(r"\b[a-z](?:\.[a-z])+\b", re.ASCII) _HAS_LETTER = re.compile(r"[a-z]") def _fold(text: str) -> str: """Accents off, lower case (searchWords.ts fold).""" return _COMBINING.sub("", unicodedata.normalize("NFD", text)).lower() def _split(text: str) -> list[str]: return [t for t in _NOT_WORD.split(text) if t] def tokenize(text: str) -> list[str]: """The plain split: accents off, lower case, split on anything not a-z or 0-9 (searchWords.ts tokenize). Brand and name keys still use it.""" return _split(_fold(text)) def _canonical(text: str) -> str: """Apostrophes out without splitting, "&" between letters joins, any other "&" is "and".""" return _AMP_JOIN.sub(r"\1", APOSTROPHES.sub("", _fold(text))).replace("&", " and ") def search_tokens(text: str) -> list[str]: """What a person typed, as search words (searchWords.ts searchTokens).""" return _split(_canonical(text)) def joined_word(text: str) -> str: """A name as ONE search word: "B Fit" -> "bfit" (searchWords.ts joinedWord).""" return "".join(search_tokens(text)) def index_words(text: str) -> list[str]: """The words to store for `text` (searchWords.ts indexWords): searchTokens first, then the plain split, "&" spelt out, and hyphen or dot chains as one word.""" words = dict.fromkeys(search_tokens(text)) for w in tokenize(text): words.setdefault(w) for w in _split(APOSTROPHES.sub("", _fold(text)).replace("&", " and ")): words.setdefault(w) c = _canonical(text) for m in _HYPHEN_CHAIN.finditer(c): joined = m.group(0).replace("-", "") if _HAS_LETTER.search(joined): words.setdefault(joined) for m in _DOT_CHAIN.finditer(c): words.setdefault(m.group(0).replace(".", "")) return list(words) def singular(token: str) -> str | None: """A search word ending in one "s" also matches the word without it, whole (searchWords.ts).""" return token[:-1] if len(token) >= 4 and token.endswith("s") and not token.endswith("ss") else None def word_forms(tokens: list[str], i: int) -> list[tuple[str, bool]]: """(text, whole) for every stored word search word `i` may match (searchWords.ts wordForms): the start of a word, glued to the words typed before it, or its singular as a whole word.""" forms = [(tokens[i], False)] for j in range(i - 1, -1, -1): forms.append(("".join(tokens[j : i + 1]), False)) one = singular(tokens[i]) if one: forms.append((one, True)) return forms LEGAL = re.compile( r"[,.]?\s+(inc\.?|incorporated|llc\.?|l\.l\.c\.?|ltd\.?|limited|corp\.?|corporation|co\.?|company|ulc|lp|gmbh|s\.?a\.?)$", re.I, ) PLACEHOLDER = re.compile(r"^(none|n/?a|null|not applicable|no brand|unknown|generic|sans marque|-+)$", re.I) # Brand keys that are the same brand. Keys are brand_key() output. BRAND_ALIASES = { "kirkland": "kirklandsignature", "pc": "presidentschoice", "presidentchoice": "presidentschoice", "lechoixdupresident": "presidentschoice", "pcblacklabel": "presidentschoice", "on": "optimumnutrition", "optimumnutritionon": "optimumnutrition", "fairlifellc": "fairlife", "corepowerbyfairlife": "corepower", "fairlifecorepower": "corepower", "premier": "premierprotein", "premiernutrition": "premierprotein", "musclemilkbycytosport": "musclemilk", "cytosport": "musclemilk", "greatvaluewalmart": "greatvalue", "nonamebrand": "noname", "365": "365wholefoodsmarket", "365everydayvalue": "365wholefoodsmarket", "365bywholefoodsmarket": "365wholefoodsmarket", "wholefoodsmarket": "365wholefoodsmarket", "traderjoe": "traderjoes", "kraftheinz": "kraft", "thekraftheinzcompany": "kraft", "quest": "questnutrition", "questbar": "questnutrition", # USDA files some Quest bars under the brand "Quest Bar" "ghostwhey": "ghost", "premierprotien": "premierprotein", # typos of a top brand, seen in the data "premiereprotein": "premierprotein", "bodyarmor": "bodyarmour", "mccain": "mccainfoods", } def tidy(s: str) -> str: return re.sub(r"\s+", " ", s or "").strip() # A "!" standing in for an "i" ("Fa!rlife", "N!ck's", "m!lk"): only between letters, before a # lower-case one, so "Yahoo!" or "Go!Smart" are left alone. BANG_I = re.compile(r"(?<=[A-Za-z])!(?=[a-z])") def bang_as_i(s: str) -> str: return BANG_I.sub("i", s) def clean_brand(raw: str) -> str | None: """First brand of a comma list, no ®/™, no Inc./LLC; None for placeholders.""" if not raw: return None s = bang_as_i(tidy(re.sub(r"[®™©]", "", raw.split(",")[0]))) if not s or PLACEHOLDER.match(s): return None for _ in range(3): if LEGAL.search(s): s = LEGAL.sub("", s).strip() if not s: return None if not any(ch.islower() for ch in s): # ALL CAPS -> Title Case s = " ".join(w.capitalize() for w in s.lower().split(" ")) return s[:80] def brand_key(brand: str | None) -> str: if not brand: return "" k = "".join(tokenize(brand.replace("&", " and "))) k = re.sub(r"^the", "", k) return BRAND_ALIASES.get(k, k) def name_key(name: str, brand: str | None) -> str: """The name's words without the brand in front ("Premier Protein Shake" under Premier Protein -> "shake"). The brand counts however it is spelt or glued: its key ("Premier" is Premier Protein, "KitKat" is Kit Kat, "ARBY'S Sauce" under Arbys), or its own words with or without "&" as "and". The longest such start is dropped, so two spellings of one brand give the same key.""" toks = tokenize(name) if not brand: return " ".join(toks) starts = {brand_key(brand), "".join(tokenize(brand)), "".join(tokenize(brand.replace("&", " and ")))} - {""} for i in range(len(toks), 0, -1): if "".join(toks[:i]) in starts: return " ".join(toks[i:]) return " ".join(toks) def clean_name(s: str) -> str | None: s = tidy(re.sub(r"[®™©]", "", s or "")) if len(s) < 2: return None if not any(ch.islower() for ch in s) and any(ch.isalpha() for ch in s): s = s.lower().capitalize() return s[:120] JUST_AN_AMOUNT = re.compile(r"^\s*\d+(?:[.,]\d+)?\s*(?:g|gr|grams?|ml|l|oz)\s*$", re.I) def num(v: str) -> float | None: if v is None: return None v = v.strip().replace(",", ".") if not v: return None try: n = float(v) except ValueError: return None return n if n == n and abs(n) != float("inf") else None def g100(v: str) -> float | None: n = num(v) return round(n, 2) if n is not None and 0 <= n <= 100 else None def r1(n: float | None) -> float | None: return None if n is None else round(n, 1) # --- sources -------------------------------------------------------------------------------------- class Food: __slots__ = ( "src", "src_id", "name", "brand", "codes", "kcal", "protein", "carbs", "fat", "fiber", "sugar", "sat_fat", "sodium", "serving_g", "serving_label", "countries", "scans", "region", "protein_product", "in_off", "in_usda", "main", "off_extras", ) def __init__(self, **kw): for k in self.__slots__: setattr(self, k, kw.get(k)) self.scans = self.scans or 0 self.region = self.region or 0 def has_off_extras(self) -> bool: """USDA's numbers with a detail Open Food Facts filled in (merge_by_barcode). getattr: a parse cache written before this field existed has no value for it.""" return bool(getattr(self, "off_extras", False)) def filled(self) -> int: return sum(x is not None for x in (self.fiber, self.sugar, self.sat_fat, self.sodium, self.serving_g)) def read_usda_categories(zip_path: str | None) -> dict[str, str]: if not zip_path: return {} cats: dict[str, str] = {} with zipfile.ZipFile(zip_path) as z: member = next(n for n in z.namelist() if n.endswith("/branded_food.csv")) with z.open(member) as fh: r = csv.DictReader(io.TextIOWrapper(fh, encoding="utf-8", newline="")) for row in r: c = row.get("branded_food_category") or "" if c: cats[row["fdc_id"]] = c return cats def read_usda(copy_csv: str, cats: dict[str, str], stats: Counter) -> list[Food]: out = [] with open(copy_csv, newline="", encoding="utf-8") as fh: for row in csv.DictReader(fh): stats["usda_read"] += 1 code = canonical_barcode(row["gtin"]) name = clean_name(row["name"]) if not code or not name: stats["usda_drop_code_or_name"] += 1 continue vals = {k: num(row[k + "_100g"]) for k in ("kcal", "protein", "carbs", "fat")} if any(v is None for v in vals.values()): stats["usda_drop_incomplete"] += 1 continue brand = clean_brand(row["brand"]) cat = cats.get(row["source_id"], "") sg = num(row["serving_g"]) label = tidy(row["serving_label"]) or None out.append(Food( src="usda", src_id=row["source_id"], name=name, brand=brand, codes={code}, kcal=r1(vals["kcal"]), protein=r1(vals["protein"]), carbs=r1(vals["carbs"]), fat=r1(vals["fat"]), fiber=r1(num(row["fibre_100g"])), sugar=r1(num(row["sugar_100g"])), sat_fat=r1(num(row["sat_fat_100g"])), sodium=r1(num(row["sodium_mg_100g"])), serving_g=sg if sg and 0 < sg <= 5000 else None, serving_label=None if not label or JUST_AN_AMOUNT.match(label) else label[:60], countries={"US"}, scans=0, region=0, protein_product=bool(PROTEIN_USDA_CATEGORIES.search(cat) or PROTEIN_NAME.search(name)), in_off=False, in_usda=True, )) stats["usda_kept"] += 1 return out def off_kcal(row: dict) -> float | None: k = num(row["energy-kcal_100g"]) if k is not None: return k kj = num(row["energy-kj_100g"]) or num(row["energy_100g"]) return kj / KJ_PER_KCAL if kj is not None else None OFF_FIELDS = [ "code", "product_name", "brands", "countries_tags", "serving_size", "serving_quantity", "unique_scans_n", "categories_tags", "energy-kcal_100g", "energy-kj_100g", "energy_100g", "fat_100g", "saturated-fat_100g", "carbohydrates_100g", "sugars_100g", "fiber_100g", "proteins_100g", "salt_100g", "sodium_100g", "alcohol_100g", "popularity_tags", ] OFF_NUTRIENTS = [f[: -len("_100g")] for f in OFF_FIELDS if f.endswith("_100g")] REGION_TAG = re.compile(r"top-(\d+)-(us|ca)-scans-(\d{4})") def region_popularity(tags: str) -> int: """How popular a product is in Canada/the US, from OFF's yearly per-country scan ranks (popularity_tags such as top-1000-ca-scans-2025). unique_scans_n counts every country, so a French biscuit with one Canadian scan would otherwise outrank a Canadian staple. top-1000 -> 1000, top-100000 -> 10, none -> 0; only the last three years count.""" best = None year_now = int(time.strftime("%Y")) for m in REGION_TAG.finditer(tags or ""): if int(m.group(3)) >= year_now - 3: n = int(m.group(1)) best = n if best is None else min(best, n) return round(1_000_000 / best) if best else 0 def off_rows_csv(path: str, stats: Counter): """The nightly CSV export. WARNING (checked 2026-10-03): it leaves the nutrition columns empty for many products whose numbers the API and the Parquet export do have, so it undercounts.""" with gzip.open(path, "rt", encoding="utf-8", errors="replace", newline="") as fh: r = csv.reader(fh, delimiter="\t", quoting=csv.QUOTE_NONE) header = next(r) ix = {h: i for i, h in enumerate(header)} missing = [n for n in OFF_FIELDS if n not in ix] if missing: raise SystemExit(f"Open Food Facts export is missing columns: {missing}") width = len(header) for fields in r: stats["off_read"] += 1 if len(fields) != width: stats["off_drop_bad_line"] += 1 continue yield {n: fields[ix[n]] for n in OFF_FIELDS} def off_rows_parquet(path: str, stats: Counter): """The Parquet export (huggingface.co/datasets/openfoodfacts/product-database, food.parquet). Needs `pip install duckdb`. Only US/CA, non-obsolete rows leave DuckDB.""" import duckdb # optional dependency, only for this path def nutrient(n: str) -> str: return f"list_filter(nutriments, x -> x.name = '{n}')[1].\"100g\" AS \"{n}_100g\"" sql = f""" SELECT code, coalesce(list_filter(product_name, x -> x.lang = 'en')[1].text, list_filter(product_name, x -> x.lang = 'main')[1].text, product_name[1].text) AS product_name, brands, array_to_string(countries_tags, ',') AS countries_tags, serving_size, serving_quantity, unique_scans_n, array_to_string(categories_tags, ',') AS categories_tags, array_to_string(popularity_tags, ',') AS popularity_tags, {", ".join(nutrient(n) for n in OFF_NUTRIENTS)} FROM read_parquet(?) WHERE list_has_any(countries_tags, ['en:united-states', 'en:canada']) AND coalesce(obsolete, false) = false """ con = duckdb.connect() stats["off_read"] = con.execute("SELECT count(*) FROM read_parquet(?)", [path]).fetchone()[0] cur = con.execute(sql, [path]) cols = [d[0] for d in cur.description] while True: batch = cur.fetchmany(20000) if not batch: break for t in batch: yield {k: ("" if v is None else str(v)) for k, v in zip(cols, t)} def off_rows_jsonl(path: str, stats: Counter): """The full JSONL export (static.openfoodfacts.org/data/openfoodfacts-products.jsonl.gz, ~13 GB), or a pre-filtered copy holding only lines that mention en:united-states / en:canada (README). This is the complete source: the CSV export drops numbers the JSONL has. A label typed per serving only (common on US products) has `_serving` but no `_100g` when OFF never worked out 100 g; then, if the serving weight is known in g or ml, the per-100 value is serving value x 100 / serving grams (the same sum OFF itself does), counted in stats.""" import json with gzip.open(path, "rt", encoding="utf-8", errors="replace") as fh: for line in fh: stats["off_read"] += 1 try: p = json.loads(line) except ValueError: stats["off_drop_bad_line"] += 1 continue if p.get("obsolete") in (True, "on", 1): stats["off_drop_obsolete"] += 1 continue n = p.get("nutriments") or {} sq = num(str(p.get("serving_quantity") or "")) sq_unit = str(p.get("serving_quantity_unit") or "g").lower() row = { "code": str(p.get("code") or ""), "product_name": p.get("product_name_en") or p.get("product_name") or "", "brands": p.get("brands") or "", "countries_tags": ",".join(p.get("countries_tags") or []), "serving_size": p.get("serving_size") or "", "serving_quantity": "" if sq is None else str(sq), "unique_scans_n": str(p.get("unique_scans_n") or 0), "categories_tags": ",".join(p.get("categories_tags") or []), "popularity_tags": ",".join(p.get("popularity_tags") or []), } # Products saved since OFF's 2026 schema change (schema_version 1003) keep their label in # nutrition.aggregated_set and leave `nutriments` empty: without this, almost every # Canadian product reads as "no nutrition" (checked 2026-10-03). agg = (p.get("nutrition") or {}).get("aggregated_set") or {} agg_ok = agg.get("preparation") in (None, "as_sold") and agg.get("per") in ("100g", "100ml") agg_n = (agg.get("nutrients") or {}) if agg_ok else {} derived = False used_agg = False for nut in OFF_NUTRIENTS: v = n.get(nut + "_100g") if v in (None, "") and nut in agg_n: a = agg_n[nut] av = a.get("value") unit = (a.get("unit") or "").lower() if isinstance(av, (int, float)) and unit in ("g", "kcal", "kj", "% vol", "%", ""): v = av used_agg = True elif isinstance(av, (int, float)) and unit == "mg": v = av / 1000.0 used_agg = True if v in (None, "") and sq and sq > 0 and sq_unit in ("g", "ml"): vs = n.get(nut + "_serving") vs = num(str(vs)) if vs not in (None, "") else None if vs is not None: v = vs * 100.0 / sq derived = True row[nut + "_100g"] = "" if v in (None, "") else str(v) if derived: row["_derived_from_serving"] = "1" if used_agg: row["_from_aggregated_set"] = "1" yield row def read_off(path: str, stats: Counter, extra: dict) -> list[Food]: out = [] by_country = Counter() if path.endswith(".parquet"): rows = off_rows_parquet(path, stats) elif ".jsonl" in path: rows = off_rows_jsonl(path, stats) else: rows = off_rows_csv(path, stats) for row in rows: ctags = row["countries_tags"] countries = set() if "en:united-states" in ctags: countries.add("US") if "en:canada" in ctags: countries.add("CA") if not countries: continue stats["off_us_ca"] += 1 for c in countries: by_country[c] += 1 code = canonical_barcode(row["code"]) name = clean_name(row["product_name"]) if not code or not name: stats["off_drop_code_or_name"] += 1 continue kcal = off_kcal(row) p, c, f = g100(row["proteins_100g"]), g100(row["carbohydrates_100g"]), g100(row["fat_100g"]) if kcal is None or p is None or c is None or f is None: stats["off_drop_incomplete"] += 1 continue for cc in countries: by_country[cc + "_complete"] += 1 if kcal < 0 or kcal > MAX_KCAL_100 or p + c + f > 105: stats["off_drop_bad_numbers"] += 1 continue fib = g100(row["fiber_100g"]) alcohol = (g100(row["alcohol_100g"]) or 0.0) * 0.789 # OFF stores % vol atw = 4 * p + 4 * c + 9 * f + 7 * alcohol # Both ways: kJ typed as kcal (4.2x too high) and per-serving numbers typed as per-100 g. if kcal < 0.7 * (atw - 4 * (fib or 0)) - 30 or kcal > 1.3 * atw + 30: stats["off_drop_calorie_mismatch"] += 1 continue sodium_g = g100(row["sodium_100g"]) if sodium_g is None: salt = g100(row["salt_100g"]) sodium_g = salt / SALT_PER_SODIUM if salt is not None else None sq = num(row["serving_quantity"]) label = tidy(row["serving_size"]) or None scans = int(num(row["unique_scans_n"]) or 0) ctag = row["categories_tags"] out.append(Food( src="off", src_id=code, name=name, brand=clean_brand(row["brands"]), codes={code}, kcal=r1(kcal), protein=r1(p), carbs=r1(c), fat=r1(f), fiber=r1(fib), sugar=r1(g100(row["sugars_100g"])), sat_fat=r1(g100(row["saturated-fat_100g"])), sodium=r1(sodium_g * 1000) if sodium_g is not None else None, serving_g=sq if sq and 0 < sq <= 5000 else None, serving_label=None if not label or JUST_AN_AMOUNT.match(label) else label[:60], countries=countries, scans=scans, region=region_popularity(row["popularity_tags"]), protein_product=any(t in ctag for t in PROTEIN_OFF_TAGS) or bool(PROTEIN_NAME.search(name)), in_off=True, in_usda=False, )) stats["off_kept"] += 1 if row.get("_derived_from_serving"): stats["off_kept_per100_from_serving"] += 1 if row.get("_from_aggregated_set"): stats["off_kept_from_new_schema"] += 1 extra["off_by_country"] = dict(by_country) return out # --- numbers that can't be right ------------------------------------------------------------------ # The calorie check (read_off) passes a row whose protein is 0 when carbs and fat explain the # calories, and a row whose four numbers are all 0. Both are real label shapes (water, a diet drink, # hot sauce, a candy with no protein), so these rules only fire when the NAME says the food has # energy or protein. Checked 2026-10-04 on the 450,000 shipped rows: 772 + 44 flagged by an early # version; the lists below are narrower. Counted in stats. ENERGY_FOOD = re.compile( r"\b(bars?|protein|whey|chicken|beef|pork|turkey|cheeses?|breads?|cookies?|crackers?|granola|cereals?|" r"oats|oatmeal|pasta|noodles|muffins?|pizzas?|burgers?|jerky|peanut butter|almond butter|nut butter|" r"yogh?urts?|brownies?|cakes?|bagels?|tortillas?|chips|almonds|peanuts|cashews|nuts)\b", re.I) # Words that make a real 0/0/0/0 likely: drinks, condiments, sprays, seasonings. ZERO_IS_REAL = re.compile( r"\b(water|waters|diet|zero|sugar[- ]?free|calorie[- ]?free|unsweetened|sparkling|seltzer|soda|tea|" r"coffee|sweetener|stevia|vinegar|mustard|sauce|salsa|seasoning|spice|extract|drops|" r"enhancer|gum|mints?|broth|bouillon|stock|yeast|gelatin|spray|dressing|pickles?|brine|" r"drink mix|energy drink|electrolyte)\b", re.I) # Foods that are protein by name: 0 g protein at 100+ kcal per 100 g is a typing slip. PROTEIN_BY_NAME = re.compile(r"\b(protein|whey isolate|whey protein|beef jerky|turkey jerky|(peanut|almond|cashew|nut) butter)\b", re.I) # ...unless the name says otherwise, or the nut butter is only a flavour (candy, creamer, cookies). PROTEIN_ZERO_OK = re.compile( r"\b(low[- ]protein|protein[- ]free|no protein|cups?|candy|candies|fudge|cookies?|creamer|topping|" r"pudding|bites|trees|bats|pumpkins|eggs|hearts|bells|filled|swirl|frosting|flavou?red)\b", re.I) def implausible(f: "Food") -> str | None: """Why `f`'s numbers can't be right for what its name says, or None.""" text = f"{f.name} {f.brand or ''}" if f.kcal == 0 and f.protein == 0 and f.carbs == 0 and f.fat == 0: if ENERGY_FOOD.search(text) and not ZERO_IS_REAL.search(f.name): return "all_zero" if f.protein == 0 and (f.kcal or 0) >= 100 and PROTEIN_BY_NAME.search(f.name) and not PROTEIN_ZERO_OK.search(f.name): return "no_protein" return None def drop_implausible(foods: list["Food"], stats: Counter) -> list["Food"]: """`foods` without the rows `implausible` flags (before merging, so a good row from the other source can stand in for that barcode).""" out = [] for f in foods: why = implausible(f) if why: stats[f"{f.src}_drop_{why}"] += 1 else: out.append(f) return out # --- merge, fold, rank ---------------------------------------------------------------------------- # What Open Food Facts may fill in on a USDA row that left it empty. OFF_FILLS = ("fiber", "sugar", "sat_fat", "sodium", "serving_g", "serving_label", "brand") def merge_by_barcode(usda: list[Food], off: list[Food], stats: Counter) -> list[Food]: by_code: dict[str, Food] = {} for f in usda: code = next(iter(f.codes)) by_code[code] = f for o in off: code = next(iter(o.codes)) u = by_code.get(code) if u is None: by_code[code] = o continue if u.in_off: # a second OFF row with the same canonical code (12 vs 13 digits) u.scans += o.scans u.region = max(u.region, o.region) stats["off_dup_code"] += 1 continue # USDA numbers win (manufacturer data); OFF adds scans, countries, missing extras. stats["merged_usda_off"] += 1 u.in_off = True u.scans += o.scans u.region = max(u.region, o.region) u.countries |= o.countries u.protein_product = u.protein_product or o.protein_product for k in OFF_FILLS: if getattr(u, k) is None and getattr(o, k) is not None: setattr(u, k, getattr(o, k)) u.off_extras = True if u.has_off_extras(): stats["usda_rows_filled_from_off"] += 1 return list(by_code.values()) def close(a: float | None, b: float | None, rel: float, floor: float) -> bool: if a is None or b is None: return a is None and b is None return abs(a - b) <= rel * max(min(abs(a), abs(b)), floor) # The name key of a food whose name IS its brand ("Pepsi" by Pepsi, "premier protein" by Premier # Protein): such rows fold with each other like any same-name rows. NAME_IS_BRAND = "\u0000brand" def fold_key(f: Food) -> tuple[str, str] | None: """(brand key, name key) that pack sizes share, or None for a row that never folds: a name with no a-z/0-9 words at all (an Arabic or Chinese name), which says nothing to compare.""" nk = name_key(f.name, f.brand) if nk: return (brand_key(f.brand), nk) bk = brand_key(f.brand) return (bk, NAME_IS_BRAND) if bk and tokenize(f.name) else None def fold_pack_sizes(foods: list[Food], stats: Counter) -> list[Food]: groups: dict[tuple[str, str], list[Food]] = defaultdict(list) out = [] for f in foods: key = fold_key(f) if key is None: out.append(f) else: groups[key].append(f) for rows in groups.values(): rows.sort(key=lambda f: (-f.region, -(f.scans or 0), -f.in_usda, -f.filled())) kept: list[Food] = [] for f in rows: for k in kept: if close(k.kcal, f.kcal, 0.03, 5) and close(k.protein, f.protein, 0.05, 1): k.codes |= f.codes k.scans += f.scans k.region = max(k.region, f.region) k.countries |= f.countries k.in_off = k.in_off or f.in_off k.in_usda = k.in_usda or f.in_usda k.protein_product = k.protein_product or f.protein_product stats["folded_pack_sizes"] += 1 break else: kept.append(f) out.extend(kept) return out def score(f: Food) -> int: """The rank's first key: regional rank tier first, then raw scans (capped) inside a tier.""" return f.region * 1000 + min(f.scans, 999) def rank(foods: list[Food]) -> list[Food]: brand_pop: Counter = Counter() for f in foods: brand_pop[brand_key(f.brand)] += f.region brand_pop[""] = 0 return sorted( foods, key=lambda f: (-score(f), -brand_pop[brand_key(f.brand)], -("CA" in f.countries), -f.filled(), f.name), ) def pick_top(ranked: list[Food], n: int) -> list[Food]: protein = [f for f in ranked if f.protein_product] pset = {id(f) for f in protein} rest = [f for f in ranked if id(f) not in pset] chosen = protein[:n] + rest[: max(0, n - len(protein))] return rank(chosen) # --- brand display names -------------------------------------------------------------------------- def brand_display(foods: list[Food]) -> dict[str, str]: """For each brand key, the spelling used on the most-scanned (then most) products.""" votes: dict[str, Counter] = defaultdict(Counter) for f in foods: if f.brand: votes[brand_key(f.brand)][f.brand] += 1 + f.scans return {k: c.most_common(1)[0][0] for k, c in votes.items()} # --- write ---------------------------------------------------------------------------------------- SCHEMA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "schema.sql") def schema() -> str: """The file's shape, shared with the app's Jest fixture (one copy, schema.sql).""" with open(SCHEMA_PATH, encoding="utf-8") as fh: return fh.read() # Extra search words for brands people type short ("pc protein"), keyed by brand_key. The app ranks # them as naming the brand (brandsDb.ts BRAND_SHORT_FORMS, keyed by joinedWord: the test checks the # two tables are the same). BRAND_SEARCH_EXTRA = {"presidentschoice": ["pc"], "365wholefoodsmarket": ["365"]} def index_text(brand: str | None, name: str, extra: list[str] | tuple = ()) -> str: """The words the word index stores for one food, exactly as the app's brandIndexText makes them (app/src/features/foods/brandsDb.ts): index_words of the brand and of the name, the brand as one word ("bfit" for B Fit, "corepower" for Core Power), then `extra`. Lower case a-z/0-9 words, one space apart, so FTS5's tokenizer reads exactly these.""" words = dict.fromkeys(index_words(brand or "")) for w in index_words(name): words.setdefault(w) glued = joined_word(brand or "") if glued: words.setdefault(glued) for e in extra: for w in index_words(e): words.setdefault(w) return " ".join(words) def search_text(brand: str | None, name: str, own_brand: str | None = None) -> str: """index_text with the brand's short forms (BRAND_SEARCH_EXTRA); the food's own brand spelling when the file shows its brand's main one ("Quest Bar" rows show Quest but still answer "quest bar"); and the name with each "!" that stands for an "i" read as one ("Oat m!lk" also finds "oat milk"; the name shown keeps it).""" extra = list(BRAND_SEARCH_EXTRA.get(brand_key(brand), [])) if own_brand and own_brand != brand: extra.append(own_brand) if bang_as_i(name) != name: extra.append(bang_as_i(name)) return index_text(brand, name, extra) # brand_foods.from_off: where a row came from (schema.sql; brandsDb.ts rowToBrandFood reads it). FROM_USDA = 0 FROM_OFF = 1 FROM_USDA_AND_OFF = 2 def source_code(f: Food) -> int: """FROM_OFF for Open Food Facts' numbers; FROM_USDA_AND_OFF for USDA's numbers with a detail (serving, fibre, ...) Open Food Facts filled in; FROM_USDA when every value is USDA's.""" if f.src == "off": return FROM_OFF return FROM_USDA_AND_OFF if f.has_off_extras() else FROM_USDA def x10(v: float | None) -> int | None: """A number with one decimal as whole tenths (52.3 -> 523): the file's storage form.""" return None if v is None else int(round(v * 10)) def main_code(f: Food) -> str: """The barcode a food is known by: the code of the row that was kept when pack sizes folded (the most scanned pack), else the lowest of its codes.""" m = getattr(f, "main", None) return m if m in f.codes else min(f.codes, key=barcode_int) def write_db(path: str, foods: list[Food], brands: dict[str, str], fts: bool, built_from: dict, version: int = 0) -> None: """Writes `path` (schema.sql), then VACUUMs and checks it. Call only with a path in a scratch folder: SQLite never writes on the Documents mount (the repo gets the .gz from write_gz).""" if os.path.exists(path): os.remove(path) db = sqlite3.connect(path) db.executescript("PRAGMA page_size=4096; PRAGMA journal_mode=DELETE;" + schema()) rows, codes, words = [], [], [] for i, f in enumerate(foods, start=1): brand = brands.get(brand_key(f.brand), f.brand) if f.brand else None rows.append(( i, barcode_int(main_code(f)), f.name, brand, x10(f.kcal), x10(f.protein), x10(f.carbs), x10(f.fat), x10(f.fiber), x10(f.sugar), x10(f.sat_fat), x10(f.sodium), x10(f.serving_g), f.serving_label, source_code(f), )) words.append((i, search_text(brand, f.name, f.brand))) for c in f.codes: codes.append((barcode_int(c), i)) db.executemany("INSERT INTO brand_foods VALUES (" + ",".join("?" * 15) + ")", rows) db.executemany("INSERT OR IGNORE INTO brand_barcodes VALUES (?,?)", codes) if fts: # Contentless, no positions, no per-row sizes: the app only asks which rows have the # words, then reads them in rank order. The 1- and 2-letter prefix lists (schema.sql) cost # 2.4 MB zipped at 450k rows and take the slowest search from 59 ms to 6.5 ms on a Mac # (measured 2026-10-04; a short last word like "oikos p" was 24 ms without them). db.executemany("INSERT INTO brand_fts(rowid, search) VALUES (?,?)", words) db.execute("INSERT INTO brand_fts(brand_fts) VALUES('optimize')") else: db.execute("DROP TABLE brand_fts") meta = { "licence": FILE_LICENCE, "attribution": ATTRIBUTION, "built": time.strftime("%Y-%m-%d"), "built_from": json.dumps(built_from), "version": str(version), "rows": str(len(rows)), "barcodes": str(len(codes)), "units": "Per 100 g (ml counted as g), stored as whole tenths: kcal_x10 = 523 is 52.3 kcal. " "View brand_foods_per_100g shows plain numbers.", "method": "app/scripts/build-brands/build_brands.py and schema.sql (published with the database, ODbL 4.6)", } db.executemany("INSERT INTO meta VALUES (?,?)", meta.items()) db.execute(f"PRAGMA user_version={int(version)}") db.commit() db.execute("VACUUM") check = db.execute("PRAGMA integrity_check").fetchall() if check != [("ok",)]: db.close() raise SystemExit(f"{path} failed PRAGMA integrity_check: {check[:5]}") if fts: db.execute("INSERT INTO brand_fts(brand_fts) VALUES('integrity-check')") mode = db.execute("PRAGMA journal_mode").fetchone()[0] db.close() if mode != "delete": # A WAL file can't be opened read-only from inside the app bundle. raise SystemExit(f"{path} is in {mode} mode; the app needs a rollback-journal file") def file_sha256(path: str) -> str: import hashlib h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def write_gz(db_path: str, gz_path: str) -> dict: """The repo copy: `gz_path` (gzip -9, no timestamp, so the same file gives the same bytes) and, beside it, `.sha256` with the sha256 of the UNZIPPED file in `shasum -c` form, which the app's postinstall step checks. Both are written to a temp name, then moved.""" import shutil sha_path = gz_path[: -len(".gz")] + ".sha256" if gz_path.endswith(".gz") else gz_path + ".sha256" tmp = gz_path + ".tmp" with open(db_path, "rb") as src, open(tmp, "wb") as raw: with gzip.GzipFile(filename="brands.db", mode="wb", compresslevel=9, fileobj=raw, mtime=0) as gz: shutil.copyfileobj(src, gz, 1 << 20) os.replace(tmp, gz_path) sha = file_sha256(db_path) with open(sha_path + ".tmp", "w") as fh: fh.write(f"{sha} brands.db\n") os.replace(sha_path + ".tmp", sha_path) return {"gz": gz_path, "gz_bytes": os.path.getsize(gz_path), "sha256": sha, "sha_file": sha_path} def deflated_size(path: str) -> int: """What the file adds to an .ipa (a zip, deflate).""" with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp: zpath = tmp.name try: with zipfile.ZipFile(zpath, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as z: z.write(path, "brands.db") return os.path.getsize(zpath) finally: os.remove(zpath) BENCH_QUERIES = ["premier protein", "fairlife", "bfit", "b fit", "core power", "corepower", "muscle milk", "quest bar", "greek yogurt", "kirkland", "chicken", "rice", "pc protein", "oikos pro", "ghost whey", "kraft peanut butter", "kelloggs", "mcdonalds", "eggs", "a", "ch", "c h", "b f", "a b c d e"] # brandsDb.ts limits, mirrored: what is read of the box (UTF-16 units, as JavaScript counts), words # kept, letters per word, and the fewest letters worth a search. MAX_QUERY_CHARS = 200 MAX_QUERY_TOKENS = 5 MAX_TOKEN_LETTERS = 40 MIN_QUERY_LETTERS = 2 def _js_slice(text: str, units: int) -> str: """text.slice(0, units) as JavaScript does it: counted in UTF-16 code units. A character cut in half is dropped (it is never a-z or 0-9, so the words come out the same).""" return text.encode("utf-16-le", "surrogatepass")[: 2 * units].decode("utf-16-le", "ignore") def _match(toks: list[str], starts_only: int = -1) -> str: """Every word matches one of its word_forms: (form OR form ...) AND (...). Word `starts_only` (a glued run) keeps only its word-start forms, so a glued search never finds more than the plain one.""" parts = [] for i in range(len(toks)): terms = [f'"{text}"' if whole else f'"{text}"*' for text, whole in word_forms(toks, i) if not (i == starts_only and whole)] parts.append(terms[0] if len(terms) == 1 else f"({' OR '.join(terms)})") return " AND ".join(parts) def brand_query(q: str) -> dict | None: """Exact port of the app's brandQuery(q) (app/src/features/foods/brandsDb.ts): the words from search_tokens, each matching any of its word_forms, a word start as a prefix term ("x"*) and a whole word as an exact term ("x"); every word must match (`match`). `glued`: `match` narrowed to one run of typed words glued into one word ("b fit" -> "bfit"*), which the app also reads so a brand typed in pieces isn't crowded out by popular foods. None: nothing to search.""" toks = [re.sub(r"[^a-z0-9]", "", t)[:MAX_TOKEN_LETTERS] for t in search_tokens(_js_slice(q, MAX_QUERY_CHARS))] toks = [t for t in toks if t][:MAX_QUERY_TOKENS] if len("".join(toks)) < MIN_QUERY_LETTERS: return None glued = [_match(toks[:j] + ["".join(toks[j : k + 1])] + toks[k + 1 :], j) for j in range(len(toks)) for k in range(j + 1, len(toks))] return {"tokens": toks, "match": _match(toks), "glued": glued} def fts_match(q: str) -> str | None: """brand_query(q)["match"], or None.""" bq = brand_query(q) return bq["match"] if bq else None def bench(path: str) -> dict: """Search speed on this Mac, the app's way (brandsDb.ts search): the word index for the first 200 matches in rank order, 50 more for each glued run, then those rows. A phone is slower; treat these as relative numbers.""" db = sqlite3.connect(f"file:{path}?mode=ro", uri=True) has_fts = db.execute("SELECT 1 FROM sqlite_master WHERE name='brand_fts'").fetchone() is not None out = {} for q in BENCH_QUERIES: bq = brand_query(q) if not has_fts or not bq: continue t = time.perf_counter() ids = [r[0] for r in db.execute("SELECT rowid FROM brand_fts WHERE brand_fts MATCH ? LIMIT 200", [bq["match"]])] for g in bq["glued"] if ids else []: ids += [r[0] for r in db.execute("SELECT rowid FROM brand_fts WHERE brand_fts MATCH ? LIMIT 50", [g]) if r[0] not in ids] rows = db.execute( f"SELECT id, brand, name, kcal_x10 / 10.0, protein_x10 / 10.0 FROM brand_foods WHERE id IN ({','.join('?' * len(ids))}) ORDER BY id", ids, ).fetchall() if ids else [] ms = (time.perf_counter() - t) * 1000 out[q] = {"hits": len(ids), "ms": round(ms, 2), "top": rows[:3]} db.close() return out def main() -> None: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--usda-copy", required=True) ap.add_argument("--usda-zip") ap.add_argument("--off", required=True) ap.add_argument("--top", type=int, default=450000) ap.add_argument("--out", required=True) ap.add_argument("--fts", action="store_true", help="also add an FTS5 word index (measured, not required)") ap.add_argument("--measure", help="comma list of N to size, e.g. 50000,100000,150000") ap.add_argument("--stats", help="write counts and sizes as JSON here") ap.add_argument("--cache", help="pickle of the parsed sources; reused when newer than every input") ap.add_argument("--gz", help="also write this gzip copy (+ .sha256 beside it) for the repo, e.g. app/assets/brands.db.gz") ap.add_argument("--version", type=int, default=int(time.strftime("%Y%m%d")) * 100 + 1, help="the file's PRAGMA user_version, YYYYMMDDNN (default: today, build 01)") args = ap.parse_args() if args.gz and not args.fts: raise SystemExit("--gz is the app's copy, and the app searches with the word index: add --fts") t0 = time.time() stats: Counter = Counter() extra: dict = {} inputs = [x for x in (args.usda_copy, args.usda_zip, args.off) if x] if args.cache and os.path.exists(args.cache) and all( os.path.getmtime(args.cache) > os.path.getmtime(x) for x in inputs ): print(f"Parsed sources from {args.cache}", flush=True) with open(args.cache, "rb") as fh: usda, off, saved_stats, extra = pickle.load(fh) stats.update(saved_stats) else: print("USDA categories ...", flush=True) cats = read_usda_categories(args.usda_zip) print("USDA rows ...", flush=True) usda = read_usda(args.usda_copy, cats, stats) print(f" {len(usda)} kept", flush=True) print("Open Food Facts ...", flush=True) off = read_off(args.off, stats, extra) print(f" {len(off)} kept", flush=True) if args.cache: with open(args.cache, "wb") as fh: pickle.dump((usda, off, stats, extra), fh, protocol=pickle.HIGHEST_PROTOCOL) usda = drop_implausible(usda, stats) off = drop_implausible(off, stats) for f in (*usda, *off): f.main = next(iter(f.codes)) # its own code, before merging and folding add others f.off_extras = False # clean_brand's later rules, for a parse cache written before them (no-op on a fresh read). f.brand = bang_as_i(f.brand) if f.brand else f.brand merged = merge_by_barcode(usda, off, stats) folded = fold_pack_sizes(merged, stats) ranked = rank(folded) brands = brand_display(ranked) stats["after_merge"] = len(merged) stats["after_fold"] = len(folded) stats["protein_products"] = sum(1 for f in ranked if f.protein_product) stats["with_scans"] = sum(1 for f in ranked if f.scans > 0) stats["with_region_rank"] = sum(1 for f in ranked if f.region > 0) stats["brands"] = len(brands) stats["ca_rows"] = sum(1 for f in ranked if "CA" in f.countries) stats["us_rows"] = sum(1 for f in ranked if "US" in f.countries) built_from = { "usda_copy": os.path.basename(args.usda_copy), "usda_zip": os.path.basename(args.usda_zip or ""), "off": os.path.basename(args.off), "off_mtime": time.strftime("%Y-%m-%d", time.gmtime(os.path.getmtime(args.off))), } sizes = [] ns = sorted({int(x) for x in args.measure.split(",")}) if args.measure else [] for n in ns: for fts in (False, True): sub = pick_top(ranked, n) with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as tmp: p = tmp.name write_db(p, sub, brands, fts, built_from, args.version) s = { "n": len(sub), "fts": fts, "bytes": os.path.getsize(p), "deflated": deflated_size(p), "barcodes": sum(len(f.codes) for f in sub), "protein": sum(1 for f in sub if f.protein_product), "ca": sum(1 for f in sub if "CA" in f.countries), "off_rows": sum(1 for f in sub if f.src == "off"), "usda_rows_with_off_details": sum(1 for f in sub if source_code(f) == FROM_USDA_AND_OFF), "min_popularity": min((score(f) for f in sub), default=0), "usda_only_rows": sum(1 for f in sub if not f.in_off), } os.remove(p) sizes.append(s) print(f" N={s['n']:>7} fts={fts!s:5} {s['bytes']/1e6:7.1f} MB raw, {s['deflated']/1e6:6.1f} MB in the .ipa", flush=True) top = pick_top(ranked, args.top) write_db(args.out, top, brands, args.fts, built_from, args.version) final = {"n": len(top), "bytes": os.path.getsize(args.out), "deflated": deflated_size(args.out), "version": args.version, "sources": dict(Counter({FROM_USDA: "usda", FROM_OFF: "off", FROM_USDA_AND_OFF: "usda_and_off"}[source_code(f)] for f in top))} print(f"Wrote {args.out}: {final['n']} foods, {final['bytes']/1e6:.1f} MB, {final['deflated']/1e6:.1f} MB deflated " f"({time.time()-t0:.0f} s), integrity ok") if args.gz: final["gz"] = write_gz(args.out, args.gz) print(f"Wrote {args.gz}: {final['gz']['gz_bytes']} bytes; sha256 of brands.db {final['gz']['sha256']}") final["bench"] = bench(args.out) for q, r in final["bench"].items(): print(f" search {q!r}: {r['hits']} hits, {r['ms']} ms; top {[f'{b} | {n}' for _, b, n, *_ in r['top']]}") if args.stats: with open(args.stats, "w") as fh: json.dump({"counts": stats, **extra, "sizes": sizes, "final": final, "top_brands": Counter(brands.get(brand_key(f.brand), f.brand) for f in top if f.brand).most_common(40)}, fh, indent=2) if __name__ == "__main__": main()