From 766a5739894f4e5fc5d881eaed9d21082c7b4ae4 Mon Sep 17 00:00:00 2001 From: John Doe <1115084741@qq.com> Date: Mon, 8 Jun 2026 07:01:04 +0000 Subject: [PATCH] =?UTF-8?q?feat(M2):=20Open=20Food=20Facts=20=E9=87=87?= =?UTF-8?q?=E9=9B=86=E5=AF=BC=E5=85=A5=20ETL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - adapters/openfoodfacts.py: OFF API 适配器(限速+User-Agent) + JSONL/.gz dump 读取 - etl/transform.py: 纯函数转换(GTIN 校验/净含量解析+归一/营养 per_100g 双能量/过敏原添加剂清洗/关键词分类映射) - etl/load.py: psycopg upsert(product/food_detail/product_image) + product_source 字段级溯源 - jobs/seed_off.py: CLI(--barcodes API / --dump 文件 / --limit) - 测试: 13 个离线 transform 单测(fixture) + 可跳过的 DB 集成测试 - 依赖: 增加 psycopg[binary] - 实跑: 从 OFF API 拉 5 个真实条码入本地 postgres 验证通过 - docs/etl-openfoodfacts.md: 流程/运行/字段映射 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- docs/etl-openfoodfacts.md | 38 ++++ ingestion/opengoods/adapters/openfoodfacts.py | 86 ++++++++ ingestion/opengoods/etl/load.py | 198 ++++++++++++++++++ ingestion/opengoods/etl/transform.py | 159 ++++++++++++++ ingestion/opengoods/jobs/seed_off.py | 66 ++++++ ingestion/pyproject.toml | 1 + ingestion/tests/fixtures/off_product.json | 26 +++ ingestion/tests/test_load_integration.py | 61 ++++++ ingestion/tests/test_transform.py | 67 ++++++ 9 files changed, 702 insertions(+) create mode 100644 docs/etl-openfoodfacts.md create mode 100644 ingestion/opengoods/adapters/openfoodfacts.py create mode 100644 ingestion/opengoods/etl/load.py create mode 100644 ingestion/opengoods/etl/transform.py create mode 100644 ingestion/opengoods/jobs/seed_off.py create mode 100644 ingestion/tests/fixtures/off_product.json create mode 100644 ingestion/tests/test_load_integration.py create mode 100644 ingestion/tests/test_transform.py diff --git a/docs/etl-openfoodfacts.md b/docs/etl-openfoodfacts.md new file mode 100644 index 0000000..8941a23 --- /dev/null +++ b/docs/etl-openfoodfacts.md @@ -0,0 +1,38 @@ +# ETL: Open Food Facts 导入 (M2) + +把 Open Food Facts (OFF, ODbL 许可) 的食品数据采集、转换并入库。只有 Python 采集侧写库，每条记录都以 `openfoodfacts` 为来源记录**字段级溯源**。 + +## 流程 +``` +OFF API / dump(jsonl[.gz]) + → adapters/openfoodfacts.py # 读取(限速 + User-Agent) / 解析 dump + → etl/transform.py # 字段映射 + 单位归一 + 营养 per_100g + 分类映射(关键词) + → etl/load.py # psycopg upsert(product/food_detail/product_image) + product_source 溯源 +``` + +## 运行 +先确保本地依赖与迁移就绪：`docker compose up -d postgres` + `migrate ... up`。 + +```bash +# 用 OFF API 拉指定条码(客户端限速, 默认 4s/次) +python -m opengoods.jobs.seed_off --barcodes 3017624010701 5449000000996 + +# 用下载好的 OFF dump 批量导入(可 .gz), 限制条数 +python -m opengoods.jobs.seed_off --dump products.jsonl.gz --limit 1000 +``` +DSN 默认读 `OPENGOODS_DATABASE_URL`。 + +## 字段映射要点 +| OFF | OpenGoods | 处理 | +|-----|-----------|------| +| `code` | `product.gtin` | GTIN-8/12/13/14 校验位验证, 不合法则不作为 gtin | +| `product_name_zh/_/_en` | `product.name` | 优先中文 | +| `brands` | `brand` | 取第一个, normalized_name 去重 | +| `quantity` | `net_content_*` | 解析 "500 g"/"1,5 L" → 经 `units.py` 归一(原始+归一双存) | +| `nutriments.*_100g` | `food_detail.nutriments` | per_100g; 能量 kJ/kcal 双存, 缺一自动换算 | +| `allergens_tags`/`additives_tags` | `allergens`/`additives` | 去 `en:` 前缀 | +| `nutriscore_grade` | `nutri_score` | 大写单字母 | +| `categories*`/name | `category_id` | 关键词映射到自建品类树(起步版, 后续换 OFF 分类→GPC 映射表) | +| `image_front_url` | `product_image` | 标 CC-BY-SA 许可 | + +> 全量 dump 约数 GB；CI 与单测用 fixture 离线验证 transform，DB 集成测试在无库时自动跳过。 diff --git a/ingestion/opengoods/adapters/openfoodfacts.py b/ingestion/opengoods/adapters/openfoodfacts.py new file mode 100644 index 0000000..9dda768 --- /dev/null +++ b/ingestion/opengoods/adapters/openfoodfacts.py @@ -0,0 +1,86 @@ +"""Open Food Facts (OFF) source adapter. + +Fetches raw product records either from the OFF read API (one product per +barcode) or from a downloaded JSONL dump file. OFF data is licensed under the +Open Database License (ODbL); product images are CC-BY-SA. We record OFF as the +source for every field we ingest. + +The adapter is read-only and rate-limited to stay well within OFF's API limits +(<= ~15 req/min/IP for product reads) and to be a good citizen. +""" + +from __future__ import annotations + +import json +import time +from collections.abc import Iterator +from pathlib import Path + +import httpx + +SOURCE_NAME = "openfoodfacts" +OFF_LICENSE = "ODbL" +USER_AGENT = "OpenGoods/0.1 (+https://github.com/baicai2026-baicai/goods) public-good product API" + +# Conservative client-side spacing between API calls (seconds). +_DEFAULT_MIN_INTERVAL = 4.0 +_API_URL = "https://world.openfoodfacts.org/api/v2/product/{barcode}.json" + + +class OpenFoodFactsAdapter: + """Read product records from the OFF API.""" + + source_name = SOURCE_NAME + + def __init__( + self, + client: httpx.Client | None = None, + min_interval: float = _DEFAULT_MIN_INTERVAL, + ) -> None: + self._client = client or httpx.Client(headers={"User-Agent": USER_AGENT}, timeout=30.0) + self._min_interval = min_interval + self._last_call = 0.0 + + def _throttle(self) -> None: + elapsed = time.monotonic() - self._last_call + wait = self._min_interval - elapsed + if wait > 0: + time.sleep(wait) + self._last_call = time.monotonic() + + def fetch_barcode(self, barcode: str) -> dict | None: + """Fetch a single product by barcode; return the raw `product` dict.""" + self._throttle() + resp = self._client.get(_API_URL.format(barcode=barcode)) + resp.raise_for_status() + payload = resp.json() + if payload.get("status") != 1: + return None + return payload["product"] + + def fetch(self, barcodes: list[str]) -> Iterator[dict]: + """Yield raw product records for the given barcodes.""" + for code in barcodes: + record = self.fetch_barcode(code) + if record is not None: + yield record + + +def read_dump(path: str | Path) -> Iterator[dict]: + """Yield raw product records from an OFF JSONL dump file. + + Each line is one product JSON object (the format of OFF's .jsonl export). + Supports plain or .gz files. + """ + p = Path(path) + if p.suffix == ".gz": + import gzip + + opener = lambda: gzip.open(p, "rt", encoding="utf-8") # noqa: E731 + else: + opener = lambda: open(p, encoding="utf-8") # noqa: E731 + with opener() as fh: + for line in fh: + line = line.strip() + if line: + yield json.loads(line) diff --git a/ingestion/opengoods/etl/load.py b/ingestion/opengoods/etl/load.py new file mode 100644 index 0000000..2460167 --- /dev/null +++ b/ingestion/opengoods/etl/load.py @@ -0,0 +1,198 @@ +"""Load transformed product records into the OpenGoods PostgreSQL database. + +Only the ingestion side writes to the database. Every load records OFF as the +source with field-level provenance in `product_source`. +""" + +from __future__ import annotations + +import json +import os +from typing import Any + +import psycopg +from psycopg.types.json import Jsonb + +from opengoods.adapters.openfoodfacts import OFF_LICENSE, SOURCE_NAME + +OFF_HOMEPAGE = "https://world.openfoodfacts.org" + + +def default_dsn() -> str: + return os.environ.get( + "OPENGOODS_DATABASE_URL", + "postgres://opengoods:opengoods@localhost:5432/opengoods?sslmode=disable", + ) + + +def _normalize_brand(name: str) -> str: + return " ".join(name.lower().split()) + + +def ensure_source(conn: psycopg.Connection) -> str: + """Upsert the Open Food Facts source row and return its id.""" + row = conn.execute( + """ + INSERT INTO source (name, homepage, license, trust_weight) + VALUES (%s, %s, %s, %s) + ON CONFLICT (name) DO UPDATE SET homepage = EXCLUDED.homepage + RETURNING id + """, + (SOURCE_NAME, OFF_HOMEPAGE, OFF_LICENSE, 0.7), + ).fetchone() + return row[0] + + +def _ensure_brand(conn: psycopg.Connection, name: str | None) -> str | None: + if not name: + return None + row = conn.execute( + """ + INSERT INTO brand (name, normalized_name) + VALUES (%s, %s) + ON CONFLICT (normalized_name) DO UPDATE SET name = brand.name + RETURNING id + """, + (name, _normalize_brand(name)), + ).fetchone() + return row[0] + + +def _category_id(conn: psycopg.Connection, path: str | None) -> tuple[str | None, str | None]: + if not path: + return None, None + row = conn.execute( + "SELECT id, gpc_brick_code FROM category WHERE path = %s::ltree", (path,) + ).fetchone() + return (row[0], row[1]) if row else (None, None) + + +def load_record(conn: psycopg.Connection, rec: dict[str, Any], source_id: str, raw: dict) -> str: + """Upsert one transformed record; return the product id.""" + brand_id = _ensure_brand(conn, rec.get("brand")) + category_id, gpc_brick = _category_id(conn, rec.get("category_path")) + + fields = ["name", "brand", "net_content", "category", "country_of_origin"] + + if rec.get("gtin"): + prod = conn.execute( + """ + INSERT INTO product (gtin, name, brand_id, category_id, gpc_brick_code, + net_content_value, net_content_unit, net_content_canonical, + country_of_origin, attributes) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (gtin) WHERE gtin IS NOT NULL DO UPDATE SET + name = EXCLUDED.name, + brand_id = COALESCE(EXCLUDED.brand_id, product.brand_id), + category_id = COALESCE(EXCLUDED.category_id, product.category_id), + gpc_brick_code = COALESCE(EXCLUDED.gpc_brick_code, product.gpc_brick_code), + net_content_value = EXCLUDED.net_content_value, + net_content_unit = EXCLUDED.net_content_unit, + net_content_canonical = EXCLUDED.net_content_canonical, + country_of_origin = EXCLUDED.country_of_origin + RETURNING id + """, + ( + rec["gtin"], + rec["name"], + brand_id, + category_id, + gpc_brick, + rec.get("net_content_value"), + rec.get("net_content_unit"), + rec.get("net_content_canonical"), + rec.get("country_of_origin"), + Jsonb({}), + ), + ).fetchone() + else: + prod = conn.execute( + """ + INSERT INTO product (name, brand_id, category_id, gpc_brick_code, + net_content_value, net_content_unit, net_content_canonical, + country_of_origin, attributes) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) + RETURNING id + """, + ( + rec["name"], + brand_id, + category_id, + gpc_brick, + rec.get("net_content_value"), + rec.get("net_content_unit"), + rec.get("net_content_canonical"), + rec.get("country_of_origin"), + Jsonb({}), + ), + ).fetchone() + product_id = prod[0] + + food = rec.get("food") or {} + conn.execute( + """ + INSERT INTO food_detail (product_id, ingredients_text, allergens, additives, + nutriments, nutrition_basis, serving_size, nutri_score) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s) + ON CONFLICT (product_id) DO UPDATE SET + ingredients_text = EXCLUDED.ingredients_text, + allergens = EXCLUDED.allergens, + additives = EXCLUDED.additives, + nutriments = EXCLUDED.nutriments, + nutrition_basis = EXCLUDED.nutrition_basis, + serving_size = EXCLUDED.serving_size, + nutri_score = EXCLUDED.nutri_score + """, + ( + product_id, + food.get("ingredients_text"), + food.get("allergens") or [], + food.get("additives") or [], + Jsonb(food.get("nutriments") or {}), + food.get("nutrition_basis"), + food.get("serving_size"), + food.get("nutri_score"), + ), + ) + + if rec.get("image_url"): + conn.execute( + """ + INSERT INTO product_image (product_id, url, kind, license, source_id) + VALUES (%s,%s,'front',%s,%s) + """, + (product_id, rec["image_url"], "CC-BY-SA", source_id), + ) + fields.append("image") + + conn.execute( + """ + INSERT INTO product_source (product_id, source_id, url, fields, fetched_at, raw) + VALUES (%s,%s,%s,%s, now(), %s) + """, + ( + product_id, + source_id, + f"{OFF_HOMEPAGE}/product/{rec.get('gtin') or ''}", + fields, + Jsonb(_jsonable(raw)), + ), + ) + return product_id + + +def _jsonable(raw: dict) -> dict: + """Drop values that are not JSON-serializable from a raw record.""" + try: + json.dumps(raw) + return raw + except (TypeError, ValueError): + return {k: v for k, v in raw.items() if _is_jsonable(v)} + + +def _is_jsonable(v: object) -> bool: + try: + json.dumps(v) + return True + except (TypeError, ValueError): + return False diff --git a/ingestion/opengoods/etl/transform.py b/ingestion/opengoods/etl/transform.py new file mode 100644 index 0000000..4201f9f --- /dev/null +++ b/ingestion/opengoods/etl/transform.py @@ -0,0 +1,159 @@ +"""Transform raw Open Food Facts records into the OpenGoods internal shape. + +Pure functions (no DB, no network) so they are easy to unit-test against +fixtures. The output dict mirrors the columns the loader writes. +""" + +from __future__ import annotations + +import re +from decimal import Decimal + +from opengoods.units import UnitError, normalize + +# OFF nutriment key -> our attribute key. Energy handled separately. +_NUTRIMENT_KEYS = { + "proteins_100g": "proteins", + "fat_100g": "fat", + "saturated-fat_100g": "saturated_fat", + "carbohydrates_100g": "carbohydrates", + "sugars_100g": "sugars", + "salt_100g": "salt", +} + +# Very small keyword -> category path map (starter; replaced by a proper +# OFF taxonomy -> GPC mapping table later). +_CATEGORY_KEYWORDS: list[tuple[tuple[str, ...], str]] = [ + (("water", "eau", "饮用水", "矿泉水"), "food.beverages.water"), + (("soda", "carbonated", "汽水", "碳酸"), "food.beverages.carbonated"), + (("juice", "jus", "果汁"), "food.beverages.juice"), + (("milk", "lait", "牛奶"), "food.dairy.milk"), + (("yogurt", "yoghurt", "yaourt", "酸奶"), "food.dairy.yogurt"), + (("cheese", "fromage", "奶酪", "干酪"), "food.dairy.cheese"), + (("bread", "pain", "面包"), "food.bakery.bread"), + (("biscuit", "cookie", "饼干"), "food.bakery.biscuits"), + (("chips", "crisps", "薯片", "膨化"), "food.snacks.chips"), + (("chocolate", "chocolat", "巧克力"), "food.snacks.chocolate"), + (("rice", "riz", "大米", "稻米"), "food.staple.rice"), + (("noodle", "pasta", "面条", "挂面"), "food.staple.noodles"), + (("oil", "huile", "食用油", "食油"), "food.staple.cooking_oil"), + (("soy sauce", "酱油"), "food.condiments.soy_sauce"), + (("salt", "sel", "食盐"), "food.condiments.salt"), +] + +_QTY_RE = re.compile(r"(?P\d+(?:[.,]\d+)?)\s*(?P[a-zA-Z\u4e00-\u9fff%]+)") + + +def is_valid_gtin(code: str) -> bool: + """Validate a GTIN-8/12/13/14 using the standard check digit.""" + if not code.isdigit() or len(code) not in (8, 12, 13, 14): + return False + digits = [int(c) for c in code] + check = digits[-1] + body = digits[:-1][::-1] + total = sum(d * (3 if i % 2 == 0 else 1) for i, d in enumerate(body)) + return (10 - total % 10) % 10 == check + + +def parse_quantity(text: str) -> tuple[Decimal, str] | None: + """Parse a free-text quantity like '500 g' or '1,5 L' -> (value, unit).""" + if not text: + return None + m = _QTY_RE.search(text) + if not m: + return None + value = Decimal(m.group("value").replace(",", ".")) + return value, m.group("unit") + + +def map_category(raw: dict) -> str | None: + """Best-effort map OFF categories/name to a self-built category path.""" + haystack = " ".join( + str(raw.get(k, "")) + for k in ("categories", "categories_tags", "product_name", "product_name_en") + ).lower() + for keywords, path in _CATEGORY_KEYWORDS: + if any(kw.lower() in haystack for kw in keywords): + return path + return None + + +def _clean_tags(tags: list[str] | None, prefix: str = "") -> list[str]: + out: list[str] = [] + for t in tags or []: + v = t.split(":", 1)[-1] if ":" in t else t + v = v.strip().replace("-", " ") + if v: + out.append(v) + return out + + +def transform_nutriments(off_nutriments: dict) -> dict: + """Build a nutriments dict on a per_100g basis with dual energy units.""" + out: dict[str, object] = {} + for off_key, our_key in _NUTRIMENT_KEYS.items(): + if off_key in off_nutriments and off_nutriments[off_key] is not None: + out[our_key] = float(off_nutriments[off_key]) + + kj = off_nutriments.get("energy-kj_100g") + kcal = off_nutriments.get("energy-kcal_100g") + if kj is None and kcal is not None: + kj = float(Decimal(str(kcal)) * Decimal("4.184")) + if kcal is None and kj is not None: + kcal = float(Decimal(str(kj)) / Decimal("4.184")) + if kj is not None: + out["energy_kj"] = round(float(kj), 3) + if kcal is not None: + out["energy_kcal"] = round(float(kcal), 3) + return out + + +def transform(raw: dict) -> dict | None: + """Transform one raw OFF product record into an internal product dict. + + Returns None if the record lacks a usable name. + """ + name = raw.get("product_name_zh") or raw.get("product_name") or raw.get("product_name_en") + if not name: + return None + + code = str(raw.get("code", "")).strip() + gtin = code if code and is_valid_gtin(code) else None + + brands = raw.get("brands") or "" + brand = brands.split(",")[0].strip() or None + + net_value = net_unit = net_canonical = None + parsed = parse_quantity(raw.get("quantity", "")) + if parsed: + value, unit = parsed + try: + norm = normalize(value, unit) + net_value, net_unit, net_canonical = ( + norm.value, + norm.unit, + norm.canonical_value, + ) + except UnitError: + net_value, net_unit = value, unit + + return { + "gtin": gtin, + "name": str(name).strip(), + "brand": brand, + "category_path": map_category(raw), + "net_content_value": net_value, + "net_content_unit": net_unit, + "net_content_canonical": net_canonical, + "country_of_origin": (raw.get("countries") or "").split(",")[0].strip() or None, + "food": { + "ingredients_text": raw.get("ingredients_text") or None, + "allergens": _clean_tags(raw.get("allergens_tags")), + "additives": _clean_tags(raw.get("additives_tags")), + "nutriments": transform_nutriments(raw.get("nutriments") or {}), + "nutrition_basis": "per_100g", + "serving_size": raw.get("serving_size") or None, + "nutri_score": (raw.get("nutriscore_grade") or "").upper()[:1] or None, + }, + "image_url": raw.get("image_front_url") or raw.get("image_url") or None, + } diff --git a/ingestion/opengoods/jobs/seed_off.py b/ingestion/opengoods/jobs/seed_off.py new file mode 100644 index 0000000..03c8ab0 --- /dev/null +++ b/ingestion/opengoods/jobs/seed_off.py @@ -0,0 +1,66 @@ +"""Seed the database with Open Food Facts data. + +Usage: + # from a list of barcodes via the OFF API + python -m opengoods.jobs.seed_off --barcodes 3017624010701 5449000000996 + + # from a downloaded OFF JSONL dump (optionally .gz), limited to N records + python -m opengoods.jobs.seed_off --dump products.jsonl.gz --limit 1000 + +The OFF read API is rate-limited client-side; for large imports use a dump. +""" + +from __future__ import annotations + +import argparse +import sys +from collections.abc import Iterator + +import psycopg + +from opengoods.adapters.openfoodfacts import OpenFoodFactsAdapter, read_dump +from opengoods.etl.load import default_dsn, ensure_source, load_record +from opengoods.etl.transform import transform + + +def _raw_records(args: argparse.Namespace) -> Iterator[dict]: + if args.dump: + records = read_dump(args.dump) + else: + adapter = OpenFoodFactsAdapter(min_interval=args.min_interval) + records = adapter.fetch(args.barcodes) + for i, rec in enumerate(records): + if args.limit and i >= args.limit: + break + yield rec + + +def run(args: argparse.Namespace) -> int: + loaded = skipped = 0 + with psycopg.connect(args.dsn, autocommit=False) as conn: + source_id = ensure_source(conn) + for raw in _raw_records(args): + rec = transform(raw) + if rec is None: + skipped += 1 + continue + load_record(conn, rec, source_id, raw) + loaded += 1 + conn.commit() + print(f"loaded={loaded} skipped={skipped}") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Seed OpenGoods from Open Food Facts") + src = parser.add_mutually_exclusive_group(required=True) + src.add_argument("--barcodes", nargs="+", help="barcodes to fetch via the OFF API") + src.add_argument("--dump", help="path to an OFF JSONL dump (.jsonl or .jsonl.gz)") + parser.add_argument("--limit", type=int, default=0, help="max records to load (0 = all)") + parser.add_argument("--min-interval", type=float, default=4.0, help="API throttle seconds") + parser.add_argument("--dsn", default=default_dsn(), help="PostgreSQL DSN") + return run(parser.parse_args(argv)) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/ingestion/pyproject.toml b/ingestion/pyproject.toml index 8be7953..0cefe1c 100644 --- a/ingestion/pyproject.toml +++ b/ingestion/pyproject.toml @@ -5,6 +5,7 @@ description = "OpenGoods (天工·商品标签) ingestion & ETL: collect product requires-python = ">=3.11" dependencies = [ "httpx>=0.27", + "psycopg[binary]>=3.2", ] [project.optional-dependencies] diff --git a/ingestion/tests/fixtures/off_product.json b/ingestion/tests/fixtures/off_product.json new file mode 100644 index 0000000..a674519 --- /dev/null +++ b/ingestion/tests/fixtures/off_product.json @@ -0,0 +1,26 @@ +{ + "code": "3017624010701", + "product_name": "Nutella", + "product_name_en": "Nutella hazelnut spread", + "brands": "Ferrero, Nutella", + "quantity": "400 g", + "countries": "France, China", + "categories": "Spreads, Hazelnut spreads, Chocolate spreads", + "categories_tags": ["en:spreads", "en:chocolate-spreads"], + "ingredients_text": "Sugar, palm oil, hazelnuts, cocoa, skimmed milk powder", + "allergens_tags": ["en:milk", "en:nuts"], + "additives_tags": ["en:e322"], + "serving_size": "15 g", + "nutriscore_grade": "e", + "image_front_url": "https://images.openfoodfacts.org/images/products/301/762/401/0701/front_en.jpg", + "nutriments": { + "energy-kj_100g": 2252, + "energy-kcal_100g": 539, + "fat_100g": 30.9, + "saturated-fat_100g": 10.6, + "carbohydrates_100g": 57.5, + "sugars_100g": 56.3, + "proteins_100g": 6.3, + "salt_100g": 0.107 + } +} diff --git a/ingestion/tests/test_load_integration.py b/ingestion/tests/test_load_integration.py new file mode 100644 index 0000000..e0eef00 --- /dev/null +++ b/ingestion/tests/test_load_integration.py @@ -0,0 +1,61 @@ +"""Integration test for the DB loader. + +Skipped automatically when no database is reachable (e.g. local runs without +docker, or CI jobs without a postgres service). Requires migrations applied. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from opengoods.etl.load import default_dsn, ensure_source, load_record +from opengoods.etl.transform import transform + +psycopg = pytest.importorskip("psycopg") + +FIXTURE = json.loads((Path(__file__).parent / "fixtures" / "off_product.json").read_text()) + + +@pytest.fixture() +def conn(): + try: + c = psycopg.connect(default_dsn(), connect_timeout=3) + except psycopg.OperationalError as exc: # pragma: no cover - env dependent + pytest.skip(f"no database available: {exc}") + # ensure schema present + has_product = c.execute("SELECT to_regclass('public.product') IS NOT NULL").fetchone()[0] + if not has_product: + c.close() + pytest.skip("migrations not applied") + yield c + c.rollback() + c.close() + + +def test_load_record_roundtrip(conn): + source_id = ensure_source(conn) + rec = transform(FIXTURE) + product_id = load_record(conn, rec, source_id, FIXTURE) + + row = conn.execute( + "SELECT name, gtin, net_content_unit FROM product WHERE id = %s", (product_id,) + ).fetchone() + assert row[0] == "Nutella" + assert row[1] == "3017624010701" + assert row[2] == "g" + + nutri = conn.execute( + "SELECT nutriments ->> 'energy_kcal' FROM food_detail WHERE product_id = %s", + (product_id,), + ).fetchone() + assert nutri[0] == "539.0" + + prov = conn.execute( + "SELECT count(*) FROM product_source WHERE product_id = %s", (product_id,) + ).fetchone() + assert prov[0] >= 1 + + conn.rollback() # keep the test DB clean diff --git a/ingestion/tests/test_transform.py b/ingestion/tests/test_transform.py new file mode 100644 index 0000000..e4bf98e --- /dev/null +++ b/ingestion/tests/test_transform.py @@ -0,0 +1,67 @@ +import json +from decimal import Decimal +from pathlib import Path + +import pytest + +from opengoods.etl.transform import ( + is_valid_gtin, + map_category, + parse_quantity, + transform, + transform_nutriments, +) + +FIXTURE = json.loads((Path(__file__).parent / "fixtures" / "off_product.json").read_text()) + + +def test_is_valid_gtin(): + assert is_valid_gtin("3017624010701") # real EAN-13 + assert is_valid_gtin("5449000000996") # Coca-Cola + assert not is_valid_gtin("3017624010700") # bad check digit + assert not is_valid_gtin("123") + assert not is_valid_gtin("notanumber") + + +def test_parse_quantity(): + assert parse_quantity("400 g") == (Decimal("400"), "g") + assert parse_quantity("1,5 L") == (Decimal("1.5"), "L") + assert parse_quantity("") is None + assert parse_quantity("family size") is None + + +def test_map_category(): + assert map_category({"product_name": "Spring Water"}) == "food.beverages.water" + assert map_category({"categories": "Dark chocolate"}) == "food.snacks.chocolate" + assert map_category({"product_name": "Mystery"}) is None + + +def test_transform_nutriments_dual_energy(): + out = transform_nutriments(FIXTURE["nutriments"]) + assert out["energy_kj"] == 2252.0 + assert out["energy_kcal"] == 539.0 + assert out["fat"] == 30.9 + assert out["salt"] == 0.107 + + +def test_transform_nutriments_fills_missing_energy(): + out = transform_nutriments({"energy-kcal_100g": 100}) + assert out["energy_kj"] == pytest.approx(418.4) + + +def test_transform_full_record(): + rec = transform(FIXTURE) + assert rec is not None + assert rec["gtin"] == "3017624010701" + assert rec["name"] == "Nutella" + assert rec["brand"] == "Ferrero" + assert rec["net_content_unit"] == "g" + assert rec["net_content_canonical"] == Decimal("400") + assert rec["country_of_origin"] == "France" + assert rec["food"]["nutri_score"] == "E" + assert "milk" in rec["food"]["allergens"] + assert rec["image_url"].endswith(".jpg") + + +def test_transform_drops_unnamed(): + assert transform({"code": "0000000000000"}) is None