Meal-Planner/backend/app/services/recipe_catalog_audit.py

"""Deterministic provenance and quality audit for the published recipe catalog."""
from __future__ import annotations

from datetime import datetime, timezone
from decimal import Decimal
from typing import Any

from sqlalchemy import and_
from sqlalchemy.orm import Session

from app.models import (
    Ingredient,
    Recipe,
    RecipeCatalogStatus,
    RecipeImportStaging,
    RecipeImportStatus,
)
from app.services.wikibooks_import import WIKIBOOKS_SOURCE, WikibooksImportService


OPERATOR_QUARANTINE_FINDING_CODE = "operator_quarantine"


def audit_recipe_catalog(
    db: Session,
    *,
    apply: bool = False,
    reviewed_at: datetime | None = None,
) -> dict[str, Any]:
    """Audit every recipe and optionally apply active/quarantined state changes."""
    timestamp = reviewed_at or datetime.now(timezone.utc)
    ingredient_names = {
        str(ingredient_id): name
        for ingredient_id, name in db.query(Ingredient.id, Ingredient.name).all()
    }
    catalog_query = (
        db.query(
            Recipe,
            RecipeImportStaging.source,
            RecipeImportStaging.external_id,
            RecipeImportStaging.source_url,
        )
        .outerjoin(
            RecipeImportStaging,
            and_(
                RecipeImportStaging.recipe_id == Recipe.id,
                RecipeImportStaging.status == RecipeImportStatus.APPROVED,
            ),
        )
        .order_by(Recipe.id, RecipeImportStaging.id)
    )
    if apply:
        catalog_query = catalog_query.with_for_update(of=Recipe)
    recipes: dict[str, Recipe] = {}
    approved_links: dict[str, set[tuple[str, str, str]]] = {}
    for recipe, source, external_id, source_url in catalog_query.all():
        recipe_id = str(recipe.id)
        recipes[recipe_id] = recipe
        if source is not None and external_id is not None and source_url is not None:
            approved_links.setdefault(recipe_id, set()).add(
                (source, external_id, source_url)
            )
    validator = WikibooksImportService("MealPlanner catalog audit")
    records: list[dict[str, Any]] = []

    for recipe_id, recipe in recipes.items():
        prior_findings = (
            recipe.catalog_findings
            if isinstance(recipe.catalog_findings, list)
            else []
        )
        operator_quarantine_findings = [
            finding
            for finding in prior_findings
            if isinstance(finding, dict)
            and finding.get("code") == OPERATOR_QUARANTINE_FINDING_CODE
        ]
        provenance, findings = _provenance_findings(
            recipe, approved_links.get(recipe_id, set())
        )
        candidate, reference_findings = _validation_candidate(recipe, ingredient_names)
        findings.extend(reference_findings)
        validation = validator.validate_recipe(candidate)
        findings.extend(
            {**issue, "severity": severity}
            for severity, key in (("error", "errors"), ("warning", "warnings"))
            for issue in validation[key]
        )
        findings.extend(operator_quarantine_findings)
        target_status = (
            RecipeCatalogStatus.QUARANTINED
            if operator_quarantine_findings
            or any(item["severity"] == "error" for item in findings)
            else RecipeCatalogStatus.ACTIVE
        )
        if apply:
            recipe.catalog_status = target_status
            recipe.catalog_findings = findings
            recipe.catalog_reviewed_at = timestamp
        records.append(
            {
                "id": str(recipe.id),
                "name": recipe.name,
                "provenance": provenance,
                "catalog_status": target_status.value,
                "findings": findings,
            }
        )

    if apply:
        db.flush()
    quarantined = sum(
        record["catalog_status"] == RecipeCatalogStatus.QUARANTINED.value
        for record in records
    )
    return {
        "version": 1,
        "generated_at": timestamp.isoformat(),
        "applied": apply,
        "total": len(records),
        "summary": {
            "active": len(records) - quarantined,
            "quarantined": quarantined,
        },
        "recipes": records,
    }


def _provenance_findings(
    recipe: Recipe, approved_links: set[tuple[str, str, str]]
) -> tuple[str, list[dict[str, str]]]:
    if recipe.is_manually_added and not recipe.external_source:
        return "manual", []
    if recipe.external_source == WIKIBOOKS_SOURCE:
        findings = []
        expected_link = (recipe.external_source, recipe.external_id, recipe.source_url)
        if expected_link not in approved_links:
            findings.append(
                _finding(
                    "missing_import_provenance",
                    "external_source",
                    "Published Wikibooks recipe is not linked to an approved staged import.",
                )
            )
        if not recipe.source_url:
            findings.append(
                _finding("missing_source_url", "source_url", "External recipe has no source URL.")
            )
        return "staged_wikibooks", findings
    if recipe.external_source == "spoonacular":
        return "spoonacular", [
            _finding(
                "unlicensed_external_source",
                "external_source",
                "Spoonacular data was persisted without durable catalog rights.",
            )
        ]
    if recipe.external_source:
        return "external_unverified", [
            _finding(
                "unverified_external_source",
                "external_source",
                "External recipe has no approved staged license record.",
            )
        ]
    return "missing", [
        _finding(
            "missing_provenance",
            "external_source",
            "Non-manual recipe has no auditable source or license record.",
        )
    ]


def _validation_candidate(
    recipe: Recipe, ingredient_names: dict[str, str]
) -> tuple[dict[str, Any], list[dict[str, str]]]:
    ingredients: list[dict[str, Any]] = []
    findings: list[dict[str, str]] = []
    for index, item in enumerate(recipe.ingredients or []):
        if not isinstance(item, dict):
            findings.append(
                _finding(
                    "invalid_ingredient_reference",
                    f"ingredients[{index}]",
                    "Ingredient reference must be an object.",
                )
            )
            continue
        ingredient_id = str(item.get("ingredient_id") or "")
        name = ingredient_names.get(ingredient_id)
        if not name:
            findings.append(
                _finding(
                    "orphan_ingredient_reference",
                    f"ingredients[{index}].ingredient_id",
                    "Ingredient reference does not resolve to the canonical ingredient table.",
                )
            )
            name = str(item.get("name") or "")
        quantity = item.get("qty")
        if isinstance(quantity, Decimal):
            quantity = float(quantity)
        ingredients.append({"name": name, "quantity": quantity})
    return (
        {
            "name": recipe.name,
            "description": recipe.description,
            "servings": recipe.servings,
            "prep_time_minutes": recipe.prep_time_minutes,
            "cook_time_minutes": recipe.cook_time_minutes,
            "baking_time_minutes": None,
            "ingredients": ingredients,
            "instructions": list(recipe.instructions or []),
        },
        findings,
    )


def _finding(code: str, field: str, message: str) -> dict[str, str]:
    return {"code": code, "severity": "error", "field": field, "message": message}