M4: ingestion management (incremental, GS1 supplement, dedup/conflict, quality, scheduler)

- OFF incremental fetch via search API + persistent watermark (ingest_state, migration 0004)
- GS1 barcode supplement adapter (offline mapping + GS1-style API) filling only gaps with field-level provenance
- Non-GTIN dedup with canonical selection + merge_log; field-level conflict resolution (source trust > recency)
- Quality scoring (0.4 completeness + 0.3 source trust + 0.2 multi-source + 0.1 freshness) wired into load/merge
- Jobs: update_off, dedup, schedule; docs/ingestion-management.md
- 19 new tests (pure + DB-integration), ruff clean

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

.github/workflows/ci.yml

@@ -12,12 +12,31 @@ jobs:
     defaults:
       run:
         working-directory: api
+    services:
+      postgres:
+        image: postgres:16-alpine
+        env:
+          POSTGRES_USER: opengoods
+          POSTGRES_PASSWORD: opengoods
+          POSTGRES_DB: opengoods
+        ports:
+          - "5432:5432"
+        options: >-
+          --health-cmd "pg_isready -U opengoods"
+          --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      OPENGOODS_DATABASE_URL: postgres://opengoods:opengoods@localhost:5432/opengoods?sslmode=disable
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5
         with:
           go-version: "1.23"
           cache-dependency-path: api/go.sum
+      - name: Apply migrations
+        working-directory: .
+        run: |
+          go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.18.1
+          migrate -path migrations -database "$OPENGOODS_DATABASE_URL" up
       - name: Verify gofmt
         run: test -z "$(gofmt -l .)"
       - run: go vet ./...
@@ -43,3 +62,32 @@ jobs:
         run: ruff format --check .
       - name: Pytest
         run: pytest -q
+
+  migrations:
+    name: Migrations (postgres)
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:16-alpine
+        env:
+          POSTGRES_USER: opengoods
+          POSTGRES_PASSWORD: opengoods
+          POSTGRES_DB: opengoods
+        ports:
+          - "5432:5432"
+        options: >-
+          --health-cmd "pg_isready -U opengoods"
+          --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      DBURL: postgres://opengoods:opengoods@localhost:5432/opengoods?sslmode=disable
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-go@v5
+        with:
+          go-version: "1.23"
+      - name: Install golang-migrate
+        run: go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.18.1
+      - name: Migrate up
+        run: migrate -path migrations -database "$DBURL" up
+      - name: Migrate down (reversibility)
+        run: migrate -path migrations -database "$DBURL" down -all

api/cmd/server/main.go

@@ -2,20 +2,39 @@
 package main
 
 import (
+	"context"
 	"log"
 	"net/http"
 	"time"
 
+	"github.com/jackc/pgx/v5/pgxpool"
+
 	"github.com/baicai2026-baicai/goods/api/internal/config"
 	"github.com/baicai2026-baicai/goods/api/internal/handler"
+	"github.com/baicai2026-baicai/goods/api/internal/store"
 )
 
 func main() {
 	cfg := config.Load()
 
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, cfg.DatabaseURL)
+	if err != nil {
+		log.Fatalf("failed to create db pool: %v", err)
+	}
+	defer pool.Close()
+
+	pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	if err := pool.Ping(pingCtx); err != nil {
+		log.Printf("warning: database not reachable at startup: %v", err)
+	}
+
+	h := handler.New(store.New(pool))
+
 	srv := &http.Server{
 		Addr:              cfg.Addr,
-		Handler:           handler.Router(),
+		Handler:           h.Router(),
 		ReadHeaderTimeout: 10 * time.Second,
 	}
 

api/go.mod

@@ -2,4 +2,16 @@ module github.com/baicai2026-baicai/goods/api
 
 go 1.23.4
 
-require github.com/go-chi/chi/v5 v5.1.0
+require (
+	github.com/go-chi/chi/v5 v5.1.0
+	github.com/jackc/pgx/v5 v5.7.2
+)
+
+require (
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	golang.org/x/crypto v0.31.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+)

api/go.sum

@@ -1,2 +1,30 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw=
 github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.7.2 h1:mLoDLV6sonKlvjIEsV56SkWNCnuNv531l94GaIzO+XI=
+github.com/jackc/pgx/v5 v5.7.2/go.mod h1:ncY89UGWxg82EykZUwSpUKEfccBGGYq1xjrOpsbsfGQ=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

api/internal/handler/handler.go

@@ -1,5 +1,4 @@
 // Package handler wires up the public, read-only OpenGoods HTTP API.
-//
 // The OpenGoods service is a public-good product information API: it only
 // collects and serves product facts. It exposes no purchase, checkout, or
 // commerce endpoints by design.
@@ -7,48 +6,188 @@ package handler
 
 import (
 	"encoding/json"
+	"errors"
 	"net/http"
+	"strconv"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/chi/v5/middleware"
+
+	"github.com/baicai2026-baicai/goods/api/internal/store"
 )
 
 // APIVersion is the current public API version prefix.
 const APIVersion = "v1"
 
+const (
+	defaultPageSize = 20
+	maxPageSize     = 100
+)
+
+// Handler holds dependencies shared by the HTTP routes.
+type Handler struct {
+	store *store.Store
+}
+
+// New constructs a Handler backed by the given store.
+func New(s *store.Store) *Handler {
+	return &Handler{store: s}
+}
+
 // Router builds the top-level HTTP handler with middleware and routes mounted.
-func Router() http.Handler {
+func (h *Handler) Router() http.Handler {
 	r := chi.NewRouter()
 	r.Use(middleware.RequestID)
 	r.Use(middleware.RealIP)
 	r.Use(middleware.Recoverer)
 
-	r.Get("/healthz", Healthz)
+	r.Get("/healthz", h.Healthz)
 
 	r.Route("/api/"+APIVersion, func(r chi.Router) {
 		r.Route("/products", func(r chi.Router) {
-			r.Get("/barcode/{gtin}", notImplemented)
-			r.Get("/search", notImplemented)
-			r.Get("/{id}", notImplemented)
-			r.Get("/{id}/nutriments", notImplemented)
-			r.Get("/{id}/msrp", notImplemented)
+			r.Get("/barcode/{gtin}", h.ProductByBarcode)
+			r.Get("/search", h.SearchProducts)
+			r.Get("/{id}", h.ProductByID)
+			r.Get("/{id}/nutriments", h.ProductNutriments)
+			r.Get("/{id}/msrp", h.ProductMSRP)
 		})
-		r.Get("/brands", notImplemented)
-		r.Get("/categories", notImplemented)
-		r.Get("/sources/{id}", notImplemented)
+		r.Get("/brands", h.ListBrands)
+		r.Get("/categories", h.ListCategories)
+		r.Get("/sources/{id}", h.SourceByID)
 	})
 
 	return r
 }
 
 // Healthz reports liveness of the service.
-func Healthz(w http.ResponseWriter, r *http.Request) {
+func (h *Handler) Healthz(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
 }
 
-// notImplemented is a placeholder for endpoints scoped to later milestones.
-func notImplemented(w http.ResponseWriter, r *http.Request) {
-	writeError(w, r, http.StatusNotImplemented, "not_implemented", "endpoint not implemented yet")
+// ProductByBarcode returns a product by its GTIN.
+func (h *Handler) ProductByBarcode(w http.ResponseWriter, r *http.Request) {
+	p, err := h.store.ProductByGTIN(r.Context(), chi.URLParam(r, "gtin"))
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, p)
+}
+
+// ProductByID returns a product by its UUID.
+func (h *Handler) ProductByID(w http.ResponseWriter, r *http.Request) {
+	p, err := h.store.ProductByID(r.Context(), chi.URLParam(r, "id"))
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, p)
+}
+
+// SearchProducts runs a fuzzy name search with optional category filter + paging.
+func (h *Handler) SearchProducts(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query().Get("q")
+	category := r.URL.Query().Get("category")
+	page, size := pageParams(r)
+
+	items, total, err := h.store.SearchProducts(r.Context(), q, category, size, (page-1)*size)
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"items": items,
+		"page":  page,
+		"size":  size,
+		"total": total,
+	})
+}
+
+// ProductNutriments returns just the nutrition facts of a product.
+func (h *Handler) ProductNutriments(w http.ResponseWriter, r *http.Request) {
+	n, err := h.store.Nutriments(r.Context(), chi.URLParam(r, "id"))
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, n)
+}
+
+// ProductMSRP returns official suggested retail price snapshots (no purchase link).
+func (h *Handler) ProductMSRP(w http.ResponseWriter, r *http.Request) {
+	items, err := h.store.ListMSRP(r.Context(), chi.URLParam(r, "id"))
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"items":      items,
+		"disclaimer": "厂商建议零售价历史快照，仅供参考，不构成购买建议，本服务不提供任何购买入口。",
+	})
+}
+
+// ListBrands returns a paginated list of brands.
+func (h *Handler) ListBrands(w http.ResponseWriter, r *http.Request) {
+	page, size := pageParams(r)
+	items, total, err := h.store.ListBrands(r.Context(), size, (page-1)*size)
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"items": items, "page": page, "size": size, "total": total,
+	})
+}
+
+// ListCategories returns the full category tree.
+func (h *Handler) ListCategories(w http.ResponseWriter, r *http.Request) {
+	items, err := h.store.ListCategories(r.Context())
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{"items": items})
+}
+
+// SourceByID returns a single data source.
+func (h *Handler) SourceByID(w http.ResponseWriter, r *http.Request) {
+	src, err := h.store.SourceByID(r.Context(), chi.URLParam(r, "id"))
+	if h.handleErr(w, r, err) {
+		return
+	}
+	writeJSON(w, http.StatusOK, src)
+}
+
+// handleErr writes an appropriate error response; returns true if it handled one.
+func (h *Handler) handleErr(w http.ResponseWriter, r *http.Request, err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, store.ErrNotFound) {
+		writeError(w, r, http.StatusNotFound, "not_found", "resource not found")
+		return true
+	}
+	writeError(w, r, http.StatusInternalServerError, "internal_error", "internal server error")
+	return true
+}
+
+func pageParams(r *http.Request) (page, size int) {
+	page = atoiDefault(r.URL.Query().Get("page"), 1)
+	if page < 1 {
+		page = 1
+	}
+	size = atoiDefault(r.URL.Query().Get("size"), defaultPageSize)
+	if size < 1 {
+		size = defaultPageSize
+	}
+	if size > maxPageSize {
+		size = maxPageSize
+	}
+	return page, size
+}
+
+func atoiDefault(s string, fallback int) int {
+	if s == "" {
+		return fallback
+	}
+	v, err := strconv.Atoi(s)
+	if err != nil {
+		return fallback
+	}
+	return v
 }
 
 func writeJSON(w http.ResponseWriter, status int, body any) {

api/internal/handler/handler_db_test.go

@@ -0,0 +1,134 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/baicai2026-baicai/goods/api/internal/store"
+)
+
+// newTestHandler connects to the test database, skipping if unavailable or
+// unmigrated. It inserts a known product (cleaned up via t.Cleanup) so the
+// endpoint assertions are deterministic.
+func newTestHandler(t *testing.T) (*Handler, string) {
+	t.Helper()
+	dsn := os.Getenv("OPENGOODS_DATABASE_URL")
+	if dsn == "" {
+		dsn = "postgres://opengoods:opengoods@localhost:5432/opengoods?sslmode=disable"
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Skipf("no database: %v", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		t.Skipf("database not reachable: %v", err)
+	}
+
+	var hasProduct bool
+	if err := pool.QueryRow(ctx, "SELECT to_regclass('public.product') IS NOT NULL").Scan(&hasProduct); err != nil || !hasProduct {
+		pool.Close()
+		t.Skip("migrations not applied")
+	}
+
+	gtin := "4006381333931"
+	_, err = pool.Exec(context.Background(), `
+		INSERT INTO product (gtin, name, category_id, net_content_value, net_content_unit)
+		VALUES ($1, 'Test Cola', (SELECT id FROM category WHERE path='food.beverages.carbonated'), 330, 'ml')
+		ON CONFLICT (gtin) WHERE gtin IS NOT NULL DO UPDATE SET name = EXCLUDED.name`, gtin)
+	if err != nil {
+		pool.Close()
+		t.Fatalf("seed insert failed: %v", err)
+	}
+	var pid string
+	_ = pool.QueryRow(context.Background(), "SELECT id FROM product WHERE gtin=$1", gtin).Scan(&pid)
+	_, _ = pool.Exec(context.Background(), `
+		INSERT INTO food_detail (product_id, nutrition_basis, nutriments)
+		VALUES ($1, 'per_100ml', '{"energy_kcal": 42}'::jsonb)
+		ON CONFLICT (product_id) DO UPDATE SET nutriments = EXCLUDED.nutriments`, pid)
+
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), "DELETE FROM product WHERE gtin=$1", gtin)
+		pool.Close()
+	})
+	return New(store.New(pool)), gtin
+}
+
+func doGET(t *testing.T, h *Handler, path string) *httptest.ResponseRecorder {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodGet, path, nil)
+	rec := httptest.NewRecorder()
+	h.Router().ServeHTTP(rec, req)
+	return rec
+}
+
+func TestProductByBarcode(t *testing.T) {
+	h, gtin := newTestHandler(t)
+	rec := doGET(t, h, "/api/"+APIVersion+"/products/barcode/"+gtin)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+	}
+	var p store.Product
+	if err := json.NewDecoder(rec.Body).Decode(&p); err != nil {
+		t.Fatal(err)
+	}
+	if p.Name != "Test Cola" || p.GTIN == nil || *p.GTIN != gtin {
+		t.Fatalf("unexpected product: %+v", p)
+	}
+	if p.CategoryPath == nil || *p.CategoryPath != "food.beverages.carbonated" {
+		t.Fatalf("category not joined: %+v", p.CategoryPath)
+	}
+}
+
+func TestProductByBarcodeNotFound(t *testing.T) {
+	h, _ := newTestHandler(t)
+	rec := doGET(t, h, "/api/"+APIVersion+"/products/barcode/0000000000000")
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("expected 404, got %d", rec.Code)
+	}
+}
+
+func TestSearchProducts(t *testing.T) {
+	h, _ := newTestHandler(t)
+	rec := doGET(t, h, "/api/"+APIVersion+"/products/search?q=Cola&category=food.beverages")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	var body struct {
+		Items []store.ProductSummary `json:"items"`
+		Total int                    `json:"total"`
+	}
+	if err := json.NewDecoder(rec.Body).Decode(&body); err != nil {
+		t.Fatal(err)
+	}
+	if body.Total < 1 {
+		t.Fatalf("expected at least 1 result, got %d", body.Total)
+	}
+}
+
+func TestListCategories(t *testing.T) {
+	h, _ := newTestHandler(t)
+	rec := doGET(t, h, "/api/"+APIVersion+"/categories")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	var body struct {
+		Items []store.Category `json:"items"`
+	}
+	if err := json.NewDecoder(rec.Body).Decode(&body); err != nil {
+		t.Fatal(err)
+	}
+	if len(body.Items) < 20 {
+		t.Fatalf("expected seeded categories, got %d", len(body.Items))
+	}
+}

api/internal/handler/handler_test.go

@@ -11,7 +11,7 @@ func TestHealthz(t *testing.T) {
 	req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
 	rec := httptest.NewRecorder()
 
-	Router().ServeHTTP(rec, req)
+	New(nil).Router().ServeHTTP(rec, req)
 
 	if rec.Code != http.StatusOK {
 		t.Fatalf("expected status %d, got %d", http.StatusOK, rec.Code)
@@ -26,13 +26,23 @@ func TestHealthz(t *testing.T) {
 	}
 }
 
-func TestProductEndpointNotImplemented(t *testing.T) {
-	req := httptest.NewRequest(http.MethodGet, "/api/"+APIVersion+"/products/barcode/3017624010701", nil)
-	rec := httptest.NewRecorder()
-
-	Router().ServeHTTP(rec, req)
-
-	if rec.Code != http.StatusNotImplemented {
-		t.Fatalf("expected status %d, got %d", http.StatusNotImplemented, rec.Code)
+func TestPageParams(t *testing.T) {
+	cases := []struct {
+		query            string
+		wantPage, wantSz int
+	}{
+		{"", 1, defaultPageSize},
+		{"page=3&size=10", 3, 10},
+		{"page=0&size=-5", 1, defaultPageSize},
+		{"size=1000", 1, maxPageSize},
+		{"page=abc", 1, defaultPageSize},
+	}
+	for _, c := range cases {
+		req := httptest.NewRequest(http.MethodGet, "/?"+c.query, nil)
+		page, size := pageParams(req)
+		if page != c.wantPage || size != c.wantSz {
+			t.Errorf("query %q: got page=%d size=%d, want page=%d size=%d",
+				c.query, page, size, c.wantPage, c.wantSz)
+		}
 	}
 }

api/internal/store/store.go

@@ -0,0 +1,278 @@
+// Package store is the read-only data access layer for the OpenGoods API.
+// It only issues SELECT queries; all writes happen in the Python ingestion path.
+package store
+
+import (
+	"context"
+	"errors"
+	"strconv"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// ErrNotFound is returned when a requested row does not exist.
+var ErrNotFound = errors.New("not found")
+
+// Store wraps a PostgreSQL connection pool.
+type Store struct {
+	pool *pgxpool.Pool
+}
+
+// New constructs a Store from an existing pgx pool.
+func New(pool *pgxpool.Pool) *Store {
+	return &Store{pool: pool}
+}
+
+// Ping verifies database connectivity.
+func (s *Store) Ping(ctx context.Context) error {
+	return s.pool.Ping(ctx)
+}
+
+// Product is the full public view of a product.
+type Product struct {
+	ID              string         `json:"id"`
+	GTIN            *string        `json:"gtin"`
+	Name            string         `json:"name"`
+	Brand           *string        `json:"brand"`
+	CategoryPath    *string        `json:"category_path"`
+	GPCBrickCode    *string        `json:"gpc_brick_code"`
+	NetContentValue *float64       `json:"net_content_value"`
+	NetContentUnit  *string        `json:"net_content_unit"`
+	CountryOfOrigin *string        `json:"country_of_origin"`
+	QualityScore    float64        `json:"quality_score"`
+	Nutriments      map[string]any `json:"nutriments,omitempty"`
+	NutritionBasis  *string        `json:"nutrition_basis,omitempty"`
+	NutriScore      *string        `json:"nutri_score,omitempty"`
+	Ingredients     *string        `json:"ingredients_text,omitempty"`
+	Allergens       []string       `json:"allergens,omitempty"`
+	Additives       []string       `json:"additives,omitempty"`
+}
+
+// ProductSummary is a lightweight row used in search/listing responses.
+type ProductSummary struct {
+	ID           string  `json:"id"`
+	GTIN         *string `json:"gtin"`
+	Name         string  `json:"name"`
+	Brand        *string `json:"brand"`
+	CategoryPath *string `json:"category_path"`
+}
+
+const productSelect = `
+SELECT p.id, p.gtin, p.name, b.name, c.path::text, p.gpc_brick_code,
+       p.net_content_value, p.net_content_unit, p.country_of_origin, p.quality_score,
+       f.nutriments, f.nutrition_basis, f.nutri_score, f.ingredients_text,
+       f.allergens, f.additives
+FROM product p
+LEFT JOIN brand b ON b.id = p.brand_id
+LEFT JOIN category c ON c.id = p.category_id
+LEFT JOIN food_detail f ON f.product_id = p.id
+`
+
+func scanProduct(row pgx.Row) (*Product, error) {
+	var p Product
+	err := row.Scan(
+		&p.ID, &p.GTIN, &p.Name, &p.Brand, &p.CategoryPath, &p.GPCBrickCode,
+		&p.NetContentValue, &p.NetContentUnit, &p.CountryOfOrigin, &p.QualityScore,
+		&p.Nutriments, &p.NutritionBasis, &p.NutriScore, &p.Ingredients,
+		&p.Allergens, &p.Additives,
+	)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &p, nil
+}
+
+// ProductByGTIN looks up an active product by its barcode.
+func (s *Store) ProductByGTIN(ctx context.Context, gtin string) (*Product, error) {
+	row := s.pool.QueryRow(ctx, productSelect+" WHERE p.gtin = $1 AND p.status = 'active'", gtin)
+	return scanProduct(row)
+}
+
+// ProductByID looks up a product by its UUID.
+func (s *Store) ProductByID(ctx context.Context, id string) (*Product, error) {
+	row := s.pool.QueryRow(ctx, productSelect+" WHERE p.id = $1", id)
+	return scanProduct(row)
+}
+
+// SearchProducts performs a fuzzy name search with optional category subtree filter.
+func (s *Store) SearchProducts(ctx context.Context, q, category string, limit, offset int) ([]ProductSummary, int, error) {
+	args := []any{}
+	where := "WHERE p.status = 'active'"
+	if q != "" {
+		args = append(args, q)
+		where += " AND p.name ILIKE '%' || $1 || '%'"
+	}
+	if category != "" {
+		args = append(args, category)
+		where += " AND c.path <@ $" + strconv.Itoa(len(args)) + "::ltree"
+	}
+
+	countSQL := "SELECT count(*) FROM product p LEFT JOIN category c ON c.id = p.category_id " + where
+	var total int
+	if err := s.pool.QueryRow(ctx, countSQL, args...).Scan(&total); err != nil {
+		return nil, 0, err
+	}
+
+	args = append(args, limit, offset)
+	listSQL := `
+SELECT p.id, p.gtin, p.name, b.name, c.path::text
+FROM product p
+LEFT JOIN brand b ON b.id = p.brand_id
+LEFT JOIN category c ON c.id = p.category_id ` + where +
+		" ORDER BY p.name LIMIT $" + strconv.Itoa(len(args)-1) + " OFFSET $" + strconv.Itoa(len(args))
+
+	rows, err := s.pool.Query(ctx, listSQL, args...)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer rows.Close()
+
+	out := []ProductSummary{}
+	for rows.Next() {
+		var ps ProductSummary
+		if err := rows.Scan(&ps.ID, &ps.GTIN, &ps.Name, &ps.Brand, &ps.CategoryPath); err != nil {
+			return nil, 0, err
+		}
+		out = append(out, ps)
+	}
+	return out, total, rows.Err()
+}
+
+// Nutriments returns just the nutrition payload for a product.
+type Nutriments struct {
+	ProductID  string         `json:"product_id"`
+	Basis      *string        `json:"basis"`
+	NutriScore *string        `json:"nutri_score"`
+	Values     map[string]any `json:"values"`
+}
+
+// Nutriments fetches the nutrition facts of a product.
+func (s *Store) Nutriments(ctx context.Context, id string) (*Nutriments, error) {
+	var n Nutriments
+	n.ProductID = id
+	err := s.pool.QueryRow(ctx,
+		"SELECT nutriments, nutrition_basis, nutri_score FROM food_detail WHERE product_id = $1", id,
+	).Scan(&n.Values, &n.Basis, &n.NutriScore)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &n, nil
+}
+
+// MSRP is an official suggested retail price snapshot (never a purchase link).
+type MSRP struct {
+	Amount        float64 `json:"amount"`
+	Currency      string  `json:"currency"`
+	Region        string  `json:"region"`
+	EffectiveDate *string `json:"effective_date"`
+	SourceURL     *string `json:"source_url"`
+	Note          *string `json:"note"`
+}
+
+// ListMSRP returns all MSRP snapshots for a product.
+func (s *Store) ListMSRP(ctx context.Context, id string) ([]MSRP, error) {
+	rows, err := s.pool.Query(ctx,
+		`SELECT amount, currency, region, effective_date::text, source_url, note
+		 FROM product_msrp WHERE product_id = $1 ORDER BY effective_date DESC NULLS LAST`, id)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []MSRP{}
+	for rows.Next() {
+		var m MSRP
+		if err := rows.Scan(&m.Amount, &m.Currency, &m.Region, &m.EffectiveDate, &m.SourceURL, &m.Note); err != nil {
+			return nil, err
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
+
+// Brand is a public brand entry.
+type Brand struct {
+	ID   string `json:"id"`
+	Name string `json:"name"`
+}
+
+// ListBrands returns brands ordered by name.
+func (s *Store) ListBrands(ctx context.Context, limit, offset int) ([]Brand, int, error) {
+	var total int
+	if err := s.pool.QueryRow(ctx, "SELECT count(*) FROM brand").Scan(&total); err != nil {
+		return nil, 0, err
+	}
+	rows, err := s.pool.Query(ctx, "SELECT id, name FROM brand ORDER BY name LIMIT $1 OFFSET $2", limit, offset)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer rows.Close()
+	out := []Brand{}
+	for rows.Next() {
+		var b Brand
+		if err := rows.Scan(&b.ID, &b.Name); err != nil {
+			return nil, 0, err
+		}
+		out = append(out, b)
+	}
+	return out, total, rows.Err()
+}
+
+// Category is a node in the self-built category tree.
+type Category struct {
+	ID           string  `json:"id"`
+	NameZH       string  `json:"name_zh"`
+	NameEN       *string `json:"name_en"`
+	Path         string  `json:"path"`
+	GPCBrickCode *string `json:"gpc_brick_code"`
+	Level        int     `json:"level"`
+}
+
+// ListCategories returns the full category tree ordered by path.
+func (s *Store) ListCategories(ctx context.Context) ([]Category, error) {
+	rows, err := s.pool.Query(ctx,
+		"SELECT id, name_zh, name_en, path::text, gpc_brick_code, level FROM category ORDER BY path")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := []Category{}
+	for rows.Next() {
+		var c Category
+		if err := rows.Scan(&c.ID, &c.NameZH, &c.NameEN, &c.Path, &c.GPCBrickCode, &c.Level); err != nil {
+			return nil, err
+		}
+		out = append(out, c)
+	}
+	return out, rows.Err()
+}
+
+// Source describes a data source with its license and trust weight.
+type Source struct {
+	ID          string  `json:"id"`
+	Name        string  `json:"name"`
+	Homepage    *string `json:"homepage"`
+	License     *string `json:"license"`
+	TrustWeight float64 `json:"trust_weight"`
+}
+
+// SourceByID fetches a single data source.
+func (s *Store) SourceByID(ctx context.Context, id string) (*Source, error) {
+	var src Source
+	err := s.pool.QueryRow(ctx,
+		"SELECT id, name, homepage, license, trust_weight FROM source WHERE id = $1", id,
+	).Scan(&src.ID, &src.Name, &src.Homepage, &src.License, &src.TrustWeight)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &src, nil
+}

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 集成测试在无库时自动跳过。

docs/ingestion-management.md

@@ -0,0 +1,76 @@
+# 采集管理 (M4)
+
+M4 在 M2（Open Food Facts 首次导入）基础上，补齐"持续运营"所需的采集能力：
+增量更新、第二数据源补全（GS1）、去重合并与字段级冲突解决、数据质量评分，
+以及把这些串起来的定时调度。全部为 Python 侧（`ingestion/`），只写库、可单测。
+
+## 组成
+
+| 能力 | 模块 | 说明 |
+|------|------|------|
+| 增量采集 | `adapters/openfoodfacts.py: fetch_modified_since()` | 按 `last_modified_t` 拉取自上次水位后变更的商品 |
+| 采集水位 | `etl/state.py` + `ingest_state` 表 | 每个源持久化 `last_modified_t`，只前进不回退 |
+| GS1 补全 | `adapters/gs1.py` + `etl/supplement.py` | 用权威条码源补**缺失**字段（品牌/厂商/GPC/产地/净含量），不覆盖已有值 |
+| 去重合并 | `etl/dedup.py` | 非 GTIN 重复（同名+品牌+净含量）合并到质量最高的主记录 |
+| 冲突解决 | `etl/merge.py` | 多源同字段按"源权重 > 新鲜度"择优，保留字段级溯源 |
+| 质量评分 | `etl/quality.py` | 0~1 分，落到 `product.quality_score` |
+| 定时调度 | `jobs/schedule.py` | 固定周期跑"增量 + 去重"一轮，零额外依赖 |
+
+## 质量评分
+
+锁定公式（各分量均归一到 0~1）：
+
+```
+quality = 0.4 * 完整度 + 0.3 * 源权重 + 0.2 * 多源一致 + 0.1 * 新鲜度
+```
+
+- **完整度**：`name/gtin/brand/category/net_content/country/nutriments/ingredients/image` 9 项的命中比例。
+- **源权重**：贡献该商品的源中最高 `source.trust_weight`（OFF=0.7，GS1=0.9）。
+- **多源一致**：源数量代理——单源 0.5、两源 0.8、三源及以上 1.0（单源无法互证）。
+- **新鲜度**：最近一次 `product_source.fetched_at` 的时间衰减（≤30d=1.0 … >730d=0.2）。
+
+`load_record()` 与 `merge_products()` 写入后都会调 `update_quality()` 重算。
+
+## 增量水位
+
+`ingest_state`（迁移 `0004`）每源一行，记录 `last_modified_t`、`last_run_at`、`stats`。
+`set_watermark()` 用 `GREATEST(...)` 保证水位只前进，避免乱序/中断的运行回退进度。
+
+## 运行
+
+前置：`docker compose up -d postgres` 且迁移已 `up`（含 `0004`）。DSN 默认读 `OPENGOODS_DATABASE_URL`。
+
+```bash
+# 增量更新 OFF（从持久化水位开始；--since 可覆盖）
+python -m opengoods.jobs.update_off --max-pages 5
+python -m opengoods.jobs.update_off --since 1700000000
+
+# 去重合并（--dry-run 只报告不写库）
+python -m opengoods.jobs.dedup --dry-run
+python -m opengoods.jobs.dedup --actor nightly
+
+# 定时调度：单轮 / 周期循环（增量 + 去重）
+python -m opengoods.jobs.schedule --once
+python -m opengoods.jobs.schedule --interval 3600
+```
+
+## GS1 补全
+
+GS1 为付费、分区域的授权数据，适配器支持两种模式：
+
+- **离线**（默认）：从本地 JSON 映射 `{gtin: {...}}` 查（`GS1Adapter.from_file(path)`），
+  供测试与内网环境使用。
+- **在线**：传 `base_url` + `client`（+ `api_key`），`GET {base_url}/{gtin}`，按
+  Verified-by-GS1 风格字段解析。
+
+补全只填**空缺**字段并在 `product_source` 记字段级溯源，源标记为 `gs1`。
+
+## 测试
+
+```bash
+cd ingestion && pip install -e ".[dev]"
+ruff check . && ruff format --check . && pytest -q
+```
+
+纯函数测试（质量/冲突/增量分页）始终运行；依赖库的测试（水位/质量落库/去重/GS1 补全）
+在无数据库或未应用 M4 迁移时自动跳过。

ingestion/opengoods/adapters/gs1.py

@@ -0,0 +1,124 @@
+"""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

ingestion/opengoods/adapters/openfoodfacts.py

@@ -0,0 +1,135 @@
+"""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"
+_SEARCH_URL = "https://world.openfoodfacts.org/api/v2/search"
+
+# Fields requested from the search API so a returned product can be transformed
+# without an extra per-barcode round trip.
+_SEARCH_FIELDS = (
+    "code,product_name,product_name_en,product_name_zh,brands,quantity,"
+    "categories,categories_tags,countries,ingredients_text,allergens_tags,"
+    "additives_tags,nutriments,nutriscore_grade,serving_size,"
+    "image_front_url,image_url,last_modified_t"
+)
+
+
+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 fetch_modified_since(
+        self,
+        since_t: int,
+        *,
+        page_size: int = 100,
+        max_pages: int = 10,
+    ) -> Iterator[dict]:
+        """Yield products modified after ``since_t`` (unix ``last_modified_t``).
+
+        Uses the OFF search API sorted by ``last_modified_t`` (most recent
+        first) and paginates until it reaches products at or before the
+        watermark, an empty/short page, or ``max_pages``. This is the
+        incremental ingestion path: callers persist the highest
+        ``last_modified_t`` they processed as the next watermark.
+        """
+        for page in range(1, max_pages + 1):
+            self._throttle()
+            resp = self._client.get(
+                _SEARCH_URL,
+                params={
+                    "fields": _SEARCH_FIELDS,
+                    "sort_by": "last_modified_t",
+                    "page": page,
+                    "page_size": page_size,
+                },
+            )
+            resp.raise_for_status()
+            products = resp.json().get("products") or []
+            if not products:
+                return
+            reached_old = False
+            for prod in products:
+                if int(prod.get("last_modified_t") or 0) <= since_t:
+                    reached_old = True
+                    break
+                yield prod
+            if reached_old or len(products) < page_size:
+                return
+
+
+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)

ingestion/opengoods/etl/dedup.py

@@ -0,0 +1,139 @@
+"""Duplicate detection and product merging.
+
+Barcodes (GTIN) are already unique at the schema level, so duplicates here are
+non-GTIN records that describe the same product (same normalized name + brand +
+net content). For each duplicate group we keep the highest-quality product as
+canonical and merge the rest into it: child rows (provenance, images, MSRP) are
+re-pointed to the canonical product, the merged product is marked ``merged``
+with ``canonical_id`` set, and a row is written to ``merge_log``.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import psycopg
+
+from opengoods.etl.quality import update_quality
+
+
+def _norm(text: str | None) -> str:
+    return " ".join((text or "").lower().split())
+
+
+def product_signature(name: str | None, brand: str | None, net_canonical: Any | None) -> str | None:
+    """Stable signature for non-GTIN dedup, or ``None`` if too sparse to match."""
+    n = _norm(name)
+    if not n:
+        return None
+    net = "" if net_canonical is None else str(net_canonical)
+    return f"{n}|{_norm(brand)}|{net}"
+
+
+def choose_canonical(members: list[dict]) -> dict:
+    """Pick the canonical product: best quality, then oldest, then lowest id."""
+    return min(
+        members,
+        key=lambda m: (
+            -float(m.get("quality_score") or 0.0),
+            m.get("created_at"),
+            str(m.get("id")),
+        ),
+    )
+
+
+def find_duplicate_groups(conn: psycopg.Connection) -> list[list[dict]]:
+    """Return groups (size >= 2) of active products sharing a signature."""
+    rows = conn.execute(
+        """
+        SELECT p.id, p.name, b.normalized_name, p.net_content_canonical,
+               p.quality_score, p.created_at
+        FROM product p
+        LEFT JOIN brand b ON b.id = p.brand_id
+        WHERE p.status = 'active'
+        """
+    ).fetchall()
+
+    groups: dict[str, list[dict]] = {}
+    for r in rows:
+        sig = product_signature(r[1], r[2], r[3])
+        if sig is None:
+            continue
+        member = {
+            "id": r[0],
+            "name": r[1],
+            "quality_score": r[4],
+            "created_at": r[5],
+        }
+        groups.setdefault(sig, []).append(member)
+
+    return [m for m in groups.values() if len(m) >= 2]
+
+
+def merge_products(
+    conn: psycopg.Connection,
+    kept_id: str,
+    merged_id: str,
+    reason: str = "auto-dedup",
+    actor: str = "ingestion",
+) -> None:
+    """Merge ``merged_id`` into ``kept_id`` (re-point children, mark merged)."""
+    if kept_id == merged_id:
+        return
+
+    # Re-point provenance, images and MSRP to the canonical product.
+    conn.execute(
+        "UPDATE product_source SET product_id = %s WHERE product_id = %s",
+        (kept_id, merged_id),
+    )
+    conn.execute(
+        "UPDATE product_image SET product_id = %s WHERE product_id = %s",
+        (kept_id, merged_id),
+    )
+    conn.execute(
+        "UPDATE product_msrp SET product_id = %s WHERE product_id = %s",
+        (kept_id, merged_id),
+    )
+
+    # food_detail has product_id as PK, so it can only move if the canonical
+    # product does not already have one.
+    kept_has_food = conn.execute(
+        "SELECT 1 FROM food_detail WHERE product_id = %s", (kept_id,)
+    ).fetchone()
+    if not kept_has_food:
+        conn.execute(
+            "UPDATE food_detail SET product_id = %s WHERE product_id = %s",
+            (kept_id, merged_id),
+        )
+
+    conn.execute(
+        "UPDATE product SET status = 'merged', canonical_id = %s WHERE id = %s",
+        (kept_id, merged_id),
+    )
+    conn.execute(
+        """
+        INSERT INTO merge_log (kept_id, merged_id, reason, actor)
+        VALUES (%s, %s, %s, %s)
+        """,
+        (kept_id, merged_id, reason, actor),
+    )
+
+    # The canonical product gained sources, so its quality may have changed.
+    update_quality(conn, kept_id)
+
+
+def dedup_all(
+    conn: psycopg.Connection, actor: str = "ingestion", dry_run: bool = False
+) -> dict[str, int]:
+    """Merge every duplicate group. Returns counts of groups and merges."""
+    groups = find_duplicate_groups(conn)
+    merged = 0
+    for members in groups:
+        canonical = choose_canonical(members)
+        for m in members:
+            if m["id"] == canonical["id"]:
+                continue
+            if not dry_run:
+                merge_products(conn, canonical["id"], m["id"], actor=actor)
+            merged += 1
+    return {"groups": len(groups), "merged": merged}

ingestion/opengoods/etl/load.py

@@ -0,0 +1,213 @@
+"""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
+from opengoods.etl.quality import update_quality
+
+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_named(
+    conn: psycopg.Connection,
+    name: str,
+    homepage: str,
+    license: str,
+    trust_weight: float,
+) -> str:
+    """Upsert a source row by name 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
+        """,
+        (name, homepage, license, trust_weight),
+    ).fetchone()
+    return row[0]
+
+
+def ensure_source(conn: psycopg.Connection) -> str:
+    """Upsert the Open Food Facts source row and return its id."""
+    return ensure_source_named(conn, SOURCE_NAME, OFF_HOMEPAGE, OFF_LICENSE, 0.7)
+
+
+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)),
+        ),
+    )
+
+    # Recompute the data-quality score now that all facts + provenance exist.
+    update_quality(conn, product_id)
+    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

ingestion/opengoods/etl/merge.py

@@ -0,0 +1,104 @@
+"""Field-level conflict resolution for multi-source records.
+
+When more than one source describes the same product, each field may have
+several candidate values. We pick a winner per field by source trust first,
+then recency, ignoring empty values, and keep a provenance trail of which
+source won each field.
+
+These are pure functions (no DB / no network) so they are easy to unit-test;
+the DB-level record merge lives in :mod:`opengoods.etl.dedup`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+
+@dataclass(frozen=True)
+class Candidate:
+    """One source's proposed value for a field."""
+
+    value: object
+    source: str
+    trust: float = 0.5
+    fetched_at: datetime | None = None
+
+
+@dataclass
+class FieldResolution:
+    """The winning value for a field plus the source it came from."""
+
+    value: object
+    source: str | None = None
+
+
+@dataclass
+class MergedRecord:
+    """A merged record with per-field provenance (field name -> source)."""
+
+    values: dict[str, object] = field(default_factory=dict)
+    provenance: dict[str, str] = field(default_factory=dict)
+
+
+def _is_empty(value: object) -> bool:
+    if value is None:
+        return True
+    if isinstance(value, str):
+        return value.strip() == ""
+    if isinstance(value, (list, dict, tuple, set)):
+        return len(value) == 0
+    return False
+
+
+def _sort_key(c: Candidate) -> tuple[float, float]:
+    ts = c.fetched_at.timestamp() if c.fetched_at is not None else float("-inf")
+    return (c.trust, ts)
+
+
+def resolve_field(candidates: list[Candidate]) -> FieldResolution | None:
+    """Pick the best non-empty candidate for one field.
+
+    Ranking: highest source trust, then most recent ``fetched_at``. Returns
+    ``None`` when there is no usable (non-empty) candidate.
+    """
+    usable = [c for c in candidates if not _is_empty(c.value)]
+    if not usable:
+        return None
+    winner = max(usable, key=_sort_key)
+    return FieldResolution(value=winner.value, source=winner.source)
+
+
+def merge_records(records: list[dict], *, fields: list[str] | None = None) -> MergedRecord:
+    """Merge several ``{field: Candidate|value}`` records into one.
+
+    Each input record maps field name -> :class:`Candidate` (preferred) or a
+    bare value (treated as trust 0.5, no timestamp). The result keeps, for each
+    field, the winning value and the name of the source that supplied it.
+    """
+    keys: list[str]
+    if fields is not None:
+        keys = list(fields)
+    else:
+        seen: dict[str, None] = {}
+        for rec in records:
+            for k in rec:
+                seen.setdefault(k, None)
+        keys = list(seen)
+
+    merged = MergedRecord()
+    for key in keys:
+        candidates: list[Candidate] = []
+        for rec in records:
+            if key not in rec:
+                continue
+            cand = rec[key]
+            if not isinstance(cand, Candidate):
+                cand = Candidate(value=cand, source="unknown")
+            candidates.append(cand)
+        resolution = resolve_field(candidates)
+        if resolution is not None:
+            merged.values[key] = resolution.value
+            if resolution.source is not None:
+                merged.provenance[key] = resolution.source
+    return merged

ingestion/opengoods/etl/quality.py

@@ -0,0 +1,176 @@
+"""Product data-quality scoring.
+
+The quality score is a 0..1 number combining four signals, per the locked
+project decision:
+
+    quality = 0.4 * completeness
+            + 0.3 * source_trust
+            + 0.2 * multi_source_agreement
+            + 0.1 * freshness
+
+Each component is itself normalized to 0..1. The pure helpers below are
+unit-testable; :func:`compute_quality` / :func:`update_quality` read the signals
+for a product out of the database and persist the result on ``product``.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import psycopg
+
+W_COMPLETENESS = 0.4
+W_SOURCE_TRUST = 0.3
+W_AGREEMENT = 0.2
+W_FRESHNESS = 0.1
+
+# Fields that count towards completeness (weighted equally).
+COMPLETENESS_FIELDS = (
+    "name",
+    "gtin",
+    "brand",
+    "category",
+    "net_content",
+    "country_of_origin",
+    "nutriments",
+    "ingredients",
+    "image",
+)
+
+
+def completeness(present: set[str]) -> float:
+    """Fraction of :data:`COMPLETENESS_FIELDS` that are present for a product."""
+    if not COMPLETENESS_FIELDS:
+        return 0.0
+    hits = sum(1 for f in COMPLETENESS_FIELDS if f in present)
+    return hits / len(COMPLETENESS_FIELDS)
+
+
+def agreement_from_sources(source_count: int) -> float:
+    """Multi-source corroboration proxy from the number of distinct sources.
+
+    A single source cannot be corroborated, so it scores a neutral 0.5; more
+    independent sources that describe the same product raise confidence.
+    """
+    if source_count <= 1:
+        return 0.5
+    if source_count == 2:
+        return 0.8
+    return 1.0
+
+
+def freshness_from_age(age_days: float | None) -> float:
+    """Recency score from the age (in days) of the most recent source fetch."""
+    if age_days is None:
+        return 0.5
+    if age_days <= 30:
+        return 1.0
+    if age_days <= 180:
+        return 0.8
+    if age_days <= 365:
+        return 0.6
+    if age_days <= 730:
+        return 0.4
+    return 0.2
+
+
+def score(
+    *,
+    completeness_score: float,
+    source_trust: float,
+    agreement: float,
+    freshness: float,
+) -> float:
+    """Combine the four normalized components into a 0..1 quality score."""
+    raw = (
+        W_COMPLETENESS * completeness_score
+        + W_SOURCE_TRUST * source_trust
+        + W_AGREEMENT * agreement
+        + W_FRESHNESS * freshness
+    )
+    return round(max(0.0, min(1.0, raw)), 3)
+
+
+def _present_fields(prod: dict, has_image: bool) -> set[str]:
+    present: set[str] = set()
+    if prod.get("name"):
+        present.add("name")
+    if prod.get("gtin"):
+        present.add("gtin")
+    if prod.get("brand_id"):
+        present.add("brand")
+    if prod.get("category_id"):
+        present.add("category")
+    if prod.get("net_content_canonical") is not None:
+        present.add("net_content")
+    if prod.get("country_of_origin"):
+        present.add("country_of_origin")
+    if prod.get("nutriments"):
+        present.add("nutriments")
+    if prod.get("ingredients_text"):
+        present.add("ingredients")
+    if has_image:
+        present.add("image")
+    return present
+
+
+def compute_quality(conn: psycopg.Connection, product_id: str) -> float:
+    """Compute (but do not persist) the quality score for one product."""
+    row = conn.execute(
+        """
+        SELECT p.name, p.gtin, p.brand_id, p.category_id, p.net_content_canonical,
+               p.country_of_origin, f.nutriments, f.ingredients_text,
+               EXISTS (SELECT 1 FROM product_image pi WHERE pi.product_id = p.id)
+        FROM product p
+        LEFT JOIN food_detail f ON f.product_id = p.id
+        WHERE p.id = %s
+        """,
+        (product_id,),
+    ).fetchone()
+    if row is None:
+        return 0.0
+    prod = {
+        "name": row[0],
+        "gtin": row[1],
+        "brand_id": row[2],
+        "category_id": row[3],
+        "net_content_canonical": row[4],
+        "country_of_origin": row[5],
+        "nutriments": row[6],
+        "ingredients_text": row[7],
+    }
+    has_image = bool(row[8])
+
+    src = conn.execute(
+        """
+        SELECT count(DISTINCT ps.source_id), COALESCE(max(s.trust_weight), 0), max(ps.fetched_at)
+        FROM product_source ps
+        LEFT JOIN source s ON s.id = ps.source_id
+        WHERE ps.product_id = %s
+        """,
+        (product_id,),
+    ).fetchone()
+    source_count = int(src[0] or 0)
+    source_trust = float(src[1] or 0.0)
+    last_fetched: datetime | None = src[2]
+
+    age_days: float | None = None
+    if last_fetched is not None:
+        now = datetime.now(UTC)
+        if last_fetched.tzinfo is None:
+            last_fetched = last_fetched.replace(tzinfo=UTC)
+        age_days = max(0.0, (now - last_fetched).total_seconds() / 86400.0)
+
+    return score(
+        completeness_score=completeness(_present_fields(prod, has_image)),
+        source_trust=source_trust,
+        agreement=agreement_from_sources(source_count),
+        freshness=freshness_from_age(age_days),
+    )
+
+
+def update_quality(conn: psycopg.Connection, product_id: str) -> float:
+    """Compute the quality score and write it to ``product.quality_score``."""
+    value = compute_quality(conn, product_id)
+    conn.execute("UPDATE product SET quality_score = %s WHERE id = %s", (value, product_id))
+    return value

ingestion/opengoods/etl/state.py

@@ -0,0 +1,45 @@
+"""Persistent ingestion watermark stored in the ``ingest_state`` table.
+
+The incremental updater uses this to remember how far it got for each source
+(e.g. Open Food Facts exposes a ``last_modified_t`` unix timestamp on every
+product) so repeated runs only fetch what changed.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import psycopg
+from psycopg.types.json import Jsonb
+
+
+def get_watermark(conn: psycopg.Connection, source: str) -> int:
+    """Return the last processed ``last_modified_t`` for *source* (0 if none)."""
+    row = conn.execute(
+        "SELECT last_modified_t FROM ingest_state WHERE source = %s", (source,)
+    ).fetchone()
+    return int(row[0]) if row else 0
+
+
+def set_watermark(
+    conn: psycopg.Connection,
+    source: str,
+    last_modified_t: int,
+    stats: dict[str, Any] | None = None,
+) -> None:
+    """Upsert the watermark and run metadata for *source*.
+
+    The watermark only ever moves forward: a lower ``last_modified_t`` is
+    ignored so an out-of-order or partial run cannot rewind progress.
+    """
+    conn.execute(
+        """
+        INSERT INTO ingest_state (source, last_modified_t, last_run_at, stats)
+        VALUES (%s, %s, now(), %s)
+        ON CONFLICT (source) DO UPDATE SET
+            last_modified_t = GREATEST(ingest_state.last_modified_t, EXCLUDED.last_modified_t),
+            last_run_at = now(),
+            stats = EXCLUDED.stats
+        """,
+        (source, int(last_modified_t), Jsonb(stats or {})),
+    )

ingestion/opengoods/etl/supplement.py

@@ -0,0 +1,133 @@
+"""Apply GS1 (or other authoritative) supplements to existing products.
+
+A supplement only fills *gaps*: a field is written only when the product does
+not already have a value. Each applied supplement records field-level provenance
+in ``product_source`` and refreshes the product's quality score.
+"""
+
+from __future__ import annotations
+
+from decimal import Decimal, InvalidOperation
+from typing import Any
+
+import psycopg
+from psycopg.types.json import Jsonb
+
+from opengoods import units
+from opengoods.adapters.gs1 import GS1_HOMEPAGE, GS1_LICENSE, GS1_TRUST, SOURCE_NAME
+from opengoods.etl.load import _ensure_brand, _normalize_brand, ensure_source_named
+from opengoods.etl.quality import update_quality
+
+
+def ensure_gs1_source(conn: psycopg.Connection) -> str:
+    """Upsert the GS1 source row and return its id."""
+    return ensure_source_named(conn, SOURCE_NAME, GS1_HOMEPAGE, GS1_LICENSE, GS1_TRUST)
+
+
+def _ensure_manufacturer(conn: psycopg.Connection, name: str | None) -> str | None:
+    if not name:
+        return None
+    row = conn.execute(
+        """
+        INSERT INTO manufacturer (name, normalized_name)
+        VALUES (%s, %s)
+        ON CONFLICT (normalized_name) DO UPDATE SET name = manufacturer.name
+        RETURNING id
+        """,
+        (name, _normalize_brand(name)),
+    ).fetchone()
+    return row[0]
+
+
+def _net_content(rec: dict) -> tuple[Decimal, str, Decimal | None] | None:
+    raw_value = rec.get("net_content_value")
+    unit = rec.get("net_content_unit")
+    if raw_value is None or not unit:
+        return None
+    try:
+        value = Decimal(str(raw_value))
+    except (InvalidOperation, ValueError):
+        return None
+    try:
+        canonical = units.normalize(value, unit).canonical_value
+    except units.UnitError:
+        canonical = None
+    return value, unit, canonical
+
+
+def apply_supplement(conn: psycopg.Connection, rec: dict[str, Any], source_id: str) -> list[str]:
+    """Fill missing fields of the GTIN-matched product from ``rec``.
+
+    Returns the list of field names actually filled (empty if the product is
+    unknown or already complete for the supplied fields).
+    """
+    gtin = rec.get("gtin")
+    if not gtin:
+        return []
+    prod = conn.execute(
+        """
+        SELECT id, brand_id, manufacturer_id, gpc_brick_code, country_of_origin,
+               net_content_value
+        FROM product
+        WHERE gtin = %s AND status = 'active'
+        """,
+        (gtin,),
+    ).fetchone()
+    if prod is None:
+        return []
+
+    product_id, brand_id, manufacturer_id, gpc, country, net_value = prod
+    sets: list[str] = []
+    params: list[Any] = []
+    filled: list[str] = []
+
+    if brand_id is None and rec.get("brand"):
+        new_brand_id = _ensure_brand(conn, rec["brand"])
+        if new_brand_id is not None:
+            sets.append("brand_id = %s")
+            params.append(new_brand_id)
+            filled.append("brand")
+
+    if manufacturer_id is None and rec.get("manufacturer"):
+        new_mfr_id = _ensure_manufacturer(conn, rec["manufacturer"])
+        if new_mfr_id is not None:
+            sets.append("manufacturer_id = %s")
+            params.append(new_mfr_id)
+            filled.append("manufacturer")
+
+    if gpc is None and rec.get("gpc_brick_code"):
+        sets.append("gpc_brick_code = %s")
+        params.append(rec["gpc_brick_code"])
+        filled.append("gpc_brick_code")
+
+    if country is None and rec.get("country_of_origin"):
+        sets.append("country_of_origin = %s")
+        params.append(rec["country_of_origin"])
+        filled.append("country_of_origin")
+
+    if net_value is None:
+        net = _net_content(rec)
+        if net is not None:
+            value, unit, canonical = net
+            sets += [
+                "net_content_value = %s",
+                "net_content_unit = %s",
+                "net_content_canonical = %s",
+            ]
+            params += [value, unit, canonical]
+            filled.append("net_content")
+
+    if not filled:
+        return []
+
+    params.append(product_id)
+    conn.execute(f"UPDATE product SET {', '.join(sets)} WHERE id = %s", params)
+    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, GS1_HOMEPAGE, filled, Jsonb(rec)),
+    )
+    update_quality(conn, product_id)
+    return filled

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<value>\d+(?:[.,]\d+)?)\s*(?P<unit>[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,
+    }

ingestion/opengoods/jobs/dedup.py

@@ -0,0 +1,40 @@
+"""Deduplicate products: merge non-GTIN duplicates into a canonical record.
+
+Usage:
+    python -m opengoods.jobs.dedup --dry-run
+    python -m opengoods.jobs.dedup --actor nightly
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import psycopg
+
+from opengoods.etl.dedup import dedup_all
+from opengoods.etl.load import default_dsn
+
+
+def run(args: argparse.Namespace) -> int:
+    with psycopg.connect(args.dsn, autocommit=False) as conn:
+        summary = dedup_all(conn, actor=args.actor, dry_run=args.dry_run)
+        if args.dry_run:
+            conn.rollback()
+        else:
+            conn.commit()
+    mode = "dry-run" if args.dry_run else "applied"
+    print(f"{mode} groups={summary['groups']} merged={summary['merged']}")
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Deduplicate OpenGoods products")
+    parser.add_argument("--actor", default="ingestion", help="merge_log actor label")
+    parser.add_argument("--dry-run", action="store_true", help="report only, do not write")
+    parser.add_argument("--dsn", default=default_dsn(), help="PostgreSQL DSN")
+    return run(parser.parse_args(argv))
+
+
+if __name__ == "__main__":
+    sys.exit(main())

ingestion/opengoods/jobs/schedule.py

@@ -0,0 +1,66 @@
+"""Lightweight recurring ingestion scheduler.
+
+Runs one ingestion cycle (incremental OFF update, then dedup) on a fixed
+interval. Dependency-free: a plain sleep loop rather than a cron/APScheduler
+dependency, so it is trivial to run in a container or under systemd/supervisor.
+
+Usage:
+    python -m opengoods.jobs.schedule --once            # single cycle, then exit
+    python -m opengoods.jobs.schedule --interval 3600   # every hour
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+import time
+from datetime import UTC, datetime
+
+from opengoods.etl.load import default_dsn
+from opengoods.jobs import dedup as dedup_job
+from opengoods.jobs import update_off as update_job
+
+
+def _cycle(args: argparse.Namespace) -> None:
+    ts = datetime.now(UTC).isoformat(timespec="seconds")
+    print(f"[{ts}] cycle start")
+    update_job.run(
+        argparse.Namespace(
+            since=None,
+            page_size=args.page_size,
+            max_pages=args.max_pages,
+            min_interval=args.min_interval,
+            dsn=args.dsn,
+        )
+    )
+    if not args.skip_dedup:
+        dedup_job.run(argparse.Namespace(actor="scheduler", dry_run=False, dsn=args.dsn))
+
+
+def run(args: argparse.Namespace) -> int:
+    _cycle(args)
+    if args.once:
+        return 0
+    while True:
+        time.sleep(args.interval)
+        try:
+            _cycle(args)
+        except Exception as exc:  # noqa: BLE001 - keep the loop alive across failures
+            print(f"cycle error: {exc}", file=sys.stderr)
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Recurring OpenGoods ingestion")
+    parser.add_argument("--interval", type=int, default=3600, help="seconds between cycles")
+    parser.add_argument("--once", action="store_true", help="run a single cycle and exit")
+    parser.add_argument("--skip-dedup", action="store_true", help="run update only")
+    parser.add_argument("--page-size", type=int, default=100, help="search page size")
+    parser.add_argument("--max-pages", type=int, default=10, help="max pages to scan")
+    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())

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())

ingestion/opengoods/jobs/update_off.py

@@ -0,0 +1,65 @@
+"""Incremental Open Food Facts update.
+
+Fetches products modified since the persisted watermark, loads them, then
+advances the watermark to the newest ``last_modified_t`` processed so the next
+run only sees what changed.
+
+Usage:
+    python -m opengoods.jobs.update_off --max-pages 5
+    python -m opengoods.jobs.update_off --since 1700000000   # override watermark
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import psycopg
+
+from opengoods.adapters.openfoodfacts import SOURCE_NAME, OpenFoodFactsAdapter
+from opengoods.etl.load import default_dsn, ensure_source, load_record
+from opengoods.etl.state import get_watermark, set_watermark
+from opengoods.etl.transform import transform
+
+
+def run(args: argparse.Namespace) -> int:
+    adapter = OpenFoodFactsAdapter(min_interval=args.min_interval)
+    loaded = skipped = 0
+    high_watermark = 0
+    with psycopg.connect(args.dsn, autocommit=False) as conn:
+        source_id = ensure_source(conn)
+        since = args.since if args.since is not None else get_watermark(conn, SOURCE_NAME)
+        high_watermark = since
+        for raw in adapter.fetch_modified_since(
+            since, page_size=args.page_size, max_pages=args.max_pages
+        ):
+            high_watermark = max(high_watermark, int(raw.get("last_modified_t") or 0))
+            rec = transform(raw)
+            if rec is None:
+                skipped += 1
+                continue
+            load_record(conn, rec, source_id, raw)
+            loaded += 1
+        set_watermark(
+            conn,
+            SOURCE_NAME,
+            high_watermark,
+            stats={"loaded": loaded, "skipped": skipped, "since": since},
+        )
+        conn.commit()
+    print(f"since={since} loaded={loaded} skipped={skipped} watermark={high_watermark}")
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Incremental OFF update")
+    parser.add_argument("--since", type=int, default=None, help="override watermark (unix ts)")
+    parser.add_argument("--page-size", type=int, default=100, help="search page size")
+    parser.add_argument("--max-pages", type=int, default=10, help="max pages to scan")
+    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())

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]

ingestion/tests/conftest.py

@@ -0,0 +1,31 @@
+"""Shared test fixtures.
+
+`db_conn` yields a psycopg connection inside a transaction that is rolled back
+after each test, so DB tests stay isolated and leave no residue. Tests are
+skipped automatically when no database is reachable or M4 migrations are not
+applied (e.g. local runs without docker).
+"""
+
+from __future__ import annotations
+
+import psycopg
+import pytest
+
+from opengoods.etl.load import default_dsn
+
+
+@pytest.fixture()
+def db_conn():
+    try:
+        conn = psycopg.connect(default_dsn(), connect_timeout=3)
+    except psycopg.OperationalError as exc:  # pragma: no cover - env dependent
+        pytest.skip(f"no database available: {exc}")
+    has_state = conn.execute("SELECT to_regclass('public.ingest_state') IS NOT NULL").fetchone()[0]
+    if not has_state:
+        conn.close()
+        pytest.skip("M4 migrations not applied")
+    try:
+        yield conn
+    finally:
+        conn.rollback()
+        conn.close()

ingestion/tests/fixtures/gs1_mapping.json

@@ -0,0 +1,14 @@
+{
+  "06901234567892": {
+    "name": "示例矿泉水 550ml",
+    "brand": "示例品牌",
+    "manufacturer": "示例饮品有限公司",
+    "gpc_brick_code": "10000224",
+    "country_of_origin": "China",
+    "net_content_value": 550,
+    "net_content_unit": "ml"
+  },
+  "00000000000000": {
+    "brand": ""
+  }
+}

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
+  }
+}

ingestion/tests/test_dedup.py

@@ -0,0 +1,62 @@
+from opengoods.etl.dedup import choose_canonical, dedup_all, product_signature
+from opengoods.etl.load import _ensure_brand, ensure_source
+
+
+def test_product_signature_normalization():
+    a = product_signature("  Spring  Water ", "Acme", 500)
+    b = product_signature("spring water", "acme", 500)
+    assert a == b
+    assert product_signature("", "x", 1) is None
+
+
+def test_choose_canonical_prefers_quality():
+    members = [
+        {"id": "a", "quality_score": 0.2, "created_at": 1},
+        {"id": "b", "quality_score": 0.9, "created_at": 2},
+    ]
+    assert choose_canonical(members)["id"] == "b"
+
+
+def test_dedup_merges_duplicates(db_conn):
+    brand_id = _ensure_brand(db_conn, "DupBrand")
+
+    def mk(quality):
+        return db_conn.execute(
+            """
+            INSERT INTO product (name, brand_id, net_content_canonical, quality_score)
+            VALUES (%s, %s, %s, %s) RETURNING id
+            """,
+            ("Dup Snack", brand_id, 100, quality),
+        ).fetchone()[0]
+
+    keep = mk(0.9)
+    drop = mk(0.2)
+
+    src = ensure_source(db_conn)
+    db_conn.execute(
+        "INSERT INTO product_source (product_id, source_id, fields) VALUES (%s, %s, %s)",
+        (drop, src, ["name"]),
+    )
+
+    summary = dedup_all(db_conn)
+    assert summary == {"groups": 1, "merged": 1}
+
+    keep_status = db_conn.execute("SELECT status FROM product WHERE id = %s", (keep,)).fetchone()[0]
+    drop_status, canonical_id = db_conn.execute(
+        "SELECT status, canonical_id FROM product WHERE id = %s", (drop,)
+    ).fetchone()
+    assert keep_status == "active"
+    assert drop_status == "merged"
+    assert str(canonical_id) == str(keep)
+
+    # The merged product's source row was re-pointed to the canonical product.
+    reattached = db_conn.execute(
+        "SELECT count(*) FROM product_source WHERE product_id = %s", (keep,)
+    ).fetchone()[0]
+    assert reattached == 1
+
+    logged = db_conn.execute(
+        "SELECT count(*) FROM merge_log WHERE kept_id = %s AND merged_id = %s",
+        (keep, drop),
+    ).fetchone()[0]
+    assert logged == 1

ingestion/tests/test_gs1.py

@@ -0,0 +1,53 @@
+from pathlib import Path
+
+from opengoods.adapters.gs1 import GS1Adapter
+from opengoods.etl.supplement import apply_supplement, ensure_gs1_source
+
+MAPPING = Path(__file__).parent / "fixtures" / "gs1_mapping.json"
+GTIN = "06901234567892"
+
+
+def test_gs1_adapter_offline_lookup():
+    adapter = GS1Adapter.from_file(MAPPING)
+    rec = adapter.fetch_barcode(GTIN)
+    assert rec["brand"] == "示例品牌"
+    assert rec["net_content_value"] == 550
+    assert rec["net_content_unit"] == "ml"
+    # An entry that only has empty values yields no supplement.
+    assert adapter.fetch_barcode("00000000000000") is None
+    # Unknown barcode -> None.
+    assert adapter.fetch_barcode("99999999999999") is None
+
+
+def test_gs1_supplement_fills_only_gaps(db_conn):
+    pid = db_conn.execute(
+        "INSERT INTO product (gtin, name) VALUES (%s, %s) RETURNING id", (GTIN, "水")
+    ).fetchone()[0]
+
+    adapter = GS1Adapter.from_file(MAPPING)
+    rec = adapter.fetch_barcode(GTIN)
+    source_id = ensure_gs1_source(db_conn)
+
+    filled = apply_supplement(db_conn, rec, source_id)
+    assert {"brand", "country_of_origin", "net_content"} <= set(filled)
+
+    brand_id, country, net_value, net_unit = db_conn.execute(
+        """
+        SELECT brand_id, country_of_origin, net_content_value, net_content_unit
+        FROM product WHERE id = %s
+        """,
+        (pid,),
+    ).fetchone()
+    assert brand_id is not None
+    assert country == "China"
+    assert float(net_value) == 550.0
+    assert net_unit == "ml"
+
+    fields = db_conn.execute(
+        "SELECT fields FROM product_source WHERE product_id = %s AND source_id = %s",
+        (pid, source_id),
+    ).fetchone()[0]
+    assert "brand" in fields
+
+    # Re-applying does nothing because the gaps are now filled.
+    assert apply_supplement(db_conn, rec, source_id) == []

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

ingestion/tests/test_merge.py

@@ -0,0 +1,62 @@
+from datetime import UTC, datetime
+
+from opengoods.etl.merge import Candidate, merge_records, resolve_field
+
+
+def _ts(y, m, d):
+    return datetime(y, m, d, tzinfo=UTC)
+
+
+def test_resolve_field_prefers_trust_then_recency():
+    cands = [
+        Candidate(value="A", source="off", trust=0.7, fetched_at=_ts(2024, 1, 1)),
+        Candidate(value="B", source="gs1", trust=0.9, fetched_at=_ts(2023, 1, 1)),
+    ]
+    res = resolve_field(cands)
+    assert res is not None
+    assert res.value == "B"
+    assert res.source == "gs1"
+
+
+def test_resolve_field_recency_tiebreak_on_equal_trust():
+    cands = [
+        Candidate(value="old", source="a", trust=0.7, fetched_at=_ts(2023, 1, 1)),
+        Candidate(value="new", source="b", trust=0.7, fetched_at=_ts(2024, 6, 1)),
+    ]
+    assert resolve_field(cands).value == "new"
+
+
+def test_resolve_field_skips_empty():
+    cands = [
+        Candidate(value="", source="a", trust=0.99),
+        Candidate(value=None, source="b", trust=0.99),
+        Candidate(value="kept", source="c", trust=0.1),
+    ]
+    assert resolve_field(cands).value == "kept"
+    assert resolve_field([Candidate(value="", source="a")]) is None
+
+
+def test_merge_records_provenance():
+    records = [
+        {
+            "name": Candidate("Water", "off", 0.7, _ts(2024, 1, 1)),
+            "brand": Candidate("", "off", 0.7),
+        },
+        {
+            "brand": Candidate("Acme", "gs1", 0.9, _ts(2024, 2, 1)),
+            "gtin": Candidate("123", "gs1", 0.9),
+        },
+    ]
+    merged = merge_records(records)
+    assert merged.values["name"] == "Water"
+    assert merged.values["brand"] == "Acme"
+    assert merged.values["gtin"] == "123"
+    assert merged.provenance["brand"] == "gs1"
+    assert merged.provenance["name"] == "off"
+
+
+def test_merge_records_accepts_bare_values():
+    merged = merge_records([{"x": 1}, {"x": 2}])
+    # both bare -> trust tie, no timestamps -> first max() wins deterministically
+    assert merged.values["x"] in (1, 2)
+    assert merged.provenance["x"] == "unknown"

ingestion/tests/test_off_incremental.py

@@ -0,0 +1,44 @@
+import httpx
+
+from opengoods.adapters.openfoodfacts import OpenFoodFactsAdapter
+
+
+def _product(code, lm):
+    return {"code": code, "product_name": f"P{code}", "last_modified_t": lm}
+
+
+def _adapter(pages):
+    """Build an adapter whose search endpoint serves the given pages."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        page = int(request.url.params.get("page", "1"))
+        products = pages.get(page, [])
+        return httpx.Response(200, json={"products": products, "page": page})
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    return OpenFoodFactsAdapter(client=client, min_interval=0)
+
+
+def test_incremental_yields_only_newer_and_stops_at_watermark():
+    pages = {
+        1: [_product("1", 300), _product("2", 250), _product("3", 100)],
+    }
+    adapter = _adapter(pages)
+    got = list(adapter.fetch_modified_since(200, page_size=3, max_pages=5))
+    codes = [p["code"] for p in got]
+    assert codes == ["1", "2"]  # 100 <= 200 stops iteration
+
+
+def test_incremental_paginates_until_short_page():
+    pages = {
+        1: [_product("1", 900), _product("2", 800)],
+        2: [_product("3", 700)],  # short page -> stop after
+    }
+    adapter = _adapter(pages)
+    got = list(adapter.fetch_modified_since(0, page_size=2, max_pages=5))
+    assert [p["code"] for p in got] == ["1", "2", "3"]
+
+
+def test_incremental_empty_first_page():
+    adapter = _adapter({1: []})
+    assert list(adapter.fetch_modified_since(0, page_size=10, max_pages=3)) == []

ingestion/tests/test_quality.py

@@ -0,0 +1,38 @@
+from opengoods.etl.quality import (
+    COMPLETENESS_FIELDS,
+    agreement_from_sources,
+    completeness,
+    freshness_from_age,
+    score,
+)
+
+
+def test_completeness_bounds():
+    assert completeness(set()) == 0.0
+    assert completeness(set(COMPLETENESS_FIELDS)) == 1.0
+    half = set(list(COMPLETENESS_FIELDS)[: len(COMPLETENESS_FIELDS) // 2])
+    assert 0.0 < completeness(half) < 1.0
+
+
+def test_agreement_from_sources():
+    assert agreement_from_sources(0) == 0.5
+    assert agreement_from_sources(1) == 0.5
+    assert agreement_from_sources(2) == 0.8
+    assert agreement_from_sources(5) == 1.0
+
+
+def test_freshness_from_age():
+    assert freshness_from_age(None) == 0.5
+    assert freshness_from_age(1) == 1.0
+    assert freshness_from_age(100) == 0.8
+    assert freshness_from_age(300) == 0.6
+    assert freshness_from_age(700) == 0.4
+    assert freshness_from_age(5000) == 0.2
+
+
+def test_score_weighted_sum_and_bounds():
+    assert score(completeness_score=0, source_trust=0, agreement=0, freshness=0) == 0.0
+    assert score(completeness_score=1, source_trust=1, agreement=1, freshness=1) == 1.0
+    # 0.4*1 + 0.3*0.5 + 0.2*0.5 + 0.1*1 = 0.75
+    got = score(completeness_score=1.0, source_trust=0.5, agreement=0.5, freshness=1.0)
+    assert got == 0.75

ingestion/tests/test_quality_db.py

@@ -0,0 +1,21 @@
+import json
+from pathlib import Path
+
+from opengoods.etl.load import ensure_source, load_record
+from opengoods.etl.quality import compute_quality
+from opengoods.etl.transform import transform
+
+FIXTURE = json.loads((Path(__file__).parent / "fixtures" / "off_product.json").read_text())
+
+
+def test_quality_score_set_on_load(db_conn):
+    source_id = ensure_source(db_conn)
+    rec = transform(FIXTURE)
+    pid = load_record(db_conn, rec, source_id, FIXTURE)
+
+    stored = float(
+        db_conn.execute("SELECT quality_score FROM product WHERE id = %s", (pid,)).fetchone()[0]
+    )
+    assert 0.0 < stored <= 1.0
+    # The persisted value matches a fresh recomputation.
+    assert abs(stored - compute_quality(db_conn, pid)) < 1e-9

ingestion/tests/test_state.py

@@ -0,0 +1,16 @@
+from opengoods.etl.state import get_watermark, set_watermark
+
+
+def test_watermark_roundtrip_and_monotonic(db_conn):
+    src = "test-source"
+    assert get_watermark(db_conn, src) == 0
+
+    set_watermark(db_conn, src, 100, stats={"loaded": 1})
+    assert get_watermark(db_conn, src) == 100
+
+    # A lower watermark must not rewind progress.
+    set_watermark(db_conn, src, 50)
+    assert get_watermark(db_conn, src) == 100
+
+    set_watermark(db_conn, src, 150)
+    assert get_watermark(db_conn, src) == 150

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

migrations/0001_init.down.sql

@@ -0,0 +1,21 @@
+DROP TRIGGER IF EXISTS trg_product_sync ON product;
+DROP FUNCTION IF EXISTS product_sync_tsv();
+
+DROP TABLE IF EXISTS merge_log;
+DROP TABLE IF EXISTS product_source;
+DROP TABLE IF EXISTS product_image;
+DROP TABLE IF EXISTS product_msrp;
+DROP TABLE IF EXISTS food_detail;
+DROP TABLE IF EXISTS product;
+DROP TABLE IF EXISTS attribute_definition;
+DROP TABLE IF EXISTS unit;
+DROP TABLE IF EXISTS category_schema;
+DROP TABLE IF EXISTS category;
+DROP TABLE IF EXISTS manufacturer;
+DROP TABLE IF EXISTS brand;
+DROP TABLE IF EXISTS source;
+
+DROP EXTENSION IF EXISTS ltree;
+DROP EXTENSION IF EXISTS pg_trgm;
+-- keep pgcrypto (commonly shared); drop only if you are sure:
+-- DROP EXTENSION IF EXISTS pgcrypto;

migrations/0001_init.up.sql

@@ -0,0 +1,182 @@
+-- OpenGoods (天工·商品标签) initial schema.
+-- Public-good product information store: facts only, no commerce.
+
+CREATE EXTENSION IF NOT EXISTS pgcrypto;   -- gen_random_uuid()
+CREATE EXTENSION IF NOT EXISTS pg_trgm;     -- fuzzy name search
+CREATE EXTENSION IF NOT EXISTS ltree;       -- category subtree queries
+
+-- Data sources (Open Food Facts / USDA / GS1 ...) with trust + license.
+CREATE TABLE source (
+    id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    name         TEXT NOT NULL,
+    homepage     TEXT,
+    license      TEXT,
+    trust_weight NUMERIC(3,2) NOT NULL DEFAULT 0.5,
+    notes        TEXT,
+    UNIQUE (name)
+);
+
+CREATE TABLE brand (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    name            TEXT NOT NULL,
+    normalized_name TEXT NOT NULL,
+    aliases         TEXT[] NOT NULL DEFAULT '{}',
+    UNIQUE (normalized_name)
+);
+
+CREATE TABLE manufacturer (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    name            TEXT NOT NULL,
+    normalized_name TEXT NOT NULL,
+    country         VARCHAR(64),
+    UNIQUE (normalized_name)
+);
+
+-- Self-built category tree, each node optionally mapped to a GS1 GPC brick.
+CREATE TABLE category (
+    id             UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    name_zh        TEXT NOT NULL,
+    name_en        TEXT,
+    parent_id      UUID REFERENCES category(id),
+    path           LTREE NOT NULL,
+    gpc_brick_code VARCHAR(10),
+    level          INT NOT NULL DEFAULT 0,
+    UNIQUE (path)
+);
+
+-- Parameter template / constraints per category.
+CREATE TABLE category_schema (
+    category_id           UUID PRIMARY KEY REFERENCES category(id) ON DELETE CASCADE,
+    required_attributes   TEXT[] NOT NULL DEFAULT '{}',
+    recommended_attributes TEXT[] NOT NULL DEFAULT '{}',
+    nutriment_basis       VARCHAR(16)
+);
+
+-- Unit dictionary: each unit maps to a canonical unit within its dimension.
+CREATE TABLE unit (
+    code                VARCHAR(16) PRIMARY KEY,
+    dimension           VARCHAR(16) NOT NULL,
+    canonical           VARCHAR(16) NOT NULL,
+    to_canonical_factor NUMERIC,
+    aliases             TEXT[] NOT NULL DEFAULT '{}',
+    display             TEXT
+);
+
+-- Parameter dictionary: standard attribute keys with default unit.
+CREATE TABLE attribute_definition (
+    key          VARCHAR(64) PRIMARY KEY,
+    label_zh     TEXT,
+    label_en     TEXT,
+    dimension    VARCHAR(16),
+    default_unit VARCHAR(16) REFERENCES unit(code),
+    aliases      TEXT[] NOT NULL DEFAULT '{}'
+);
+
+CREATE TABLE product (
+    id                    UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    gtin                  VARCHAR(14),
+    name                  TEXT NOT NULL,
+    brand_id              UUID REFERENCES brand(id),
+    manufacturer_id       UUID REFERENCES manufacturer(id),
+    category_id           UUID REFERENCES category(id),
+    gpc_brick_code        VARCHAR(10),
+    net_content_value     NUMERIC,
+    net_content_unit      VARCHAR(16),
+    net_content_canonical NUMERIC,
+    country_of_origin     VARCHAR(64),
+    shelf_life_days       INT,
+    storage               TEXT,
+    attributes            JSONB NOT NULL DEFAULT '{}',
+    quality_score         NUMERIC(4,3) NOT NULL DEFAULT 0,
+    status                VARCHAR(16) NOT NULL DEFAULT 'active',
+    canonical_id          UUID REFERENCES product(id),
+    search_tsv            TSVECTOR,
+    created_at            TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at            TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT product_status_chk CHECK (status IN ('active','merged','deprecated')),
+    CONSTRAINT product_quality_chk CHECK (quality_score >= 0 AND quality_score <= 1)
+);
+
+CREATE TABLE food_detail (
+    product_id       UUID PRIMARY KEY REFERENCES product(id) ON DELETE CASCADE,
+    ingredients_text TEXT,
+    ingredients      JSONB,
+    allergens        TEXT[] NOT NULL DEFAULT '{}',
+    additives        TEXT[] NOT NULL DEFAULT '{}',
+    nutriments       JSONB,
+    nutrition_basis  VARCHAR(16),
+    serving_size     VARCHAR(32),
+    nutri_score      CHAR(1),
+    labels           TEXT[] NOT NULL DEFAULT '{}',
+    CONSTRAINT food_basis_chk CHECK (nutrition_basis IS NULL OR nutrition_basis IN ('per_100g','per_100ml','per_serving'))
+);
+
+-- Official manufacturer-suggested retail price snapshot (no purchase link).
+CREATE TABLE product_msrp (
+    id             UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    product_id     UUID NOT NULL REFERENCES product(id) ON DELETE CASCADE,
+    amount         NUMERIC(12,2) NOT NULL,
+    currency       CHAR(3) NOT NULL,
+    region         VARCHAR(8) NOT NULL DEFAULT 'CN',
+    source_id      UUID REFERENCES source(id),
+    source_url     TEXT,
+    effective_date DATE,
+    note           TEXT,
+    created_at     TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE product_image (
+    id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    product_id UUID NOT NULL REFERENCES product(id) ON DELETE CASCADE,
+    url        TEXT NOT NULL,
+    kind       VARCHAR(16) NOT NULL DEFAULT 'other',
+    license    TEXT,
+    source_id  UUID REFERENCES source(id),
+    CONSTRAINT image_kind_chk CHECK (kind IN ('front','ingredients','nutrition','other'))
+);
+
+-- Field-level provenance: which source provided which fields.
+CREATE TABLE product_source (
+    id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    product_id UUID NOT NULL REFERENCES product(id) ON DELETE CASCADE,
+    source_id  UUID REFERENCES source(id),
+    url        TEXT,
+    fields     TEXT[] NOT NULL DEFAULT '{}',
+    fetched_at TIMESTAMPTZ,
+    raw        JSONB
+);
+
+CREATE TABLE merge_log (
+    id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    kept_id    UUID,
+    merged_id  UUID,
+    reason     TEXT,
+    actor      TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Indexes
+CREATE UNIQUE INDEX idx_product_gtin ON product (gtin) WHERE gtin IS NOT NULL;
+CREATE INDEX idx_product_name_trgm ON product USING gin (name gin_trgm_ops);
+CREATE INDEX idx_product_attrs ON product USING gin (attributes);
+CREATE INDEX idx_product_tsv ON product USING gin (search_tsv);
+CREATE INDEX idx_product_category ON product (category_id);
+CREATE INDEX idx_product_brand ON product (brand_id);
+CREATE INDEX idx_product_updated ON product (updated_at);
+CREATE INDEX idx_food_nutriments ON food_detail USING gin (nutriments);
+CREATE INDEX idx_category_path ON category USING gist (path);
+CREATE INDEX idx_msrp_product ON product_msrp (product_id);
+CREATE INDEX idx_psource_product ON product_source (product_id);
+
+-- Keep search_tsv and updated_at in sync.
+CREATE OR REPLACE FUNCTION product_sync_tsv() RETURNS trigger AS $$
+BEGIN
+    NEW.search_tsv := to_tsvector('simple', coalesce(NEW.name, ''));
+    NEW.updated_at := now();
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE TRIGGER trg_product_sync
+    BEFORE INSERT OR UPDATE ON product
+    FOR EACH ROW EXECUTE FUNCTION product_sync_tsv();

migrations/0002_seed_units.down.sql

@@ -0,0 +1,2 @@
+DELETE FROM attribute_definition;
+DELETE FROM unit;

migrations/0002_seed_units.up.sql

@@ -0,0 +1,29 @@
+-- Unit dictionary seed. Keep factors aligned with ingestion/opengoods/units.py.
+
+INSERT INTO unit (code, dimension, canonical, to_canonical_factor, aliases, display) VALUES
+    ('mg',   'mass',     'g',   0.001,  ARRAY['毫克'],                 'mg'),
+    ('g',    'mass',     'g',   1,      ARRAY['克','gram','grams'],    'g'),
+    ('kg',   'mass',     'g',   1000,   ARRAY['kgs','千克','公斤'],     'kg'),
+    ('ml',   'volume',   'ml',  1,      ARRAY['毫升','milliliter'],     'mL'),
+    ('cl',   'volume',   'ml',  10,     ARRAY['厘升'],                 'cL'),
+    ('l',    'volume',   'ml',  1000,   ARRAY['L','升','litre','liter'], 'L'),
+    ('kj',   'energy',   'kJ',  1,      ARRAY['kJ','千焦'],            'kJ'),
+    ('kcal', 'energy',   'kJ',  4.184,  ARRAY['千卡','大卡'],          'kcal'),
+    ('pct',  'ratio',    'pct', 1,      ARRAY['%','percent','百分比'],  '%'),
+    ('unit', 'count',    'unit',1,      ARRAY['个','件','pcs','piece'], '个'),
+    ('mm',   'length',   'mm',  1,      ARRAY['毫米'],                 'mm'),
+    ('cm',   'length',   'mm',  10,     ARRAY['厘米'],                 'cm'),
+    ('day',  'duration', 'day', 1,      ARRAY['天','日','days'],        'day')
+ON CONFLICT (code) DO NOTHING;
+
+-- A few common food attribute definitions referencing the unit dictionary.
+INSERT INTO attribute_definition (key, label_zh, label_en, dimension, default_unit, aliases) VALUES
+    ('energy',        '能量',   'Energy',        'energy', 'kj',   ARRAY['energy_kj']),
+    ('proteins',      '蛋白质', 'Proteins',      'mass',   'g',    ARRAY['protein']),
+    ('fat',           '脂肪',   'Fat',           'mass',   'g',    ARRAY['fats']),
+    ('saturated_fat', '饱和脂肪','Saturated fat','mass',   'g',    ARRAY['saturated-fat']),
+    ('carbohydrates', '碳水化合物','Carbohydrates','mass', 'g',    ARRAY['carbs']),
+    ('sugars',        '糖',     'Sugars',        'mass',   'g',    ARRAY['sugar']),
+    ('salt',          '盐',     'Salt',          'mass',   'g',    ARRAY['sodium_salt']),
+    ('net_content',   '净含量', 'Net content',   NULL,     NULL,   ARRAY['quantity'])
+ON CONFLICT (key) DO NOTHING;

migrations/0003_seed_categories.down.sql

@@ -0,0 +1,3 @@
+-- remove seeded categories (children first via path depth)
+DELETE FROM category_schema;
+DELETE FROM category;

migrations/0003_seed_categories.up.sql

@@ -0,0 +1,53 @@
+-- Seed a FOOD-focused category skeleton.
+-- Structure = GS1 GPC backbone (segment/family/class) mapped to a self-built
+-- Chinese tree. ltree labels are english slugs (ltree forbids spaces/CJK);
+-- Chinese names live in name_zh. gpc_brick_code on leaves is a representative
+-- starter value to be replaced by a full official GPC import later.
+
+-- Root segment: Food/Beverage/Tobacco (GPC segment 50000000)
+INSERT INTO category (name_zh, name_en, parent_id, path, gpc_brick_code, level)
+VALUES ('食品饮料', 'Food/Beverage', NULL, 'food', '50000000', 0);
+
+-- Families (level 1)
+INSERT INTO category (name_zh, name_en, parent_id, path, gpc_brick_code, level)
+SELECT v.name_zh, v.name_en, c.id, v.path::ltree, v.code, 1
+FROM (VALUES
+    ('饮料',     'Beverages',        'food.beverages',  '50130000'),
+    ('乳制品蛋类','Dairy/Eggs',       'food.dairy',      '50180000'),
+    ('烘焙',     'Bakery',           'food.bakery',     '50100000'),
+    ('零食',     'Snacks',           'food.snacks',     '50190000'),
+    ('粮油',     'Staples/Oils',     'food.staple',     '50160000'),
+    ('调味品',   'Condiments',       'food.condiments', '50170000')
+) AS v(name_zh, name_en, path, code)
+JOIN category c ON c.path = 'food';
+
+-- Classes / leaves (level 2) with representative GPC brick codes
+INSERT INTO category (name_zh, name_en, parent_id, path, gpc_brick_code, level)
+SELECT v.name_zh, v.name_en, c.id, v.path::ltree, v.code, 2
+FROM (VALUES
+    ('包装饮用水', 'Bottled water',   'food.beverages.water',      '10000224', 'food.beverages'),
+    ('碳酸饮料',   'Carbonated',      'food.beverages.carbonated', '10000225', 'food.beverages'),
+    ('果汁',       'Juice',           'food.beverages.juice',      '10000226', 'food.beverages'),
+    ('牛奶',       'Milk',            'food.dairy.milk',           '10000158', 'food.dairy'),
+    ('酸奶',       'Yogurt',          'food.dairy.yogurt',         '10000159', 'food.dairy'),
+    ('奶酪',       'Cheese',          'food.dairy.cheese',         '10000160', 'food.dairy'),
+    ('面包',       'Bread',           'food.bakery.bread',         '10000040', 'food.bakery'),
+    ('饼干',       'Biscuits',        'food.bakery.biscuits',      '10000041', 'food.bakery'),
+    ('薯片膨化',   'Chips/Snacks',    'food.snacks.chips',         '10000310', 'food.snacks'),
+    ('巧克力',     'Chocolate',       'food.snacks.chocolate',     '10000311', 'food.snacks'),
+    ('大米',       'Rice',            'food.staple.rice',          '10000500', 'food.staple'),
+    ('面条',       'Noodles',         'food.staple.noodles',       '10000501', 'food.staple'),
+    ('食用油',     'Cooking oil',     'food.staple.cooking_oil',   '10000502', 'food.staple'),
+    ('酱油',       'Soy sauce',       'food.condiments.soy_sauce', '10000600', 'food.condiments'),
+    ('食盐',       'Table salt',      'food.condiments.salt',      '10000601', 'food.condiments')
+) AS v(name_zh, name_en, path, code, parent_path)
+JOIN category c ON c.path = v.parent_path::ltree;
+
+-- Parameter templates: leaf food categories use per_100g/ml nutrition basis.
+INSERT INTO category_schema (category_id, required_attributes, recommended_attributes, nutriment_basis)
+SELECT id,
+       ARRAY['net_content'],
+       ARRAY['energy','proteins','fat','carbohydrates','sugars','salt'],
+       CASE WHEN path <@ 'food.beverages' THEN 'per_100ml' ELSE 'per_100g' END
+FROM category
+WHERE level = 2;

migrations/0004_ingest_state.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS ingest_state;

migrations/0004_ingest_state.up.sql

@@ -0,0 +1,10 @@
+-- M4 ingestion management: persistent per-source incremental watermark.
+-- The updater reads/writes one row per source to resume incremental imports
+-- (e.g. Open Food Facts `last_modified_t`) and to record run statistics.
+CREATE TABLE ingest_state (
+    source          TEXT PRIMARY KEY,
+    last_modified_t BIGINT NOT NULL DEFAULT 0,
+    last_run_at     TIMESTAMPTZ,
+    cursor          TEXT,
+    stats           JSONB NOT NULL DEFAULT '{}'
+);

migrations/README.md

@@ -1,3 +1,28 @@
 # Database migrations (golang-migrate)
 
-SQL migrations live here from milestone M1. Format: `NNNN_description.up.sql` / `.down.sql`.
+SQL migrations for the OpenGoods database, applied with
+[golang-migrate](https://github.com/golang-migrate/migrate).
+Naming: `NNNN_description.up.sql` / `NNNN_description.down.sql`.
+
+## Files
+| Version | Up | 内容 |
+|---------|----|------|
+| 0001 | `0001_init` | 扩展(pgcrypto/pg_trgm/ltree) + 全部核心表 + 索引 + tsvector 触发器 |
+| 0002 | `0002_seed_units` | 单位字典（与 `ingestion/opengoods/units.py` 一致）+ 常用营养参数定义 |
+| 0003 | `0003_seed_categories` | 食品品类骨架（GS1 GPC 映射 + 自建中文树）+ 品类参数模板 |
+
+## 运行
+
+先起本地依赖：`docker compose up -d postgres`
+
+```bash
+export DBURL="postgres://opengoods:opengoods@localhost:5432/opengoods?sslmode=disable"
+migrate -path migrations -database "$DBURL" up        # 升级到最新
+migrate -path migrations -database "$DBURL" down -all # 全部回滚
+migrate -path migrations -database "$DBURL" version   # 查看当前版本
+```
+
+安装 CLI：`go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.18.1`
+
+> ltree 标签为英文 slug（不支持空格/中文），中文名存于 `category.name_zh`。
+> `gpc_brick_code` 为食品子集的代表值，后续用官方 GPC 全量导入替换。