#!/usr/bin/env python3 """Builds the packaged-food files the barcode scanner looks up: products sold in India, the United States and the United Kingdom, from the Open Food Facts export, with their name, brand, pack and serving size and their label's values per 100 g. One pass over the export writes all three (docs/food-global.md). python3 -I program/foods/build_packaged_foods.py --export [--country en:india] python3 -I program/foods/build_packaged_foods.py --check --export is Open Food Facts' CSV export (https://static.openfoodfacts.org/data/en.openfoodfacts.org.products.csv.gz, about 1.5 GB, tab-separated), read as a stream; it isn't kept. Open Food Facts publishes a new one every night, so a rebuild gives a different file: new products, corrections, removals. That's expected, and the file records the export it came from. --check doesn't need the export: it checks the committed file is well formed (codes unique and sorted, values in range, counts and licence present) for every country's file. This script is the method of making the file (ODbL 1.0 section 4.6). What it keeps: - products whose countries include the country, with a barcode of 8, 12, 13 or 14 digits (codes starting with 2 are in-store codes, not unique, and are left out), a name with a letter in it, and energy, protein, carbohydrate and fat per 100 g; - values in range (energy 0 to 900 kcal, each nutrient 0 to 100 g, protein, carbohydrate and fat together at most 101 g) and energy within a quarter, or 25 kcal, of 4 kcal per gram of protein and carbohydrate, 9 per gram of fat and 7 per gram of alcohol; - none that Open Food Facts flags with a data-quality error; - for the United States and the United Kingdom, only products an Open Food Facts user scanned in that country in 2023 or later (the export's popularity tags, top-country-us-scans-2023 and later; the UK's tags use gb). Those countries list hundreds of thousands of products; scans are what people buy. Nearly every product has an edit since 2024, mostly by Open Food Facts' bots, so the date of the last edit isn't a filter. India keeps every complete product. Where a barcode is there twice, the one edited last is kept. Values are as contributors typed them from the label, rounded to 0.1 g; carbohydrate is the label's, which on Indian labels can include fibre. Writes packaged-.json and packaged--report.md: in (India), us (United States), gb (United Kingdom). """ import argparse import csv import datetime import gzip import json import re import sys from pathlib import Path HERE = Path(__file__).resolve().parent sys.dont_write_bytecode = True # Open Food Facts country tag: (file code, name, short name for the app's copy, ISO region, popularity code or None # to keep every complete product). COUNTRIES = { "en:india": ("in", "India", "India", "IN", None), "en:united-states": ("us", "United States", "the US", "US", "us"), "en:united-kingdom": ("gb", "United Kingdom", "the UK", "GB", "gb"), } SCANNED_SINCE = 2023 LICENCE = "https://opendatacommons.org/licenses/odbl/1-0/" CONTENTS = "https://opendatacommons.org/licenses/dbcl/1-0/" CREDITS = [ "Contains data from Open Food Facts, available under the Open Database License (ODbL) 1.0. " "© Open Food Facts contributors.", "Individual product values are under the Database Contents License (DbCL) 1.0.", ] LINKS = [ {"title": "Open Food Facts", "url": "https://world.openfoodfacts.org"}, {"title": "Open Database License", "url": LICENCE}, ] NAME_FIELDS = ("product_name", "product_name_en", "abbreviated_product_name", "generic_name") def num(v): try: x = float(v) except (TypeError, ValueError): return None return x if x == x and abs(x) != float("inf") else None def r1(x): """0.1 g; a whole number without its ".0", to keep the file small.""" x = round(x + 0.0, 1) return int(x) if x == int(x) else x def code_of(raw): """The barcode without leading zeros, as the app looks it up; None when it isn't a retail barcode.""" c = raw.strip() if not c.isdigit() or len(c) not in (8, 12, 13, 14): return None if len(c) == 13 and c.startswith("2"): return None c = c.lstrip("0") return c if len(c) >= 6 else None def clean_text(s, limit): s = re.sub(r"\s+", " ", (s or "").replace("\\", " ")).strip(" ,;-") return s[:limit].rstrip() if s else "" def diet_of(labels, analysis): """'vg' vegan, 'v' vegetarian, 'nv' non-vegetarian, from the pack's mark first, then the ingredients.""" if "en:non-vegetarian" in labels: return "nv" if "en:vegan" in labels: return "vg" if "en:vegetarian" in labels: return "v" if "en:non-vegetarian" in analysis: return "nv" if "en:vegan" in analysis: return "vg" if "en:vegetarian" in analysis: return "v" return None def scanned(tags, cc, since=SCANNED_SINCE): """True when the popularity tags say someone scanned it in the country (cc as Open Food Facts writes it) in `since` or later: top-country-us-scans-2024, top-50000-gb-scans-2025.""" for t in tags.split(","): m = re.fullmatch(rf"top-(?:country|\d+)-{cc}-scans-(\d{{4}})", t.strip()) if m and int(m.group(1)) >= since: return True return False def product(row, country): """One product for the file, or (None, reason) when it's left out.""" if country not in (row.get("countries_tags") or "").split(","): return None, "country" code = code_of(row.get("code", "")) if not code: return None, "barcode" names = (clean_text(row.get(k), 80) for k in NAME_FIELDS) name = next((n for n in names if re.search(r"[^\W\d_]", n)), "") if not name: return None, "name" if (row.get("data_quality_errors_tags") or "").strip(): return None, "quality" popular = COUNTRIES[country][4] if popular and not scanned(row.get("popularity_tags") or "", popular): return None, "scans" kcal = num(row.get("energy-kcal_100g")) if kcal is None and num(row.get("energy_100g")) is not None: kcal = num(row.get("energy_100g")) / 4.184 p, c, f = (num(row.get(k)) for k in ("proteins_100g", "carbohydrates_100g", "fat_100g")) if None in (kcal, p, c, f): return None, "nutrients" fib, alc = num(row.get("fiber_100g")), num(row.get("alcohol_100g")) vals = [p, c, f] + [x for x in (fib, alc) if x is not None] if not 0 <= kcal <= 900 or any(not 0 <= x <= 100 for x in vals) or p + c + f > 101: return None, "range" est = 4 * p + 4 * c + 9 * f + 7 * (alc or 0) if abs(kcal - est) > max(0.25 * max(kcal, est), 25): return None, "energy" out = {"b": code, "n": name} brand = clean_text((row.get("brands") or "").split(",")[0], 40) if brand and brand.lower() not in name.lower(): out["br"] = brand qty = clean_text(row.get("quantity"), 24) if qty: out["q"] = qty pack = num(row.get("product_quantity")) if pack and 1 <= pack <= 2000: out["pk"] = r1(pack) serving = num(row.get("serving_quantity")) if serving and 1 <= serving <= 1000 and (not pack or serving <= pack): out["s"] = r1(serving) out.update({"k": r1(kcal), "p": r1(p), "c": r1(c), "f": r1(f)}) if fib: out["fi"] = r1(fib) if alc: out["a"] = r1(alc) d = diet_of(set((row.get("labels_tags") or "").split(",")), set((row.get("ingredients_analysis_tags") or "").split(","))) if d: out["d"] = d return (out, int(row.get("last_modified_t") or 0)), None def rows(path): """The export's rows as dicts. A country extract written by an earlier run (tab-separated, backslash escapes) reads the same way.""" opener = gzip.open if str(path).endswith(".gz") else open with opener(path, "rt", encoding="utf-8", newline="") as fh: head = fh.readline().rstrip("\n").split("\t") extract = not str(path).endswith(".gz") r = csv.reader(fh, delimiter="\t", quoting=csv.QUOTE_NONE, escapechar="\\" if extract else None) for row in r: if len(row) == len(head): yield dict(zip(head, row)) def build(export, countries): """{country: (products, dropped, exported)} from one pass over the export. `exported` is the date of the export's newest edit, the same for every country.""" kept = {c: {} for c in countries} dropped = {c: {} for c in countries} newest = 0 for row in rows(export): newest = max(newest, int(row.get("last_modified_t") or 0)) for country in countries: got, why = product(row, country) if why == "country": continue if why: dropped[country][why] = dropped[country].get(why, 0) + 1 continue p, modified = got k = kept[country] if p["b"] not in k or modified > k[p["b"]][1]: if p["b"] in k: dropped[country]["duplicate"] = dropped[country].get("duplicate", 0) + 1 k[p["b"]] = (p, modified) else: dropped[country]["duplicate"] = dropped[country].get("duplicate", 0) + 1 exported = datetime.datetime.fromtimestamp(newest, datetime.UTC).date().isoformat() return {c: ([kept[c][k][0] for k in sorted(kept[c], key=lambda x: (len(x), x))], dropped[c], exported) for c in countries} def file_text(products, dropped, exported, country): code, name, short, region, popular = COUNTRIES[country] seen = len(products) + sum(dropped.values()) doc = { "kind": "packaged-foods", "schemaVersion": 1, "country": country, "countryName": name, "shortName": short, "region": region, "exported": exported, "source": { "name": "Open Food Facts", "citation": " ".join(CREDITS), "credits": CREDITS, "links": LINKS, "url": "https://world.openfoodfacts.org", "licence": f"Open Database License (ODbL) 1.0, {LICENCE}. Contents: Database Contents License (DbCL) " f"1.0, {CONTENTS}.", }, "method": f"Products sold in {name} from the Open Food Facts CSV export with edits up to {exported}: those " "with a retail barcode, a name and energy, protein, carbohydrate and fat per 100 g, in range and " "adding up, without a data-quality error" + (f", that an Open Food Facts user scanned in {name} in {SCANNED_SINCE} or later" if popular else "") + ". Made by program/foods/build_packaged_foods.py.", "notes": [ "Keys: b barcode without leading zeros, n name, br brand, q pack size as printed, pk pack grams, " "s serving grams, k kcal, p protein, c carbohydrate, f fat, fi fibre and a alcohol per 100 g, d the " "pack's mark (vg vegan, v vegetarian, nv non-vegetarian).", "Values are what Open Food Facts contributors typed from the label. They are estimates, and can be " "wrong or out of date.", ], "counts": {"products": len(products), "sold here": seen, "left out": dict(sorted(dropped.items()))}, "products": products, } text = json.dumps(doc, ensure_ascii=False, indent=1) # One product a line keeps the file small and diffs readable. head, _, _ = text.partition('"products": [') lines = ",\n".join(json.dumps(p, ensure_ascii=False, separators=(",", ":")) for p in products) return head + '"products": [\n' + lines + "\n ]\n}\n" def report_text(doc, size): c = doc["counts"] diets = {k: sum(1 for p in doc["products"] if p.get("d") == k) for k in ("vg", "v", "nv")} lines = [ f"# Packaged foods, {doc['countryName']}: build report", "", "Generated by `program/foods/build_packaged_foods.py` from the Open Food Facts CSV export. Don't edit by " f"hand. The app bundles `{Path(doc['_file']).name}` for the barcode scanner.", "", f"- Export: edits up to {doc['exported']}.", f"- Products sold in {doc['countryName']}: {c['sold here']}. Kept: {c['products']} " f"({size / 1e6:.2f} MB).", "- Left out: " + ", ".join(f"{v} {k}" for k, v in c["left out"].items()) + ".", f"- With a serving size: {sum(1 for p in doc['products'] if 's' in p)}; with a pack weight: " f"{sum(1 for p in doc['products'] if 'pk' in p)}.", f"- Marked vegan {diets['vg']}, vegetarian {diets['v']}, non-vegetarian {diets['nv']}; the rest unmarked.", "", "Reasons: `barcode` not a retail barcode; `name` no name; `nutrients` energy, protein, carbohydrate or " "fat missing; `range` a value out of range; `energy` energy doesn't match the nutrients; `quality` Open " "Food Facts flags a data-quality error; `scans` nobody scanned it in the country since " f"{SCANNED_SINCE} (the US and the UK only); `duplicate` the same barcode twice.", "", "## Licence", "", "The file is a Derivative Database of Open Food Facts under the ODbL 1.0, kept as its own file so the " "app's other food lists stay separate (a Collective Database, ODbL 4.5). It carries the licence and the " "credit; the Food guide and the support page show the credit with links. This script is the method of " "making it (ODbL 4.6), and must be offered with the app.", "", ] lines += [f"> {x}" for x in doc["source"]["credits"]] return "\n".join(lines) + "\n" def check(path): doc = json.loads(path.read_text(encoding="utf-8")) problems = [] if doc.get("kind") != "packaged-foods" or doc.get("schemaVersion") != 1: problems.append("kind or schemaVersion") if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", doc.get("exported", "")): problems.append("exported date") if LICENCE not in doc.get("source", {}).get("licence", ""): problems.append("licence") if doc.get("region") not in {c[3] for c in COUNTRIES.values()} or not doc.get("shortName"): problems.append("region or shortName") ps = doc.get("products", []) codes = [p.get("b", "") for p in ps] if codes != sorted(codes, key=lambda x: (len(x), x)) or len(set(codes)) != len(codes): problems.append("codes unsorted or repeated") for p in ps: ok = (p.get("b", "").isdigit() and not p["b"].startswith("0") and p.get("n") and all(isinstance(p.get(k), (int, float)) for k in ("k", "p", "c", "f")) and 0 <= p["k"] <= 900 and all(0 <= p[k] <= 100 for k in ("p", "c", "f"))) if not ok: problems.append(f"product {p.get('b')}") break if doc.get("counts", {}).get("products") != len(ps): problems.append("count") if problems: sys.exit(f"{path.name}: " + ", ".join(problems)) return doc def main(): ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) ap.add_argument("--export", type=Path, help="Open Food Facts CSV export (.csv.gz), or a country extract") ap.add_argument("--country", choices=sorted(COUNTRIES), help="one country only; all three by default") ap.add_argument("--check", action="store_true", help="check the committed files; needs no export") a = ap.parse_args() countries = [a.country] if a.country else list(COUNTRIES) paths = {c: (HERE / f"packaged-{COUNTRIES[c][0]}.json", HERE / f"packaged-{COUNTRIES[c][0]}-report.md") for c in countries} if a.check: for c in countries: doc = check(paths[c][0]) print(f"{paths[c][0].name} is well formed: {len(doc['products'])} products, export {doc['exported']}.") return if not a.export: ap.error("--export is needed to build; --check needs none") for c, (products, dropped, exported) in build(a.export, countries).items(): out, report = paths[c] text = file_text(products, dropped, exported, c) out.write_text(text, encoding="utf-8") doc = check(out) doc["_file"] = str(out) report.write_text(report_text(doc, len(text.encode("utf-8"))), encoding="utf-8") print(f"{doc['countryName']}: {len(products)} products, {len(text.encode('utf-8')) / 1e6:.2f} MB -> {out.name}") if __name__ == "__main__": main()