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

1import contextlib 1 ctx1a

2import shutil 1 ctx1a

3import tempfile 1 ctx1a

4import zipfile 1 ctx1a

5from pathlib import Path 1 ctx1a

6 

7from mealie.schema.recipe.recipe import Recipe 1 ctx1a

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

9 

10from ._migration_base import BaseMigrator 1 ctx1a

11from .utils.migration_alias import MigrationAlias 1 ctx1a

12from .utils.migration_helpers import MigrationReaders, split_by_comma 1 ctx1a

13 

14 

15class MealieAlphaMigrator(BaseMigrator): 1 ctx1a

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

17 super().__init__(**kwargs) 

18 

19 self.name = "mealie_alpha" 

20 

21 self.key_aliases = [ 

22 MigrationAlias(key="name", alias="title", func=None), 

23 MigrationAlias(key="recipeIngredient", alias="ingredients", func=None), 

24 MigrationAlias(key="recipeInstructions", alias="directions", func=None), 

25 MigrationAlias(key="tags", alias="tags", func=split_by_comma), 

26 ] 

27 

28 @classmethod 1 ctx1a

29 def get_zip_base_path(cls, path: Path) -> Path: 1 ctx1a

30 potential_path = super().get_zip_base_path(path) 

31 if path == potential_path: 

32 return path 

33 

34 # make sure we didn't accidentally open the "recipes" dir 

35 if potential_path.name == "recipes": 

36 return path 

37 else: 

38 return potential_path 

39 

40 def _convert_to_new_schema(self, recipe: dict) -> Recipe: 1 ctx1a

41 if recipe.get("categories", False): 

42 recipe["recipeCategory"] = recipe.get("categories") 

43 del recipe["categories"] 

44 

45 with contextlib.suppress(KeyError): 

46 del recipe["_id"] 

47 del recipe["date_added"] 

48 # Migration from list to Object Type Data 

49 with contextlib.suppress(KeyError): 

50 if "" in recipe["tags"]: 

51 recipe["tags"] = [tag for tag in recipe["tags"] if tag != ""] 

52 with contextlib.suppress(KeyError): 

53 if "" in recipe["categories"]: 

54 recipe["categories"] = [cat for cat in recipe["categories"] if cat != ""] 

55 if isinstance(recipe["extras"], list): 

56 recipe["extras"] = {} 

57 

58 recipe["comments"] = [] 

59 

60 # Reset ID on migration 

61 recipe["id"] = None 

62 

63 return Recipe(**recipe) 

64 

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

66 with tempfile.TemporaryDirectory() as tmpdir: 

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

68 zip_file.extractall(tmpdir) 

69 

70 temp_path = self.get_zip_base_path(Path(tmpdir)) 

71 recipe_lookup: dict[str, Path] = {} 

72 

73 recipes: list[Recipe] = [] 

74 for recipe_json_path in temp_path.rglob("**/recipes/**/[!.]*.json"): 

75 try: 

76 if (recipe_as_dict := MigrationReaders.json(recipe_json_path)) is not None: 

77 recipe = self._convert_to_new_schema(recipe_as_dict) 

78 recipes.append(recipe) 

79 slug = recipe_as_dict["slug"] 

80 recipe_lookup[slug] = recipe_json_path.parent 

81 except Exception as e: 

82 self.logger.exception(e) 

83 self.report_entries.append( 

84 ReportEntryCreate( 

85 report_id=self.report_id, 

86 success=False, 

87 message=f"Failed to import {recipe_json_path.name}", 

88 exception=f"{e.__class__.__name__}: {e}", 

89 ) 

90 ) 

91 

92 results = self.import_recipes_to_database(recipes) 

93 for slug, recipe_id, status in results: 

94 if not status: 

95 continue 

96 

97 dest_dir = Recipe.directory_from_id(recipe_id) 

98 source_dir = recipe_lookup.get(slug) 

99 

100 if dest_dir.exists(): 

101 shutil.rmtree(dest_dir) 

102 

103 if source_dir is None: 

104 continue 

105 

106 for dir in source_dir.iterdir(): 

107 if dir.is_dir(): 

108 shutil.copytree(dir, dest_dir / dir.name)