goods/ingestion/opengoods/adapters/gs1.py

"""GS1 barcode supplement adapter.

GS1 (e.g. *Verified by GS1* / GS1 China) is the authoritative registry that maps
a GTIN to its brand owner, product description and GPC category. We use it to
*supplement* — fill gaps in — records gathered from crowd sources like Open Food
Facts, never to overwrite existing values.

Real GS1 access is credentialed and region-specific, so this adapter supports
two modes:

* **offline** (default): look barcodes up in a local JSON mapping file. This is
  what tests and air-gapped runs use.
* **online**: GET ``{base_url}/{gtin}`` with an API key header, then normalize
  the response. Enabled by passing ``base_url`` + ``client``.

Either way :meth:`fetch_barcode` returns a normalized *supplement* dict (or
``None``); :mod:`opengoods.etl.supplement` applies it to the database.
"""

from __future__ import annotations

import json
from collections.abc import Iterator
from pathlib import Path

import httpx

SOURCE_NAME = "gs1"
GS1_HOMEPAGE = "https://www.gs1.org"
GS1_LICENSE = "proprietary"
# GS1 is the authoritative barcode registry -> high trust.
GS1_TRUST = 0.9

# Keys of a normalized supplement record.
_SUPPLEMENT_KEYS = (
    "gtin",
    "name",
    "brand",
    "manufacturer",
    "gpc_brick_code",
    "country_of_origin",
    "net_content_value",
    "net_content_unit",
)


def _normalize(code: str, data: dict) -> dict:
    """Project a raw mapping/record onto the supplement schema (non-empty only)."""
    rec: dict = {"gtin": code}
    for key in _SUPPLEMENT_KEYS:
        if key == "gtin":
            continue
        value = data.get(key)
        if value not in (None, "", []):
            rec[key] = value
    return rec


def _parse_api(code: str, payload: dict) -> dict:
    """Best-effort mapping of a Verified-by-GS1 style payload to our schema."""
    item = payload
    if isinstance(payload.get("gtinRecords"), list) and payload["gtinRecords"]:
        item = payload["gtinRecords"][0]
    return _normalize(
        code,
        {
            "name": item.get("productDescription") or item.get("description"),
            "brand": item.get("brandName"),
            "manufacturer": item.get("companyName") or item.get("licenseeName"),
            "gpc_brick_code": item.get("gpcCategoryCode"),
            "country_of_origin": item.get("countryOfSaleCode") or item.get("countryCode"),
            "net_content_value": item.get("netContent"),
            "net_content_unit": item.get("netContentUnit"),
        },
    )


class GS1Adapter:
    """Look up GTIN supplements from a local mapping or a GS1-style API."""

    source_name = SOURCE_NAME

    def __init__(
        self,
        mapping: dict | None = None,
        *,
        client: httpx.Client | None = None,
        base_url: str | None = None,
        api_key: str | None = None,
    ) -> None:
        self._mapping = mapping or {}
        self._client = client
        self._base_url = base_url.rstrip("/") if base_url else None
        self._api_key = api_key

    @classmethod
    def from_file(cls, path: str | Path) -> GS1Adapter:
        """Build an offline adapter from a JSON ``{gtin: {...}}`` mapping file."""
        data = json.loads(Path(path).read_text(encoding="utf-8"))
        return cls(mapping=data)

    def fetch_barcode(self, code: str) -> dict | None:
        """Return a normalized supplement dict for ``code`` (or ``None``)."""
        if self._base_url and self._client is not None:
            headers = {"apikey": self._api_key} if self._api_key else {}
            resp = self._client.get(f"{self._base_url}/{code}", headers=headers)
            if resp.status_code == 404:
                return None
            resp.raise_for_status()
            rec = _parse_api(code, resp.json())
        else:
            data = self._mapping.get(code)
            if not data:
                return None
            rec = _normalize(code, data)
        # A record with only the GTIN carries no supplement.
        return rec if len(rec) > 1 else None

    def fetch(self, barcodes: list[str]) -> Iterator[dict]:
        """Yield supplement records for the given barcodes."""
        for code in barcodes:
            rec = self.fetch_barcode(code)
            if rec is not None:
                yield rec