add clean folders in backup

superset_tool/utils/fileio.py

@@ -16,6 +16,7 @@ from pathlib import Path
 from typing import Any, Optional, Tuple, Dict, List, Literal, Union, BinaryIO, LiteralString
 from collections import defaultdict
 from datetime import date 
+import glob
 from contextlib import contextmanager
 
 # [IMPORTS] Third-party
@@ -674,6 +675,85 @@ def update_yamls(
         logger.error(f"[YAML_UPDATE_ERROR] Критическая ошибка: {str(e)}", exc_info=True)
         raise
 
+def consolidate_archive_folders(root_directory: Path, logger: Optional[SupersetLogger] = None) -> None:
+    """
+    Consolidates dashboard folders under a root directory based on the slug (pattern MM-0080 - two latin letters - hyphen - 4 digits)
+    and moves the contents to the folder with the latest modification date.
+
+    Args:
+        root_directory (Path): The root directory containing the dashboard folders.
+
+    Raises:
+        TypeError: If root_directory is not a Path object.
+        ValueError: If root_directory is empty.
+    
+    [CONTRACT]
+    @pre: root_directory must be a valid Path object representing an existing directory.
+    @post: The contents of all folders matching the slug pattern are moved to the folder with the latest modification date for each slug.
+    @invariant: The slug pattern remains consistent throughout the execution.
+    @raise: TypeError if root_directory is not a Path, ValueError if root_directory is empty.
+    """
+
+    # [CONTRACT] Ensure valid input
+    if not isinstance(root_directory, Path):
+        raise TypeError("root_directory must be a Path object.")
+    if not root_directory:
+        raise ValueError("root_directory cannot be empty.")
+    
+    logger.debug(f"[DEBUG] Checking root_folder: {root_directory}")
+
+    # [SECTION] Define the slug pattern
+    slug_pattern = re.compile(r"([A-Z]{2}-\d{4})")  # Capture the first occurrence of the pattern
+
+    # [SECTION] Group folders by slug
+    dashboards_by_slug: dict[str, list[str]] = {}
+    for folder_name in glob.glob(os.path.join(root_directory, '*')):
+        if os.path.isdir(folder_name):
+            logger.debug(f"[DEBUG] Checking folder: {folder_name}")  # Debug log: show the folder name being checked
+            match = slug_pattern.search(folder_name)
+            if match:
+                slug = match.group(1)  # Extract the captured group (the slug)
+                logger.info(f"[INFO] Found slug: {slug} in folder: {folder_name}") #Log when slug is matched
+                logger.debug(f"[DEBUG] Regex match object: {match}") # Log the complete match object
+                if slug not in dashboards_by_slug:
+                    dashboards_by_slug[slug] = []
+                dashboards_by_slug[slug].append(folder_name)
+            else:
+                logger.debug(f"[DEBUG] No slug found in folder: {folder_name}") # Debug log: show when slug is not matched
+        else:
+            logger.debug(f"[DEBUG] Not a directory: {folder_name}") #debug log for when its not a directory
+
+    # [SECTION] Check if any slugs were found
+    if not dashboards_by_slug:
+        logger.warning("[WARN] No folders found matching the slug pattern.")
+        return
+
+    # [SECTION] Iterate through each slug group
+    for slug, folder_list in dashboards_by_slug.items():
+        # [ACTION] Find the folder with the latest modification date
+        latest_folder = max(folder_list, key=os.path.getmtime)
+        logger.info(f"[INFO] Latest folder for slug {slug}: {latest_folder}")
+
+        # [SECTION] Move contents of other folders to the latest folder
+        for folder in folder_list:
+            if folder != latest_folder:
+                # [ACTION] Move contents
+                try:
+                    for item in os.listdir(folder):
+                        s = os.path.join(folder, item)
+                        d = os.path.join(latest_folder, item)
+                        if os.path.isdir(s):
+                            shutil.move(s, d)
+                        else:
+                            shutil.move(s, d)
+
+                    logger.info(f"[INFO] Moved contents of {folder} to {latest_folder}")
+                except Exception as e:
+                    logger.error(f"[ERROR] Failed to move contents of {folder} to {latest_folder}: {e}", exc_info=True)
+                    
+    logger.info("[INFO] Dashboard consolidation completed.")
+    # [COHERENCE_CHECK_PASSED] Function executed successfully and all contracts were met.
+
 def sync_for_git(
     source_path: str,
     destination_path: str,