254 lines
12 KiB
Python
Executable File
254 lines
12 KiB
Python
Executable File
#!/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())
|