migration rewrite

mealie/routes/migration_routes.py

@@ -8,6 +8,7 @@ from mealie.db.db_setup import generate_session
 from mealie.routes.deps import get_current_user
 from mealie.schema.migration import MigrationFile, Migrations
 from mealie.schema.snackbar import SnackResponse
+from mealie.services.migrations import migration
 from mealie.services.migrations.chowdown import chowdown_migrate as chowdow_migrate
 from mealie.services.migrations.nextcloud import migrate as nextcloud_migrate
 from sqlalchemy.orm.session import Session
@@ -16,7 +17,7 @@ router = APIRouter(prefix="/api/migrations", tags=["Migration"], dependencies=[D
 
 
 @router.get("", response_model=List[Migrations])
-def get_avaiable_nextcloud_imports():
+def get_all_migration_options():
     """ Returns a list of avaiable directories that can be imported into Mealie """
     response_data = []
     migration_dirs = [
@@ -36,23 +37,28 @@ def get_avaiable_nextcloud_imports():
     return response_data
 
 
-@router.post("/{type}/{file_name}/import")
-def import_nextcloud_directory(type: str, file_name: str, session: Session = Depends(generate_session)):
+@router.post("/{import_type}/{file_name}/import")
+def import_nextcloud_directory(
+    import_type: migration.Migration, file_name: str, session: Session = Depends(generate_session)
+):
     """ Imports all the recipes in a given directory """
-    file_path = app_dirs.MIGRATION_DIR.joinpath(type, file_name)
-    if type == "nextcloud":
-        return nextcloud_migrate(session, file_path)
-    elif type == "chowdown":
-        return chowdow_migrate(session, file_path)
-    else:
-        return SnackResponse.error("Incorrect Migration Type Selected")
+    file_path = app_dirs.MIGRATION_DIR.joinpath(import_type.value, file_name)
+    migration.migrate(import_type, file_path, session)
+
+    # file_path = app_dirs.MIGRATION_DIR.joinpath(type, file_name)
+    # if type == "nextcloud":
+    #     return nextcloud_migrate(session, file_path)
+    # elif type == "chowdown":
+    #     return chowdow_migrate(session, file_path)
+    # else:
+    #     return SnackResponse.error("Incorrect Migration Type Selected")
 
 
-@router.delete("/{type}/{file_name}/delete")
-def delete_migration_data(type: str, file_name: str):
+@router.delete("/{import_type}/{file_name}/delete")
+def delete_migration_data(import_type: migration.Migration, file_name: str):
     """ Removes migration data from the file system """
 
-    remove_path = app_dirs.MIGRATION_DIR.joinpath(type, file_name)
+    remove_path = app_dirs.MIGRATION_DIR.joinpath(import_type.value, file_name)
 
     if remove_path.is_file():
         remove_path.unlink()
@@ -64,10 +70,10 @@ def delete_migration_data(type: str, file_name: str):
     return SnackResponse.error(f"Migration Data Remove: {remove_path.absolute()}")
 
 
-@router.post("/{type}/upload")
-def upload_nextcloud_zipfile(type: str, archive: UploadFile = File(...)):
+@router.post("/{import_type}/upload")
+def upload_nextcloud_zipfile(import_type: migration.Migration, archive: UploadFile = File(...)):
     """ Upload a .zip File to later be imported into Mealie """
-    dir = app_dirs.MIGRATION_DIR.joinpath(type)
+    dir = app_dirs.MIGRATION_DIR.joinpath(import_type.value)
     dir.mkdir(parents=True, exist_ok=True)
     dest = dir.joinpath(archive.filename)
 

mealie/schema/migration.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from typing import List
 
+from mealie.schema.restore import RecipeImport
 from pydantic.main import BaseModel
 
 
@@ -23,3 +24,7 @@ class MigrationFile(BaseModel):
 class Migrations(BaseModel):
     type: str
     files: List[MigrationFile] = []
+
+
+class MigrationImport(RecipeImport):
+    pass

mealie/services/image/image.py

@@ -65,8 +65,7 @@ def write_image(recipe_slug: str, file_data: bytes, extension: str) -> Path.name
         with open(image_path, "ab") as f:
             f.write(file_data)
     else:
-        with open(image_path, "ab") as f:
-            shutil.copyfileobj(file_data, f)
+        shutil.copy2(file_data, image_path)
 
     minify.migrate_images()
 

mealie/services/migrations/_migration_base.py

@@ -0,0 +1,186 @@
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import Any, Callable, Optional
+
+import yaml
+from fastapi.logger import logger
+from mealie.db.database import db
+from mealie.schema.migration import MigrationImport
+from mealie.schema.recipe import Recipe
+from mealie.services.image import image, minify
+from mealie.services.scraper.cleaner import Cleaner
+from mealie.utils.unzip import unpack_zip
+from pydantic import BaseModel
+
+try:
+    from yaml import CLoader as Loader
+except ImportError:
+    from yaml import Loader
+
+
+class MigrationAlias(BaseModel):
+    """A datatype used by MigrationBase to pre-process a recipe dictionary to rewrite
+    the alias key in the dictionary, if it exists, to the key. If set a `func` attribute
+    will be called on the value before assigning the value to the new key
+    """
+
+    key: str
+    alias: str
+    func: Optional[Callable] = None
+
+
+class MigrationBase(BaseModel):
+    migration_report: list[MigrationImport] = []
+    migration_file: Path
+    session: Optional[Any]
+    key_aliases: Optional[list[MigrationAlias]]
+
+    @property
+    def temp_dir(self) -> TemporaryDirectory:
+        """unpacks the migration_file into a temporary directory
+        that can be used as a context manager.
+
+        Returns:
+            TemporaryDirectory:
+        """
+        return unpack_zip(self.migration_file)
+
+    @staticmethod
+    def json_reader(json_file: Path) -> dict:
+        print(json_file)
+        with open(json_file, "r") as f:
+            return json.loads(f.read())
+
+    @staticmethod
+    def yaml_reader(yaml_file: Path) -> dict:
+        """A helper function to read in a yaml file from a Path. This assumes that the
+        first yaml document is the recipe data and the second, if exists, is the description.
+
+        Args:
+            yaml_file (Path): Path to yaml file
+
+        Returns:
+            dict: Contains keys "recipe_data" and optional "description"
+        """
+        with open(yaml_file, "r") as stream:
+            try:
+                for x, item in enumerate(yaml.load_all(stream, Loader=Loader)):
+                    if x == 0:
+                        recipe_data = item
+                    elif x == 1:
+                        recipe_description = str(item)
+
+            except yaml.YAMLError:
+                return
+
+        return {"recipe_data": recipe_data, "description": recipe_description or None}
+
+    @staticmethod
+    def glob_walker(directory: Path, glob_str: str, return_parent=True) -> list[Path]:  # TODO:
+        """A Helper function that will return the glob matches for the temporary directotry
+        that was unpacked and passed in as the `directory` parameter. If `return_parent` is
+        True the return Paths will be the parent directory for the file that was matched. If
+        false the file itself will be returned.
+
+        Args:
+            directory (Path): Path to search directory
+            glob_str ([type]): glob style match string
+            return_parent (bool, optional): To return parent directory of match. Defaults to True.
+
+        Returns:
+            list[Path]:
+        """
+        directory = directory if isinstance(directory, Path) else Path(directory)
+        matches = []
+        for match in directory.glob(glob_str):
+            if return_parent:
+                matches.append(match.parent)
+            else:
+                matches.append(match)
+
+        return matches
+
+    @staticmethod
+    def import_image(src: Path, dest_slug: str):
+        """Read the successful migrations attribute and for each import the image
+        appropriately into the image directory. Minification is done in mass
+        after the migration occurs.
+        """
+        image.write_image(dest_slug, src, extension=src.suffix)
+        minify.migrate_images()  # TODO: Refactor to support single file minification that doesn't suck
+
+    def get_recipe_from_file(self, file) -> Recipe:
+        """This is the method called to read a file path and transform that data
+        into a recipe object. The expected return value is a Recipe object that is then
+        passed to
+
+        Args:
+            file ([type]): [description]
+
+        Raises:
+            NotImplementedError: [description]
+        """
+
+        raise NotImplementedError("Migration Type not Implemented")
+
+    def rewrite_alias(self, recipe_dict: dict) -> dict:
+        """A helper function to reassign attributes by an alias using a list
+        of MigrationAlias objects to rewrite the alias attribute found in the recipe_dict
+        to a
+
+        Args:
+            recipe_dict (dict): [description]
+            key_aliases (list[MigrationAlias]): [description]
+
+        Returns:
+            dict: [description]
+        """
+        if not self.key_aliases:
+            return recipe_dict
+
+        for alias in self.key_aliases:
+            try:
+                prop_value = recipe_dict.pop(alias.alias)
+            except KeyError:
+                logger.info(f"Key {alias.alias} Not Found. Skipping...")
+                continue
+
+            if alias.func:
+                prop_value = alias.func(prop_value)
+
+            recipe_dict[alias.key] = prop_value
+
+        return recipe_dict
+
+    def clean_recipe_dictionary(self, recipe_dict) -> Recipe:
+        """Calls the rewrite_alias function and the Cleaner.clean function on a
+        dictionary and returns the result unpacked into a Recipe object"""
+        recipe_dict = self.rewrite_alias(recipe_dict)
+        recipe_dict = Cleaner.clean(recipe_dict)
+
+        return Recipe(**recipe_dict)
+
+    def import_recipes_to_database(self, validated_recipes: list[Recipe]) -> None:
+        """Used as a single access point to process a list of Recipe objects into the
+        database in a predictable way. If an error occurs the session is rolled back
+        and the process will continue. All import information is appended to the
+        'migration_report' attribute to be returned to the frontend for display.
+
+        Args:
+            validated_recipes (list[Recipe]):
+        """
+
+        for recipe in validated_recipes:
+            exception = ""
+            status = False
+            try:
+                db.recipes.create(self.session, recipe.dict())
+                status = True
+
+            except Exception as inst:
+                exception = inst
+                self.session.rollback()
+
+            import_status = MigrationImport(slug=recipe.slug, name=recipe.name, status=status, exception=str(exception))
+            self.migration_report.append(import_status)

mealie/services/migrations/chowdown.py

@@ -18,12 +18,12 @@ except ImportError:
 
 def read_chowdown_file(recipe_file: Path) -> Recipe:
     """Parse through the yaml file to try and pull out the relavent information.
-    Some issues occur when ":" are used in the text. I have no put a lot of effort
+    Some issues occur when ":" are used in the text. I have not put a lot of effort
     into this so there may be better ways of going about it. Currently, I get about 80-90%
     of recipes from repos I've tried.
 
     Args:
-        recipe_file (Path): Path to the .yml file
+        recipe_file (Path): Path to the yaml file
 
     Returns:
         Recipe: Recipe class object
@@ -36,7 +36,6 @@ def read_chowdown_file(recipe_file: Path) -> Recipe:
             for x, item in enumerate(yaml.load_all(stream, Loader=Loader)):
                 if x == 0:
                     recipe_data = item
-
                 elif x == 1:
                     recipe_description = str(item)
 

mealie/services/migrations/migration.py

@@ -0,0 +1,44 @@
+from enum import Enum
+from pathlib import Path
+
+from fastapi.logger import logger
+from mealie.schema.migration import MigrationImport
+from mealie.services.migrations import chowdown, nextcloud, nextcloud_new
+from sqlalchemy.orm.session import Session
+
+
+class Migration(str, Enum):
+    """The class defining the supported types of migrations for Mealie. Pass the
+    class attribute of the class instead of the string when using.
+    """
+
+    nextcloud = "nextcloud"
+    chowdown = "chowdown"
+
+
+def migrate(migration_type: str, file_path: Path, session: Session) -> list[MigrationImport]:
+    """The new entry point for accessing migrations within the 'migrations' service.
+    Using the 'Migrations' enum class as a selector for migration_type to direct which function
+    to call. All migrations will return a MigrationImport object that is built for displaying
+    detailed information on the frontend. This will provide a single point of access
+
+    Args:
+        migration_type (str): a string option representing the migration type. See Migration attributes for options
+        file_path (Path): Path to the zip file containing the data
+        session (Session): a SqlAlchemy Session
+
+    Returns:
+        list[MigrationImport]: [description]
+    """
+
+    logger.info(f"Starting Migration from {migration_type}")
+
+    if migration_type == Migration.nextcloud.value:
+        migration_imports = nextcloud_new.migrate(session, file_path)
+
+    elif migration_type == Migration.chowdown.value:
+        migration_imports = chowdown.chowdown_migrate(session, file_path)
+
+    logger.info(f"Finishing Migration from {migration_type}")
+
+    return None

mealie/services/migrations/nextcloud_new.py

@@ -0,0 +1,73 @@
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+from mealie.schema.migration import MigrationImport
+from mealie.services.migrations._migration_base import MigrationAlias, MigrationBase
+from slugify import slugify
+from sqlalchemy.orm.session import Session
+
+
+def clean_nextcloud_tags(nextcloud_tags: str):
+    if not isinstance(nextcloud_tags, str):
+        return None
+
+    return [x.title().lstrip() for x in nextcloud_tags.split(",") if x != ""]
+
+
+@dataclass
+class NextcloudDir:
+    name: str
+    recipe: dict
+    image: Optional[Path]
+
+    @property
+    def slug(self):
+        return slugify(self.recipe["name"])
+
+    @classmethod
+    def from_dir(cls, dir: Path):
+        try:
+            json_file = next(dir.glob("*.json"))
+        except StopIteration:
+            return None
+
+        try:  # TODO: There's got to be a better way to do this.
+            image_file = next(dir.glob("full.*"))
+        except StopIteration:
+            image_file = None
+
+        return cls(name=dir.name, recipe=NextcloudMigration.json_reader(json_file), image=image_file)
+
+
+class NextcloudMigration(MigrationBase):
+    key_aliases: Optional[list[MigrationAlias]] = [
+        MigrationAlias(key="tags", alias="keywords", func=clean_nextcloud_tags)
+    ]
+
+
+def migrate(session: Session, zip_path: Path) -> list[MigrationImport]:
+
+    nc_migration = NextcloudMigration(migration_file=zip_path, session=session)
+
+    with nc_migration.temp_dir as dir:
+        potential_recipe_dirs = NextcloudMigration.glob_walker(dir, glob_str="**/[!.]*.json", return_parent=True)
+
+        nextcloud_dirs = [NextcloudDir.from_dir(x) for x in potential_recipe_dirs]
+        nextcloud_dirs = {x.slug: x for x in nextcloud_dirs}
+
+        all_recipes = []
+        for key, nc_dir in nextcloud_dirs.items():
+            recipe = nc_migration.clean_recipe_dictionary(nc_dir.recipe)
+            print("Key", key)
+            all_recipes.append(recipe)
+
+        nc_migration.import_recipes_to_database(all_recipes)
+
+        for report in nc_migration.migration_report:
+
+            if report.status:
+                print(report)
+                nc_dir: NextcloudDir = nextcloud_dirs[report.slug]
+                if nc_dir.image:
+                    NextcloudMigration.import_image(nc_dir.image, nc_dir.slug)