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

62 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:32 +0000

1import tempfile 1a

2import zipfile 1a

3from dataclasses import dataclass 1a

4from pathlib import Path 1a

5 

6from slugify import slugify 1a

7 

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

9 

10from ._migration_base import BaseMigrator 1a

11from .utils.migration_alias import MigrationAlias 1a

12from .utils.migration_helpers import MigrationReaders, glob_walker, parse_iso8601_duration, split_by_comma 1a

13 

14 

15@dataclass 1a

16class NextcloudDir: 1a

17 name: str 1a

18 recipe: dict 1a

19 image: Path | None = None 1a

20 

21 @property 1a

22 def slug(self): 1a

23 return slugify(self.recipe.get("name")) 

24 

25 @classmethod 1a

26 def from_dir(cls, dir: Path): 1a

27 try: 

28 json_file = next(dir.glob("*.json")) 

29 except StopIteration: 

30 return None 

31 

32 try: # TODO: There's got to be a better way to do this. 

33 image_file = next(dir.glob("full.*")) 

34 except StopIteration: 

35 image_file = None 

36 

37 return cls(name=dir.name, recipe=MigrationReaders.json(json_file), image=image_file) 

38 

39 

40class NextcloudMigrator(BaseMigrator): 1a

41 def __init__(self, **kwargs): 1a

42 super().__init__(**kwargs) 

43 

44 self.name = "nextcloud" 

45 

46 self.key_aliases = [ 

47 MigrationAlias(key="tags", alias="keywords", func=split_by_comma), 

48 MigrationAlias(key="orgURL", alias="url", func=None), 

49 MigrationAlias(key="totalTime", alias="totalTime", func=parse_iso8601_duration), 

50 MigrationAlias(key="prepTime", alias="prepTime", func=parse_iso8601_duration), 

51 MigrationAlias(key="performTime", alias="cookTime", func=parse_iso8601_duration), 

52 ] 

53 

54 @classmethod 1a

55 def get_zip_base_path(cls, path: Path) -> Path: 1a

56 potential_path = super().get_zip_base_path(path) 

57 if path == potential_path: 

58 return path 

59 

60 # make sure we didn't accidentally open a recipe dir 

61 if (potential_path / "recipe.json").exists(): 

62 return path 

63 else: 

64 return potential_path 

65 

66 def _migrate(self) -> None: 1a

67 # Unzip File into temp directory 

68 

69 # get potential recipe dirs 

70 with tempfile.TemporaryDirectory() as tmpdir: 

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

72 zip_file.extractall(tmpdir) 

73 

74 base_dir = self.get_zip_base_path(Path(tmpdir)) 

75 potential_recipe_dirs = glob_walker(base_dir, glob_str="**/[!.]*.json", return_parent=True) 

76 nextcloud_dirs = {y.slug: y for x in potential_recipe_dirs if (y := NextcloudDir.from_dir(x))} 

77 

78 all_recipes = [] 

79 for _, nc_dir in nextcloud_dirs.items(): 

80 try: 

81 recipe = self.clean_recipe_dictionary(nc_dir.recipe) 

82 all_recipes.append(recipe) 

83 except Exception as e: 

84 self.logger.exception(e) 

85 self.report_entries.append( 

86 ReportEntryCreate( 

87 report_id=self.report_id, 

88 success=False, 

89 message=f"Failed to import {nc_dir.name}", 

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

91 ) 

92 ) 

93 

94 all_statuses = self.import_recipes_to_database(all_recipes) 

95 

96 for slug, recipe_id, status in all_statuses: 

97 if status: 

98 nc_dir = nextcloud_dirs[slug] 

99 if nc_dir.image: 

100 self.import_image(slug, nc_dir.image, recipe_id)