From 79984ab56b859c25de6155fc3821d1fd4858bd75 Mon Sep 17 00:00:00 2001 From: Volobuev Andrey Date: Mon, 30 Jun 2025 13:09:25 +0300 Subject: [PATCH] add clean folders in backup --- backup_script.py | 94 ++++++++++++++++++++++++----------- superset_tool/utils/fileio.py | 80 +++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 30 deletions(-) diff --git a/backup_script.py b/backup_script.py index 82b51ec..a74dbe2 100644 --- a/backup_script.py +++ b/backup_script.py @@ -24,7 +24,7 @@ import keyring from superset_tool.models import SupersetConfig from superset_tool.client import SupersetClient from superset_tool.utils.logger import SupersetLogger -from superset_tool.utils.fileio import save_and_unpack_dashboard, archive_exports, sanitize_filename +from superset_tool.utils.fileio import save_and_unpack_dashboard, archive_exports, sanitize_filename,consolidate_archive_folders,remove_empty_directories # [COHERENCE_CHECK_PASSED] Все необходимые модули импортированы и согласованы. # [FUNCTION] setup_clients @@ -92,25 +92,35 @@ def setup_clients(logger: SupersetLogger): raise # [FUNCTION] backup_dashboards -# @contract: Выполняет бэкап всех доступных дашбордов для заданного клиента и окружения. -# @pre: -# - `client` должен быть инициализированным экземпляром `SupersetClient`. -# - `env_name` должен быть строкой, обозначающей окружение. -# - `backup_root` должен быть валидным путем к корневой директории бэкапа. -# - `logger` должен быть инициализирован. -# @post: -# - Дашборды экспортируются и сохраняются в поддиректориях `backup_root/env_name/dashboard_title`. -# - Старые экспорты архивируются. -# - Возвращает `True` если все дашборды были экспортированы без критических ошибок, `False` иначе. -# @side_effects: -# - Создает директории и файлы в файловой системе. -# - Логирует статус выполнения, успешные экспорты и ошибки. -# @exceptions: -# - `SupersetAPIError`, `NetworkError`, `DashboardNotFoundError`, `ExportError` могут быть подняты методами `SupersetClient` и будут логированы. -def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path, logger: SupersetLogger) -> bool: - """Выполнение бэкапа дашбордов с детальным логированием ошибок""" +def backup_dashboards(client: SupersetClient, + env_name: str, + backup_root: Path, + logger: SupersetLogger, + consolidate: bool = True, + rotate_archive: bool = True, + clean_folders:bool = True) -> bool: + """ [CONTRACT] Выполняет бэкап всех доступных дашбордов для заданного клиента и окружения. + @pre: + - `client` должен быть инициализированным экземпляром `SupersetClient`. + - `env_name` должен быть строкой, обозначающей окружение. + - `backup_root` должен быть валидным путем к корневой директории бэкапа. + - `logger` должен быть инициализирован. + @post: + - Дашборды экспортируются и сохраняются в поддиректориях `backup_root/env_name/dashboard_title`. + - Старые экспорты архивируются. + - Возвращает `True` если все дашборды были экспортированы без критических ошибок, `False` иначе. + @side_effects: + - Создает директории и файлы в файловой системе. + - Логирует статус выполнения, успешные экспорты и ошибки. + @exceptions: + - `SupersetAPIError`, `NetworkError`, `DashboardNotFoundError`, `ExportError` могут быть подняты методами `SupersetClient` и будут логированы.""" # [ANCHOR] DASHBOARD_BACKUP_PROCESS logger.info(f"[INFO] Запуск бэкапа дашбордов для окружения: {env_name}") + logger.debug( + "[PARAMS] Флаги: consolidate=%s, rotate_archive=%s, clean_folders=%s", + consolidate, rotate_archive, clean_folders, + extra={"env": env_name} # контекст для логирования + ) try: dashboard_count, dashboard_meta = client.get_dashboards() logger.info(f"[INFO] Найдено {dashboard_count} дашбордов для экспорта в {env_name}") @@ -139,7 +149,7 @@ def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path, try: # [ANCHOR] CREATE_DASHBOARD_DIR # Используем slug в пути для большей уникальности и избежания конфликтов имен - dashboard_base_dir_name = sanitize_filename(f"{dashboard_slug}-{dashboard_title}") + dashboard_base_dir_name = sanitize_filename(f"{dashboard_title}") dashboard_dir = backup_root / env_name / dashboard_base_dir_name dashboard_dir.mkdir(parents=True, exist_ok=True) logger.debug(f"[DEBUG] Директория для дашборда: {dashboard_dir}") @@ -158,18 +168,20 @@ def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path, ) logger.info(f"[INFO] Дашборд '{dashboard_title}' (ID: {dashboard_id}) успешно экспортирован.") + if rotate_archive: # [ANCHOR] ARCHIVE_OLD_BACKUPS - try: - archive_exports(dashboard_dir, logger=logger) - logger.debug(f"[DEBUG] Старые экспорты для '{dashboard_title}' архивированы.") - except Exception as cleanup_error: - logger.warning( - f"[WARN] Ошибка архивирования старых бэкапов для '{dashboard_title}': {cleanup_error}", - exc_info=False # Не показываем полный traceback для очистки, т.к. это второстепенно - ) - + try: + archive_exports(dashboard_dir, logger=logger) + logger.debug(f"[DEBUG] Старые экспорты для '{dashboard_title}' архивированы.") + except Exception as cleanup_error: + logger.warning( + f"[WARN] Ошибка архивирования старых бэкапов для '{dashboard_title}': {cleanup_error}", + exc_info=False # Не показываем полный traceback для очистки, т.к. это второстепенно + ) + success_count += 1 - + + except Exception as db_error: error_info = { 'dashboard_id': dashboard_id, @@ -184,6 +196,28 @@ def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path, extra=error_info, exc_info=True # Логируем полный traceback для ошибок экспорта ) + if consolidate: + # [ANCHOR] Объединяем архивы по SLUG в одну папку с максимальной датой + try: + consolidate_archive_folders(backup_root / env_name , logger=logger) + logger.debug(f"[DEBUG] Файлы для '{dashboard_title}' консолидированы.") + except Exception as consolidate_error: + logger.warning( + f"[WARN] Ошибка консолидации файлов для '{backup_root / env_name}': {consolidate_error}", + exc_info=False # Не показываем полный traceback для консолидации, т.к. это второстепенно + ) + + if clean_folders: + # [ANCHOR] Удаляем пустые папки + try: + dirs_count = remove_empty_directories(backup_root / env_name , logger=logger) + logger.debug(f"[DEBUG] {dirs_count} пустых папок в '{backup_root / env_name }' удалены.") + except Exception as clean_error: + logger.warning( + f"[WARN] Ошибка очистки пустых директорий в '{backup_root / env_name}': {clean_error}", + exc_info=False # Не показываем полный traceback для консолидации, т.к. это второстепенно + ) + if error_details: logger.error( f"[COHERENCE_CHECK_FAILED] Итоги экспорта для {env_name}:", @@ -195,7 +229,7 @@ def backup_dashboards(client: SupersetClient, env_name: str, backup_root: Path, f"[COHERENCE_CHECK_PASSED] Все {success_count} дашбордов для {env_name} успешно экспортированы." ) return True - + except Exception as e: logger.critical( f"[CRITICAL] Фатальная ошибка бэкапа для окружения {env_name}: {str(e)}", diff --git a/superset_tool/utils/fileio.py b/superset_tool/utils/fileio.py index f309dc9..7c41bb6 100644 --- a/superset_tool/utils/fileio.py +++ b/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,