Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/migrations/tandoor.py: 17%

1import json 1 ctx1a

2import os 1 ctx1a

3import tempfile 1 ctx1a

4import zipfile 1 ctx1a

5from pathlib import Path 1 ctx1a

6from typing import Any 1 ctx1a

7 

8from mealie.schema.recipe.recipe_ingredient import RecipeIngredientBase 1 ctx1a

9from mealie.schema.reports.reports import ReportEntryCreate 1 ctx1a

10 

11from ._migration_base import BaseMigrator 1 ctx1a

12from .utils.migration_alias import MigrationAlias 1 ctx1a

13from .utils.migration_helpers import format_time 1 ctx1a

14 

15 

16def _build_ingredient_from_ingredient_data(ingredient_data: dict[str, Any], title: str | None = None) -> dict[str, Any]: 1 ctx1a

17 quantity = ingredient_data.get("amount", "1") 

18 if unit_data := ingredient_data.get("unit"): 

19 unit = unit_data.get("plural_name") or unit_data.get("name") 

20 else: 

21 unit = None 

22 

23 if food_data := ingredient_data.get("food"): 

24 food = food_data.get("plural_name") or food_data.get("name") 

25 else: 

26 food = None 

27 

28 base_ingredient = RecipeIngredientBase(quantity=quantity, unit=unit, food=food) 

29 return {"title": title, "note": base_ingredient.display} 

30 

31 

32def extract_instructions_and_ingredients(steps: list[dict[str, Any]]) -> tuple[list[str], list[dict[str, Any]]]: 1 ctx1a

33 """Returns a list of instructions and ingredients for a recipe""" 

34 

35 instructions: list[str] = [] 

36 ingredients: list[dict[str, Any]] = [] 

37 for step in steps: 

38 if instruction_text := step.get("instruction"): 

39 instructions.append(instruction_text) 

40 if ingredients_data := step.get("ingredients"): 

41 for i, ingredient in enumerate(ingredients_data): 

42 if not i and (title := step.get("name")): 

43 ingredients.append(_build_ingredient_from_ingredient_data(ingredient, title)) 

44 else: 

45 ingredients.append(_build_ingredient_from_ingredient_data(ingredient)) 

46 

47 return instructions, ingredients 

48 

49 

50def parse_times(working_time: int, waiting_time: int) -> tuple[str, str]: 1 ctx1a

51 """Returns the performTime and totalTime""" 

52 

53 total_time = working_time + waiting_time 

54 return format_time(working_time), format_time(total_time) 

55 

56 

57class TandoorMigrator(BaseMigrator): 1 ctx1a

58 def __init__(self, **kwargs): 1 ctx1a

59 super().__init__(**kwargs) 

60 

61 self.name = "tandoor" 

62 

63 self.key_aliases = [ 

64 MigrationAlias(key="tags", alias="keywords", func=lambda kws: [kw["name"] for kw in kws if kw.get("name")]), 

65 MigrationAlias(key="orgURL", alias="source_url", func=None), 

66 ] 

67 

68 def _process_recipe_document(self, source_dir: Path, recipe_data: dict) -> dict: 1 ctx1a

69 steps_data = recipe_data.pop("steps", []) 

70 recipe_data["recipeInstructions"], recipe_data["recipeIngredient"] = extract_instructions_and_ingredients( 

71 steps_data 

72 ) 

73 recipe_data["performTime"], recipe_data["totalTime"] = parse_times( 

74 recipe_data.pop("working_time", 0), recipe_data.pop("waiting_time", 0) 

75 ) 

76 

77 recipe_data["recipeYieldQuantity"] = recipe_data.pop("servings", 0) 

78 recipe_data["recipeYield"] = recipe_data.pop("servings_text", "") 

79 

80 try: 

81 recipe_image_path = next(source_dir.glob("image.*")) 

82 recipe_data["image"] = str(recipe_image_path) 

83 except StopIteration: 

84 pass 

85 return recipe_data 

86 

87 def _migrate(self) -> None: 1 ctx1a

88 with tempfile.TemporaryDirectory() as tmpdir: 

89 with zipfile.ZipFile(self.archive) as zip_file: 

90 zip_file.extractall(tmpdir) 

91 

92 source_dir = self.get_zip_base_path(Path(tmpdir)) 

93 

94 recipes_as_dicts: list[dict] = [] 

95 for i, recipe_zip_file in enumerate(source_dir.glob("*.zip")): 

96 try: 

97 recipe_dir = str(source_dir.joinpath(f"recipe_{i + 1}")) 

98 os.makedirs(recipe_dir) 

99 

100 with zipfile.ZipFile(recipe_zip_file) as recipe_zip: 

101 recipe_zip.extractall(recipe_dir) 

102 

103 recipe_source_dir = Path(recipe_dir) 

104 try: 

105 recipe_json_path = next(recipe_source_dir.glob("*.json")) 

106 except StopIteration as e: 

107 raise Exception("recipe.json not found") from e 

108 

109 with open(recipe_json_path) as f: 

110 recipes_as_dicts.append(self._process_recipe_document(recipe_source_dir, json.load(f))) 

111 

112 except Exception as e: 

113 self.report_entries.append( 

114 ReportEntryCreate( 

115 report_id=self.report_id, 

116 success=False, 

117 message="Failed to parse recipe", 

118 exception=f"{type(e).__name__}: {e}", 

119 ) 

120 ) 

121 

122 recipes = [self.clean_recipe_dictionary(x) for x in recipes_as_dicts] 

123 results = self.import_recipes_to_database(recipes) 

124 recipe_lookup = {r.slug: r for r in recipes} 

125 for slug, recipe_id, status in results: 

126 if status: 

127 try: 

128 r = recipe_lookup.get(slug) 

129 if not r or not r.image: 

130 continue 

131 

132 except StopIteration: 

133 continue 

134 

135 self.import_image(slug, r.image, recipe_id)