#!/usr/bin/env python3 """Validate canonical culinary entities and their relationships.""" from __future__ import annotations import json import sys from pathlib import Path import yaml from jsonschema import Draft202012Validator, FormatChecker from referencing import Registry, Resource ROOT = Path(__file__).resolve().parent.parent SCHEMA_DIR = ROOT / "container/schemas" CULINARY_DIR = ROOT / "culinary" def load_yaml(path: Path, *, front_matter: bool = False): text = path.read_text(encoding="utf-8") if front_matter: lines = text.splitlines() if not lines or lines[0].strip() != "---": raise ValueError("missing opening YAML front-matter delimiter") try: closing = next(i for i, line in enumerate(lines[1:], 1) if line.strip() == "---") except StopIteration as exc: raise ValueError("missing closing YAML front-matter delimiter") from exc text = "\n".join(lines[1:closing]) data = yaml.safe_load(text) if not isinstance(data, dict): raise ValueError("document must contain a YAML object") return data def format_path(parts) -> str: return "".join(f"[{part}]" if isinstance(part, int) else f".{part}" for part in parts).lstrip(".") def display_path(path: Path) -> Path: try: return path.relative_to(ROOT) except ValueError: return path def main() -> int: errors: list[str] = [] def collect(paths, destination, validator, *, front_matter=False, id_matches_filename=False): for path in paths: relative = display_path(path) try: data = load_yaml(path, front_matter=front_matter) except (OSError, ValueError, yaml.YAMLError) as exc: errors.append(f"{relative}: {exc}") continue for issue in sorted(validator.iter_errors(data), key=lambda e: list(e.absolute_path)): location = format_path(issue.absolute_path) errors.append(f"{relative}{':' + location if location else ''}: {issue.message}") item_id = data.get("id") if not isinstance(item_id, str): continue if item_id in destination: first = display_path(destination[item_id][0]) errors.append(f"{relative}: duplicate id {item_id!r} (first defined in {first})") else: destination[item_id] = (path, data) if id_matches_filename and path.stem != item_id: errors.append(f"{relative}: filename must match id {item_id!r}") v2_specs = { "units": "unit.schema.json", "ingredients": "ingredient.schema.json", "recipes": "recipe.schema.json", "prep_actions": "prep-action.schema.json", "allergens": "allergen.schema.json", "source_mappings": "source-mapping.schema.json", "suppliers": "supplier.schema.json", "purchase_items": "purchase-item.schema.json", "equipment": "equipment.schema.json", "collections": "collection.schema.json", "derived_recipes": "derived-recipe.schema.json", } v2_schema_dir = SCHEMA_DIR / "v2" common_schema = json.loads((v2_schema_dir / "common.schema.json").read_text(encoding="utf-8")) Draft202012Validator.check_schema(common_schema) registry = Registry().with_resource( common_schema["$id"], Resource.from_contents(common_schema) ) v2_entities: dict[str, dict[str, tuple[Path, dict]]] = {} for entity_type, schema_name in v2_specs.items(): schema_path = v2_schema_dir / schema_name schema = json.loads(schema_path.read_text(encoding="utf-8")) Draft202012Validator.check_schema(schema) validator = Draft202012Validator( schema, registry=registry, format_checker=FormatChecker(), ) destination: dict[str, tuple[Path, dict]] = {} entity_dir = CULINARY_DIR / entity_type paths = list(entity_dir.glob("*.yaml")) if entity_dir.exists() else [] if paths: collect( sorted(paths), destination, validator, id_matches_filename=entity_type != "derived_recipes", ) v2_entities[entity_type] = destination unit_ids = set(v2_entities["units"]) unit_dimensions = { unit_id: unit[1].get("dimension") for unit_id, unit in v2_entities["units"].items() } ingredient_ids = set(v2_entities["ingredients"]) recipe_ids = set(v2_entities["recipes"]) action_ids = set(v2_entities["prep_actions"]) equipment_ids = set(v2_entities["equipment"]) mapping_ids = set(v2_entities["source_mappings"]) supplier_ids = set(v2_entities["suppliers"]) def require_ref(path: Path, kind: str, item_id: str, known: set[str]): if item_id not in known: errors.append(f"{display_path(path)}: unknown v2 {kind} {item_id!r}") def check_amount(path: Path, amount): if isinstance(amount, dict) and isinstance(amount.get("unit_id"), str): require_ref(path, "unit_id", amount["unit_id"], unit_ids) def require_dimension(path: Path, amount, expected: str, context: str): if not isinstance(amount, dict): return unit_id = amount.get("unit_id") if unit_id in unit_dimensions and unit_dimensions[unit_id] != expected: errors.append( f"{display_path(path)}: {context} requires a {expected} unit, " f"not {unit_id!r} ({unit_dimensions[unit_id]})" ) for _, (path, unit) in v2_entities["units"].items(): conversion = unit.get("base_conversion") if conversion: base_id = conversion["base_unit_id"] require_ref(path, "base unit_id", base_id, unit_ids) if base_id in unit_dimensions and unit_dimensions[base_id] != unit["dimension"]: errors.append(f"{display_path(path)}: base unit must have the same dimension") for _, (path, ingredient) in v2_entities["ingredients"].items(): for density in ingredient.get("density_measurements", []): check_amount(path, density.get("mass")) check_amount(path, density.get("volume")) require_dimension(path, density.get("mass"), "mass", "density mass") require_dimension(path, density.get("volume"), "volume", "density volume") for conversion in ingredient.get("measure_conversions", []): check_amount(path, conversion.get("from")) check_amount(path, conversion.get("to")) for mapping_id in ingredient.get("nutrition_mapping_ids", []) + ingredient.get("allergen_mapping_ids", []): require_ref(path, "source mapping_id", mapping_id, mapping_ids) for prep in ingredient.get("prep_actions", []): require_ref(path, "prep action_id", prep["action_id"], action_ids) for mapping_id, (path, mapping) in v2_entities["source_mappings"].items(): subject = mapping.get("subject", {}) subject_type = subject.get("type") subject_id = subject.get("id") known_subjects = { "ingredient": ingredient_ids, "purchase_item": set(v2_entities["purchase_items"]), "recipe": recipe_ids, } if subject_type in known_subjects: require_ref(path, f"{subject_type} subject", subject_id, known_subjects[subject_type]) if mapping.get("status") == "reviewed" and not mapping.get("source", {}).get("reviewed"): errors.append(f"{display_path(path)}: reviewed mapping must have source.reviewed=true") if subject_type == "ingredient" and subject_id in v2_entities["ingredients"]: ingredient = v2_entities["ingredients"][subject_id][1] field = "nutrition_mapping_ids" if mapping.get("mapping_type") == "nutrition" else "allergen_mapping_ids" if mapping_id not in ingredient.get(field, []): errors.append(f"{display_path(path)}: mapping is not referenced by ingredient {subject_id!r}") for _, (path, recipe) in v2_entities["recipes"].items(): check_amount(path, recipe.get("yield", {}).get("amount")) check_amount(path, recipe.get("yield", {}).get("serving_size")) check_amount(path, recipe.get("scaling", {}).get("basis_amount")) component_ids = {component.get("id") for component in recipe.get("components", [])} if len(component_ids) != len(recipe.get("components", [])): errors.append(f"{display_path(path)}: component IDs must be unique") item_ids: set[str] = set() for component in recipe.get("components", []): check_amount(path, component.get("yield")) for item in component.get("items", []): item_id = item.get("id") if item_id in item_ids: errors.append(f"{display_path(path)}: component item ID {item_id!r} is duplicated") item_ids.add(item_id) check_amount(path, item.get("amount")) reference = item.get("reference", {}) if "ingredient_id" in reference: require_ref(path, "ingredient_id", reference["ingredient_id"], ingredient_ids) if "recipe_id" in reference: require_ref(path, "recipe_id", reference["recipe_id"], recipe_ids) for prep in item.get("prep", []): require_ref(path, "prep action_id", prep["action_id"], action_ids) orders = [step.get("order") for step in recipe.get("steps", [])] if orders != list(range(1, len(orders) + 1)): errors.append(f"{display_path(path)}: step order must be contiguous starting at 1") step_ids = [step.get("id") for step in recipe.get("steps", [])] if len(step_ids) != len(set(step_ids)): errors.append(f"{display_path(path)}: step IDs must be unique") for equipment_id in recipe.get("equipment_ids", []): require_ref(path, "equipment_id", equipment_id, equipment_ids) for step in recipe.get("steps", []): check_amount(path, step.get("duration")) check_amount(path, step.get("temperature")) require_dimension(path, step.get("duration"), "time", "step duration") require_dimension(path, step.get("temperature"), "temperature", "step temperature") for component_id in step.get("component_ids", []): if component_id not in component_ids: errors.append(f"{display_path(path)}: step references unknown component_id {component_id!r}") for item_id in step.get("item_ids", []): if item_id not in item_ids: errors.append(f"{display_path(path)}: step references unknown item_id {item_id!r}") for equipment_id in step.get("equipment_ids", []): require_ref(path, "equipment_id", equipment_id, equipment_ids) for _, (path, item) in v2_entities["purchase_items"].items(): require_ref(path, "ingredient_id", item["ingredient_id"], ingredient_ids) if "supplier_id" in item: require_ref(path, "supplier_id", item["supplier_id"], supplier_ids) require_ref(path, "package unit_id", item["package"]["unit_id"], unit_ids) for mapping_id in item.get("nutrition_mapping_ids", []) + item.get("allergen_mapping_ids", []): require_ref(path, "source mapping_id", mapping_id, mapping_ids) for _, (path, collection) in v2_entities["collections"].items(): for entry in collection.get("entries", []): require_ref(path, "recipe_id", entry["recipe_id"], recipe_ids) if errors: print(f"Content validation failed with {len(errors)} error(s):", file=sys.stderr) for error in errors: print(f"- {error}", file=sys.stderr) return 1 v2_count = sum(len(records) for records in v2_entities.values()) print(f"Validated {v2_count} canonical culinary entities.") return 0 if __name__ == "__main__": raise SystemExit(main())