goods/ingestion/opengoods/etl/quality.py

"""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