Merge pull request dandi#61 from candleindark/enh

Add `migrate-manifests-dandisets` subcommand

src/dandisets_linkml_status_tools/cli/__init__.py

@@ -362,3 +362,21 @@ def diff_manifests_reports_(
     diff_manifests_reports(
         reports_dir1_path, reports_dir2_path, config["output_dir_path"]
     )
+
+
+@app.command("migrate-manifests-dandisets")
+def migrate_manifests_dandisets_(
+    *,
+    manifest_path: Annotated[
+        Path, typer.Argument(help="Path of the directory containing dandiset manifests")
+    ],
+):
+    """
+    Migrate `Dandiset` metadata in manifests to the latest version of the `Dandiset`
+    model
+    """
+    from dandisets_linkml_status_tools.cmd_funcs.migrate_manifests_dandisets import (
+        migrate_manifests_dandisets,
+    )
+
+    migrate_manifests_dandisets(manifest_path, config["output_dir_path"])

src/dandisets_linkml_status_tools/cmd_funcs/migrate_manifests_dandisets.py

@@ -0,0 +1,77 @@
+import json
+import logging
+from pathlib import Path
+
+from dandischema.metadata import migrate
+
+from dandisets_linkml_status_tools.tools import (
+    create_or_replace_dir,
+    get_direct_subdirs,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def migrate_manifests_dandisets(manifest_path: Path, output_dir: Path) -> None:
+    """
+    Migrate `Dandiset` metadata in manifests to the latest version of the `Dandiset`
+    model
+
+    :param manifest_path: Path of the directory containing dandiset manifests
+    :param output_dir: Path of the directory to save the migrated manifests
+    """
+    from dandisets_linkml_status_tools.cli import DANDISET_FILE_NAME
+
+    logger.info("Creating directory %s for the migrated manifests", output_dir)
+    create_or_replace_dir(output_dir)
+
+    for dandiset_dir in get_direct_subdirs(manifest_path):
+        # === In a dandiset directory ===
+        dandiset_identifier = dandiset_dir.name
+
+        for version_dir in get_direct_subdirs(dandiset_dir):
+            # === In a dandiset version directory ===
+            dandiset_version = version_dir.name
+
+            # Dandiset metadata file path
+            dandiset_md_file: Path = version_dir / DANDISET_FILE_NAME
+
+            # Skip if the dandiset metadata file does not exist
+            if not dandiset_md_file.is_file():
+                continue
+
+            # Load the dandiset metadata
+            dandiset_md = json.loads(dandiset_md_file.read_text())
+
+            # Migrate the dandiset metadata
+            try:
+                dandiset_md_migrated = migrate(dandiset_md, skip_validation=True)
+            except (NotImplemented, ValueError) as e:
+                logger.warning(
+                    "Failed to migrate dandiset metadata in %s/%s: %s",
+                    dandiset_identifier,
+                    dandiset_version,
+                    e,
+                )
+                # Construct a dummy dandiset metadata instance indicating the error
+                dandiset_md_migrated = {
+                    "metadata_migration_failed": f"This is an invalid dandiset metadata"
+                    f" instance. Migration of the original metadata instance failed "
+                    f"with error: {e!r}"
+                }
+            else:
+                logger.info(
+                    "Migrated dandiset metadata in %s/%s",
+                    dandiset_identifier,
+                    dandiset_version,
+                )
+
+            # Save the migrated dandiset metadata
+            dandiset_md_migrated_dir: Path = (
+                output_dir / dandiset_identifier / dandiset_version
+            )
+            dandiset_md_migrated_dir.mkdir(parents=True)
+            dandiset_md_migrated_file: Path = (
+                dandiset_md_migrated_dir / DANDISET_FILE_NAME
+            )
+            dandiset_md_migrated_file.write_text(json.dumps(dandiset_md_migrated))

src/dandisets_linkml_status_tools/tools/jsonschema.py

@@ -8,7 +8,7 @@
 from dandisets_linkml_status_tools.models import JsonschemaValidationErrorModel
 
 
-# todo: this function is available through the dandischema package if
+# TODO: this function is available through the dandischema package if
 #  https://github.com/dandi/dandi-schema/pull/278 is accepted
 def jsonschema_validator(
     schema: dict[str, Any],