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

1import base64 

2import io 

3import json 

4import re 

5import tempfile 

6import zipfile 

7from gzip import GzipFile 

8from pathlib import Path 

9 

10from mealie.schema.recipe import RecipeNote 

11 

12from ._migration_base import BaseMigrator 

13from .utils.migration_alias import MigrationAlias 

14 

15 

16def paprika_recipes(file: Path): 

17 """Yields all recipes inside the export file as JSON""" 

18 with tempfile.TemporaryDirectory() as tmpdir: 

19 with zipfile.ZipFile(file) as zip_file: 

20 zip_file.extractall(tmpdir) 

21 

22 for name in Path(tmpdir).glob("**/[!.]*.paprikarecipe"): 

23 with open(name, "rb") as fd: 

24 with GzipFile("r", fileobj=fd) as recipe_json: 

25 recipe = json.load(recipe_json) 

26 yield recipe 

27 

28 

29class PaprikaMigrator(BaseMigrator): 

30 def __init__(self, **kwargs): 

31 super().__init__(**kwargs) 

32 

33 self.name = "paprika" 

34 

35 re_num_list = re.compile(r"^\d+\.\s") 

36 

37 self.key_aliases = [ 

38 MigrationAlias(key="recipeIngredient", alias="ingredients", func=lambda x: x.split("\n") if x else ""), 

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

40 MigrationAlias(key="totalTime", alias="total_time", func=None), 

41 MigrationAlias(key="prepTime", alias="prep_time", func=None), 

42 MigrationAlias(key="performTime", alias="cook_time", func=None), 

43 MigrationAlias(key="recipeYield", alias="servings", func=None), 

44 MigrationAlias( 

45 key="tags", alias="categories", func=None 

46 ), # Paprika doesn't support tags, and instead puts tags in categories 

47 MigrationAlias(key="image", alias="image_url", func=None), 

48 MigrationAlias(key="dateAdded", alias="created", func=lambda x: x[: x.find(" ")]), 

49 MigrationAlias( 

50 key="notes", 

51 alias="notes", 

52 func=lambda x: [z for z in [RecipeNote(title="", text=x) if x else None] if z], 

53 ), 

54 MigrationAlias( 

55 key="recipeCategory", 

56 alias="categories", 

57 func=self.helpers.get_or_set_category, 

58 ), 

59 MigrationAlias( 

60 key="recipeInstructions", 

61 alias="directions", 

62 func=lambda x: [{"text": re.sub(re_num_list, "", s)} for s in x.split("\n\n")] if x else [], 

63 ), 

64 ] 

65 

66 def _migrate(self) -> None: 

67 recipes = [r for r in paprika_recipes(self.archive) if "name" in r] 

68 recipe_models = [self.clean_recipe_dictionary(r) for r in recipes] 

69 results = self.import_recipes_to_database(recipe_models) 

70 

71 for (slug, recipe_id, status), recipe in zip(results, recipes, strict=True): 

72 if not status: 

73 continue 

74 

75 image_data = recipe.get("photo_data") 

76 if image_data is None: 

77 self.logger.info(f"Recipe '{recipe['name']}' has no image") 

78 continue 

79 

80 try: 

81 # Images are stored as base64 encoded strings, so we need to decode them before importing. 

82 image = io.BytesIO(base64.b64decode(image_data)) 

83 with tempfile.NamedTemporaryFile(suffix=".jpeg") as temp_file: 

84 temp_file.write(image.read()) 

85 temp_file.flush() 

86 path = Path(temp_file.name) 

87 self.import_image(slug, path, recipe_id) 

88 except Exception as e: 

89 self.logger.error(f"Failed to import image for {slug}: {e}")