feat(M2): Open Food Facts 采集导入 ETL
- 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>
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user