Public Access
541 lines
20 KiB
Python
541 lines
20 KiB
Python
"""Validate and atomically load the bundled starter recipe catalog."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import Ingredient, Recipe, RecipeCatalogStatus
|
|
|
|
|
|
STARTER_CATALOG_VERSION = 2
|
|
STARTER_RECIPE_COUNT = 50
|
|
STARTER_DISCOVERY_REASON = "bundled_starter_catalog_v1"
|
|
# Fixed PostgreSQL bigint advisory-lock key for this loader (ASCII: MPLSTART).
|
|
# Changing it would allow concurrent catalog loads to bypass each other.
|
|
STARTER_CATALOG_ADVISORY_LOCK_ID = 0x4D504C5354415254
|
|
DEFAULT_CATALOG_PATH = Path(__file__).resolve().parents[1] / "data" / "starter_recipes.json"
|
|
|
|
MAX_SERVINGS = 50
|
|
MAX_TIME_MINUTES = 1_440
|
|
MAX_INGREDIENT_QUANTITY = 10_000
|
|
MAX_CALORIES_PER_SERVING = 500
|
|
MAX_SPICE_LEVEL = 5
|
|
_PLACEHOLDER_TEXT = frozenset(
|
|
{"coming soon", "directions", "instructions", "method", "n a", "tbd", "todo"}
|
|
)
|
|
_PLACEHOLDER_PHRASES = ("lorem ipsum", "placeholder")
|
|
_RECIPE_KEYS = frozenset(
|
|
{
|
|
"name",
|
|
"description",
|
|
"servings",
|
|
"prep_time_minutes",
|
|
"cook_time_minutes",
|
|
"cuisine_tags",
|
|
"dietary_tags",
|
|
"protein_type",
|
|
"spice_level",
|
|
"calories_per_serving",
|
|
"ingredients",
|
|
"instructions",
|
|
}
|
|
)
|
|
_INGREDIENT_KEYS = frozenset({"name", "quantity", "unit"})
|
|
_CATALOG_SCALAR_FIELDS = (
|
|
"description",
|
|
"servings",
|
|
"prep_time_minutes",
|
|
"cook_time_minutes",
|
|
"cuisine_tags",
|
|
"dietary_tags",
|
|
"protein_type",
|
|
"spice_level",
|
|
"calories_per_serving",
|
|
)
|
|
|
|
|
|
class StarterRecipeCatalogError(ValueError):
|
|
"""Raised when the bundled catalog is incomplete or unsafe to load."""
|
|
|
|
|
|
def read_starter_catalog(path: Path = DEFAULT_CATALOG_PATH) -> Mapping[str, Any]:
|
|
"""Read the JSON catalog without accepting non-standard numeric constants."""
|
|
try:
|
|
with path.open(encoding="utf-8") as catalog_file:
|
|
catalog = json.load(catalog_file, parse_constant=_reject_json_constant)
|
|
except OSError as exc:
|
|
raise StarterRecipeCatalogError(f"Catalog cannot be read: {path}") from exc
|
|
except ValueError as exc:
|
|
raise StarterRecipeCatalogError("Catalog is not valid JSON") from exc
|
|
if not isinstance(catalog, Mapping):
|
|
raise StarterRecipeCatalogError("Catalog root must be an object")
|
|
return catalog
|
|
|
|
|
|
def validate_starter_catalog(catalog: Mapping[str, Any]) -> list[dict[str, Any]]:
|
|
"""Return normalized recipes after validating the complete, fixed-size catalog."""
|
|
_require_exact_keys(catalog, {"version", "recipes"}, "catalog")
|
|
if catalog["version"] != STARTER_CATALOG_VERSION:
|
|
raise StarterRecipeCatalogError(
|
|
f"Catalog version must be {STARTER_CATALOG_VERSION}"
|
|
)
|
|
recipes = catalog["recipes"]
|
|
if not isinstance(recipes, list) or len(recipes) != STARTER_RECIPE_COUNT:
|
|
raise StarterRecipeCatalogError(
|
|
f"Catalog must contain exactly {STARTER_RECIPE_COUNT} recipes"
|
|
)
|
|
|
|
normalized: list[dict[str, Any]] = []
|
|
names: set[str] = set()
|
|
for index, recipe in enumerate(recipes, 1):
|
|
if not isinstance(recipe, Mapping):
|
|
raise StarterRecipeCatalogError(f"Recipe {index} must be an object")
|
|
_require_exact_keys(recipe, _RECIPE_KEYS, f"Recipe {index}")
|
|
name = _text(recipe["name"], f"Recipe {index} name", maximum=300)
|
|
name_key = _canonical_name(name)
|
|
if name_key in names:
|
|
raise StarterRecipeCatalogError(f"Recipe {index} name is not unique")
|
|
names.add(name_key)
|
|
|
|
ingredients = _validate_ingredients(recipe["ingredients"], index)
|
|
instructions = _validate_instructions(recipe["instructions"], index)
|
|
normalized.append(
|
|
{
|
|
"name": name,
|
|
"description": _text(
|
|
recipe["description"], f"Recipe {index} description", maximum=10_000
|
|
),
|
|
"servings": _positive_int(
|
|
recipe["servings"], f"Recipe {index} servings", MAX_SERVINGS
|
|
),
|
|
"prep_time_minutes": _positive_int(
|
|
recipe["prep_time_minutes"],
|
|
f"Recipe {index} prep_time_minutes",
|
|
MAX_TIME_MINUTES,
|
|
),
|
|
"cook_time_minutes": _positive_int(
|
|
recipe["cook_time_minutes"],
|
|
f"Recipe {index} cook_time_minutes",
|
|
MAX_TIME_MINUTES,
|
|
),
|
|
"cuisine_tags": _tags(recipe["cuisine_tags"], f"Recipe {index} cuisine_tags"),
|
|
"dietary_tags": _tags(recipe["dietary_tags"], f"Recipe {index} dietary_tags"),
|
|
"protein_type": _text(
|
|
recipe["protein_type"], f"Recipe {index} protein_type", maximum=50
|
|
),
|
|
"spice_level": _bounded_int(
|
|
recipe["spice_level"], f"Recipe {index} spice_level", 0, MAX_SPICE_LEVEL
|
|
),
|
|
"calories_per_serving": _positive_int(
|
|
recipe["calories_per_serving"],
|
|
f"Recipe {index} calories_per_serving",
|
|
MAX_CALORIES_PER_SERVING,
|
|
),
|
|
"ingredients": ingredients,
|
|
"instructions": instructions,
|
|
}
|
|
)
|
|
return normalized
|
|
|
|
|
|
def load_bundled_starter_catalog(
|
|
db: Session,
|
|
*,
|
|
path: Path = DEFAULT_CATALOG_PATH,
|
|
dry_run: bool = False,
|
|
update_existing: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Load the bundled JSON catalog after validating all recipes before writes."""
|
|
return load_starter_catalog(
|
|
db,
|
|
read_starter_catalog(path),
|
|
dry_run=dry_run,
|
|
update_existing=update_existing,
|
|
)
|
|
|
|
|
|
def load_starter_catalog(
|
|
db: Session,
|
|
catalog: Mapping[str, Any],
|
|
*,
|
|
dry_run: bool = False,
|
|
update_existing: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Load or explicitly synchronize bundled recipes in one transaction."""
|
|
recipes = validate_starter_catalog(catalog)
|
|
catalog_names = {recipe["name"] for recipe in recipes}
|
|
catalog_by_name = {recipe["name"]: recipe for recipe in recipes}
|
|
ingredient_names = {
|
|
_canonical_name(ingredient["name"]): ingredient
|
|
for recipe in recipes
|
|
for ingredient in recipe["ingredients"]
|
|
}
|
|
|
|
with db.begin():
|
|
# This transaction-scoped lock must precede every catalog Recipe read or write.
|
|
db.execute(select(func.pg_advisory_xact_lock(STARTER_CATALOG_ADVISORY_LOCK_ID)))
|
|
marked_recipes = list(
|
|
db.execute(
|
|
select(Recipe)
|
|
.where(Recipe.discovery_reason == STARTER_DISCOVERY_REASON)
|
|
.with_for_update()
|
|
).scalars()
|
|
)
|
|
marked_by_name = {recipe.name: recipe for recipe in marked_recipes}
|
|
if len(marked_by_name) != len(marked_recipes):
|
|
raise StarterRecipeCatalogError("Bundled catalog has duplicate persisted recipe names")
|
|
unknown_marked_names = set(marked_by_name).difference(catalog_names)
|
|
if unknown_marked_names:
|
|
raise StarterRecipeCatalogError(
|
|
"Bundled catalog marker is attached to an unknown recipe"
|
|
)
|
|
resolved_recipe_ingredients = _resolve_recipe_ingredients(db, marked_recipes)
|
|
missing_marked_names = catalog_names.difference(marked_by_name)
|
|
if update_existing and missing_marked_names:
|
|
raise StarterRecipeCatalogError(
|
|
"Update requires exactly all "
|
|
f"{STARTER_RECIPE_COUNT} existing marker-owned recipe names"
|
|
)
|
|
|
|
recipes_to_update: list[Recipe] = []
|
|
for recipe in marked_recipes:
|
|
expected = catalog_by_name[recipe.name]
|
|
if update_existing:
|
|
_assert_bundled_recipe_ownership(recipe)
|
|
if not _bundled_recipe_content_matches(
|
|
recipe, expected, resolved_recipe_ingredients
|
|
):
|
|
recipes_to_update.append(recipe)
|
|
else:
|
|
_assert_bundled_recipe(
|
|
recipe,
|
|
expected,
|
|
resolved_recipe_ingredients,
|
|
)
|
|
|
|
name_collisions = list(
|
|
db.execute(
|
|
select(Recipe).where(Recipe.name.in_(catalog_names)).with_for_update()
|
|
).scalars()
|
|
)
|
|
if any(recipe.discovery_reason != STARTER_DISCOVERY_REASON for recipe in name_collisions):
|
|
raise StarterRecipeCatalogError(
|
|
"Catalog recipe name conflicts with an existing non-bundled recipe"
|
|
)
|
|
|
|
canonical_ingredients = {
|
|
ingredient.name_lower: ingredient
|
|
for ingredient in db.execute(
|
|
select(Ingredient).where(Ingredient.name_lower.in_(ingredient_names))
|
|
).scalars()
|
|
}
|
|
new_recipes = (
|
|
[]
|
|
if update_existing
|
|
else [recipe for recipe in recipes if recipe["name"] not in marked_by_name]
|
|
)
|
|
summary = {
|
|
"catalog_version": STARTER_CATALOG_VERSION,
|
|
"dry_run": dry_run,
|
|
"validated_recipes": len(recipes),
|
|
"existing_recipes": len(marked_recipes),
|
|
"created_recipes": len(new_recipes),
|
|
"updated_recipes": len(recipes_to_update),
|
|
"existing_ingredients": len(canonical_ingredients),
|
|
"created_ingredients": len(ingredient_names) - len(canonical_ingredients),
|
|
}
|
|
if dry_run:
|
|
return summary
|
|
|
|
for name_lower, ingredient in ingredient_names.items():
|
|
db.execute(
|
|
pg_insert(Ingredient.__table__)
|
|
.values(
|
|
name=ingredient["name"],
|
|
name_lower=name_lower,
|
|
unit=ingredient["unit"],
|
|
aliases=[],
|
|
)
|
|
.on_conflict_do_nothing(index_elements=[Ingredient.name_lower])
|
|
)
|
|
canonical_ingredients = {
|
|
ingredient.name_lower: ingredient
|
|
for ingredient in db.execute(
|
|
select(Ingredient).where(Ingredient.name_lower.in_(ingredient_names))
|
|
).scalars()
|
|
}
|
|
if set(canonical_ingredients) != set(ingredient_names):
|
|
raise StarterRecipeCatalogError("Canonical ingredient creation failed")
|
|
|
|
for recipe in recipes_to_update:
|
|
_synchronize_bundled_recipe(
|
|
recipe,
|
|
catalog_by_name[recipe.name],
|
|
canonical_ingredients,
|
|
)
|
|
|
|
for recipe in new_recipes:
|
|
db.add(
|
|
Recipe(
|
|
family_profile_id=None,
|
|
name=recipe["name"],
|
|
description=recipe["description"],
|
|
image_url=None,
|
|
image_source=None,
|
|
prep_time_minutes=recipe["prep_time_minutes"],
|
|
cook_time_minutes=recipe["cook_time_minutes"],
|
|
servings=recipe["servings"],
|
|
cuisine_tags=recipe["cuisine_tags"],
|
|
dietary_tags=recipe["dietary_tags"],
|
|
protein_type=recipe["protein_type"],
|
|
spice_level=recipe["spice_level"],
|
|
calories_per_serving=recipe["calories_per_serving"],
|
|
ingredients=_catalog_recipe_ingredients(recipe, canonical_ingredients),
|
|
side_dishes=[],
|
|
instructions=recipe["instructions"],
|
|
source_url=None,
|
|
scraped_at=None,
|
|
is_manually_added=True,
|
|
external_source=None,
|
|
external_id=None,
|
|
discovery_reason=STARTER_DISCOVERY_REASON,
|
|
catalog_status=RecipeCatalogStatus.ACTIVE,
|
|
catalog_findings=[],
|
|
)
|
|
)
|
|
db.flush()
|
|
return summary
|
|
|
|
|
|
def _validate_ingredients(value: Any, recipe_index: int) -> list[dict[str, Any]]:
|
|
if not isinstance(value, list) or len(value) < 2:
|
|
raise StarterRecipeCatalogError(f"Recipe {recipe_index} needs at least two ingredients")
|
|
ingredients: list[dict[str, Any]] = []
|
|
for index, ingredient in enumerate(value, 1):
|
|
if not isinstance(ingredient, Mapping):
|
|
raise StarterRecipeCatalogError(
|
|
f"Recipe {recipe_index} ingredient {index} must be an object"
|
|
)
|
|
_require_exact_keys(
|
|
ingredient,
|
|
_INGREDIENT_KEYS,
|
|
f"Recipe {recipe_index} ingredient {index}",
|
|
)
|
|
ingredients.append(
|
|
{
|
|
"name": _text(
|
|
ingredient["name"],
|
|
f"Recipe {recipe_index} ingredient {index} name",
|
|
maximum=200,
|
|
),
|
|
"quantity": _positive_number(
|
|
ingredient["quantity"],
|
|
f"Recipe {recipe_index} ingredient {index} quantity",
|
|
MAX_INGREDIENT_QUANTITY,
|
|
),
|
|
"unit": _text(
|
|
ingredient["unit"], f"Recipe {recipe_index} ingredient {index} unit", maximum=50
|
|
),
|
|
}
|
|
)
|
|
return ingredients
|
|
|
|
|
|
def _validate_instructions(value: Any, recipe_index: int) -> list[str]:
|
|
if not isinstance(value, list) or len(value) < 2:
|
|
raise StarterRecipeCatalogError(f"Recipe {recipe_index} needs at least two instructions")
|
|
instructions = [
|
|
_text(instruction, f"Recipe {recipe_index} instruction {index}", maximum=5_000)
|
|
for index, instruction in enumerate(value, 1)
|
|
]
|
|
for instruction in instructions:
|
|
if len(instruction) < 20 or len(re.findall(r"[A-Za-z0-9]+", instruction)) < 4:
|
|
raise StarterRecipeCatalogError(
|
|
f"Recipe {recipe_index} instructions must be substantive"
|
|
)
|
|
return instructions
|
|
|
|
|
|
def _tags(value: Any, field: str) -> list[str]:
|
|
if not isinstance(value, list):
|
|
raise StarterRecipeCatalogError(f"{field} must be an array")
|
|
return [_text(tag, field, maximum=50) for tag in value]
|
|
|
|
|
|
def _text(value: Any, field: str, *, maximum: int) -> str:
|
|
if not isinstance(value, str):
|
|
raise StarterRecipeCatalogError(f"{field} must be text")
|
|
text = " ".join(value.split())
|
|
if not text or len(text) > maximum or _is_placeholder(text):
|
|
raise StarterRecipeCatalogError(f"{field} is missing, too long, or placeholder text")
|
|
return text
|
|
|
|
|
|
def _positive_int(value: Any, field: str, maximum: int) -> int:
|
|
return _bounded_int(value, field, 1, maximum)
|
|
|
|
|
|
def _bounded_int(value: Any, field: str, minimum: int, maximum: int) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or not minimum <= value <= maximum:
|
|
raise StarterRecipeCatalogError(f"{field} must be an integer between {minimum} and {maximum}")
|
|
return value
|
|
|
|
|
|
def _positive_number(value: Any, field: str, maximum: int) -> int | float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise StarterRecipeCatalogError(f"{field} must be a number")
|
|
if not math.isfinite(value) or not 0 < value <= maximum:
|
|
raise StarterRecipeCatalogError(f"{field} must be greater than zero and at most {maximum}")
|
|
return value
|
|
|
|
|
|
def _require_exact_keys(
|
|
value: Mapping[str, Any], expected: set[str] | frozenset[str], field: str
|
|
) -> None:
|
|
if set(value) != set(expected):
|
|
raise StarterRecipeCatalogError(f"{field} has unexpected or missing fields")
|
|
|
|
|
|
def _canonical_name(value: str) -> str:
|
|
return " ".join(value.split()).casefold()
|
|
|
|
|
|
def _is_placeholder(value: str) -> bool:
|
|
normalized = re.sub(r"[^a-z0-9]+", " ", value.casefold()).strip()
|
|
return normalized in _PLACEHOLDER_TEXT or any(
|
|
phrase in normalized for phrase in _PLACEHOLDER_PHRASES
|
|
)
|
|
|
|
|
|
def _resolve_recipe_ingredients(
|
|
db: Session, recipes: list[Recipe]
|
|
) -> dict[str, Ingredient]:
|
|
ingredient_ids: set[UUID] = set()
|
|
for recipe in recipes:
|
|
if not isinstance(recipe.ingredients, list):
|
|
continue
|
|
for item in recipe.ingredients:
|
|
if not isinstance(item, Mapping):
|
|
continue
|
|
try:
|
|
ingredient_ids.add(UUID(str(item.get("ingredient_id"))))
|
|
except (TypeError, ValueError, AttributeError):
|
|
continue
|
|
if not ingredient_ids:
|
|
return {}
|
|
return {
|
|
str(ingredient.id): ingredient
|
|
for ingredient in db.execute(
|
|
select(Ingredient).where(Ingredient.id.in_(ingredient_ids))
|
|
).scalars()
|
|
}
|
|
|
|
|
|
def _assert_bundled_recipe(
|
|
recipe: Recipe,
|
|
expected: Mapping[str, Any],
|
|
resolved_ingredients: Mapping[str, Ingredient],
|
|
) -> None:
|
|
_assert_bundled_recipe_ownership(recipe)
|
|
if not _bundled_recipe_content_matches(recipe, expected, resolved_ingredients):
|
|
raise StarterRecipeCatalogError("Existing bundled recipe content has drifted")
|
|
|
|
|
|
def _assert_bundled_recipe_ownership(recipe: Recipe) -> None:
|
|
if (
|
|
recipe.family_profile_id is not None
|
|
or not recipe.is_manually_added
|
|
or recipe.external_source is not None
|
|
or recipe.external_id is not None
|
|
or recipe.source_url is not None
|
|
or recipe.scraped_at is not None
|
|
or recipe.image_url is not None
|
|
or recipe.image_source is not None
|
|
or recipe.servings_scaled is not None
|
|
or recipe.side_dishes != []
|
|
or recipe.discovery_reason != STARTER_DISCOVERY_REASON
|
|
or recipe.catalog_status != RecipeCatalogStatus.ACTIVE
|
|
or recipe.catalog_findings != []
|
|
or recipe.catalog_reviewed_at is not None
|
|
):
|
|
raise StarterRecipeCatalogError(
|
|
"Existing bundled recipe does not match the required catalog ownership"
|
|
)
|
|
|
|
|
|
def _bundled_recipe_content_matches(
|
|
recipe: Recipe,
|
|
expected: Mapping[str, Any],
|
|
resolved_ingredients: Mapping[str, Ingredient],
|
|
) -> bool:
|
|
if any(getattr(recipe, field) != expected[field] for field in _CATALOG_SCALAR_FIELDS):
|
|
return False
|
|
if list(recipe.instructions or []) != expected["instructions"]:
|
|
return False
|
|
if not isinstance(recipe.ingredients, list) or len(recipe.ingredients) != len(
|
|
expected["ingredients"]
|
|
):
|
|
return False
|
|
for actual, expected_ingredient in zip(recipe.ingredients, expected["ingredients"]):
|
|
if not isinstance(actual, Mapping) or set(actual) != {
|
|
"ingredient_id",
|
|
"qty",
|
|
"unit",
|
|
"notes",
|
|
}:
|
|
return False
|
|
resolved = resolved_ingredients.get(str(actual["ingredient_id"]))
|
|
quantity = actual["qty"]
|
|
if (
|
|
resolved is None
|
|
or resolved.name_lower != _canonical_name(expected_ingredient["name"])
|
|
or isinstance(quantity, bool)
|
|
or not isinstance(quantity, (int, float))
|
|
or not math.isfinite(quantity)
|
|
or quantity != expected_ingredient["quantity"]
|
|
or actual["unit"] != expected_ingredient["unit"]
|
|
or actual["notes"] is not None
|
|
):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _synchronize_bundled_recipe(
|
|
recipe: Recipe,
|
|
expected: Mapping[str, Any],
|
|
canonical_ingredients: Mapping[str, Ingredient],
|
|
) -> None:
|
|
for field in _CATALOG_SCALAR_FIELDS:
|
|
value = expected[field]
|
|
setattr(recipe, field, list(value) if field.endswith("_tags") else value)
|
|
recipe.ingredients = _catalog_recipe_ingredients(expected, canonical_ingredients)
|
|
recipe.instructions = list(expected["instructions"])
|
|
|
|
|
|
def _catalog_recipe_ingredients(
|
|
recipe: Mapping[str, Any], canonical_ingredients: Mapping[str, Ingredient]
|
|
) -> list[dict[str, Any]]:
|
|
return [
|
|
{
|
|
"ingredient_id": str(
|
|
canonical_ingredients[_canonical_name(item["name"])].id
|
|
),
|
|
"qty": item["quantity"],
|
|
"unit": item["unit"],
|
|
"notes": None,
|
|
}
|
|
for item in recipe["ingredients"]
|
|
]
|
|
|
|
|
|
def _reject_json_constant(value: str) -> None:
|
|
raise ValueError(f"Unsupported JSON numeric constant: {value}")
|