# -*- coding: utf-8 -*- # CONTRACT: # PURPOSE: Интерактивный скрипт для миграции ассетов Superset между различными окружениями. # SPECIFICATION_LINK: mod_migration_script # PRECONDITIONS: Наличие корректных конфигурационных файлов для подключения к Superset. # POSTCONDITIONS: Выбранные ассеты успешно перенесены из исходного в целевое окружение. # IMPORTS: [argparse, superset_tool.client, superset_tool.utils.init_clients, superset_tool.utils.logger, superset_tool.utils.fileio] """ [MODULE] Superset Migration Tool @description: Интерактивный скрипт для миграции ассетов Superset между различными окружениями. """ from whiptail import Whiptail # [IMPORTS] from superset_tool.client import SupersetClient from superset_tool.utils.init_clients import setup_clients from superset_tool.utils.logger import SupersetLogger from superset_tool.utils.fileio import ( save_and_unpack_dashboard, read_dashboard_from_disk, update_yamls, create_dashboard_export ) # [ENTITY: Class('Migration')] # CONTRACT: # PURPOSE: Инкапсулирует логику и состояние процесса миграции. # SPECIFICATION_LINK: class_migration # ATTRIBUTES: # - name: logger, type: SupersetLogger, description: Экземпляр логгера. # - name: from_c, type: SupersetClient, description: Клиент для исходного окружения. # - name: to_c, type: SupersetClient, description: Клиент для целевого окружения. # - name: dashboards_to_migrate, type: list, description: Список дашбордов для миграции. # - name: db_config_replacement, type: dict, description: Конфигурация для замены данных БД. class Migration: """ Класс для управления процессом миграции дашбордов Superset. """ def __init__(self): self.logger = SupersetLogger(name="migration_script") self.from_c: SupersetClient = None self.to_c: SupersetClient = None self.dashboards_to_migrate = [] self.db_config_replacement = None # END_FUNCTION___init__ # [ENTITY: Function('run')] # CONTRACT: # PURPOSE: Запускает основной воркфлоу миграции, координируя все шаги. # SPECIFICATION_LINK: func_run_migration # PRECONDITIONS: None # POSTCONDITIONS: Процесс миграции завершен. def run(self): """Запускает основной воркфлоу миграции.""" self.logger.info("[INFO][run][ENTER] Запуск скрипта миграции.") self.select_environments() self.select_dashboards() self.confirm_db_config_replacement() self.execute_migration() self.logger.info("[INFO][run][EXIT] Скрипт миграции завершен.") # END_FUNCTION_run # [ENTITY: Function('select_environments')] # CONTRACT: # PURPOSE: Шаг 1. Обеспечивает интерактивный выбор исходного и целевого окружений. # SPECIFICATION_LINK: func_select_environments # PRECONDITIONS: None # POSTCONDITIONS: Атрибуты `self.from_c` и `self.to_c` инициализированы валидными клиентами Superset. def select_environments(self): """Шаг 1: Выбор окружений (источник и назначение).""" self.logger.info("[INFO][select_environments][ENTER] Шаг 1/4: Выбор окружений.") try: all_clients = setup_clients(self.logger) available_envs = list(all_clients.keys()) except Exception as e: self.logger.error(f"[ERROR][select_environments][FAILURE] Ошибка при инициализации клиентов: {e}", exc_info=True) w = Whiptail(title="Ошибка", backtitle="Superset Migration Tool") w.msgbox("Не удалось инициализировать клиенты. Проверьте конфигурацию.") return w = Whiptail(title="Выбор окружения", backtitle="Superset Migration Tool") # Select source environment (return_code, from_env_name) = w.menu("Выберите исходное окружение:", available_envs) if return_code == 0: self.from_c = all_clients[from_env_name] self.logger.info(f"[INFO][select_environments][STATE] Исходное окружение: {from_env_name}") else: return # Select target environment available_envs.remove(from_env_name) (return_code, to_env_name) = w.menu("Выберите целевое окружение:", available_envs) if return_code == 0: self.to_c = all_clients[to_env_name] self.logger.info(f"[INFO][select_environments][STATE] Целевое окружение: {to_env_name}") else: return self.logger.info("[INFO][select_environments][EXIT] Шаг 1 завершен.") # END_FUNCTION_select_environments # [ENTITY: Function('select_dashboards')] # CONTRACT: # PURPOSE: Шаг 2. Обеспечивает интерактивный выбор дашбордов для миграции. # SPECIFICATION_LINK: func_select_dashboards # PRECONDITIONS: `self.from_c` должен быть инициализирован. # POSTCONDITIONS: `self.dashboards_to_migrate` содержит список выбранных дашбордов. def select_dashboards(self): """Шаг 2: Выбор дашбордов для миграции.""" self.logger.info("[INFO][select_dashboards][ENTER] Шаг 2/4: Выбор дашбордов.") try: _, all_dashboards = self.from_c.get_dashboards() if not all_dashboards: self.logger.warning("[WARN][select_dashboards][STATE] В исходном окружении не найдено дашбордов.") w = Whiptail(title="Информация", backtitle="Superset Migration Tool") w.msgbox("В исходном окружении не найдено дашбордов.") return w = Whiptail(title="Выбор дашбордов", backtitle="Superset Migration Tool") dashboard_options = [(str(d['id']), d['dashboard_title']) for d in all_dashboards] (return_code, selected_ids) = w.checklist("Выберите дашборды для миграции:", dashboard_options) if return_code == 0: self.dashboards_to_migrate = [d for d in all_dashboards if str(d['id']) in selected_ids] self.logger.info(f"[INFO][select_dashboards][STATE] Выбрано дашбордов: {len(self.dashboards_to_migrate)}") except Exception as e: self.logger.error(f"[ERROR][select_dashboards][FAILURE] Ошибка при получении или выборе дашбордов: {e}", exc_info=True) w = Whiptail(title="Ошибка", backtitle="Superset Migration Tool") w.msgbox("Произошла ошибка при работе с дашбордами.") self.logger.info("[INFO][select_dashboards][EXIT] Шаг 2 завершен.") # END_FUNCTION_select_dashboards # [ENTITY: Function('confirm_db_config_replacement')] # CONTRACT: # PURPOSE: Шаг 3. Управляет процессом подтверждения и настройки замены конфигураций БД. # SPECIFICATION_LINK: func_confirm_db_config_replacement # PRECONDITIONS: `self.from_c` и `self.to_c` инициализированы. # POSTCONDITIONS: `self.db_config_replacement` содержит конфигурацию для замены или `None`. def confirm_db_config_replacement(self): """Шаг 3: Подтверждение и настройка замены конфигурации БД.""" self.logger.info("[INFO][confirm_db_config_replacement][ENTER] Шаг 3/4: Замена конфигурации БД.") w = Whiptail(title="Замена конфигурации БД", backtitle="Superset Migration Tool") if w.yesno("Хотите ли вы заменить конфигурации баз данных в YAML-файлах?"): (return_code, old_db_name) = w.inputbox("Введите имя заменяемой базы данных (например, db_dev):") if return_code != 0: return (return_code, new_db_name) = w.inputbox("Введите новое имя базы данных (например, db_prod):") if return_code != 0: return self.db_config_replacement = {"old": {"database_name": old_db_name}, "new": {"database_name": new_db_name}} self.logger.info(f"[INFO][confirm_db_config_replacement][STATE] Установлена ручная замена: {self.db_config_replacement}") else: self.logger.info("[INFO][confirm_db_config_replacement][STATE] Замена конфигурации БД пропущена.") self.logger.info("[INFO][confirm_db_config_replacement][EXIT] Шаг 3 завершен.") # END_FUNCTION_confirm_db_config_replacement # [ENTITY: Function('execute_migration')] # CONTRACT: # PURPOSE: Шаг 4. Выполняет фактическую миграцию выбранных дашбордов. # SPECIFICATION_LINK: func_execute_migration # PRECONDITIONS: Все предыдущие шаги (`select_environments`, `select_dashboards`) успешно выполнены. # POSTCONDITIONS: Выбранные дашборды перенесены в целевое окружение. def execute_migration(self): """Шаг 4: Выполнение миграции и обновления конфигураций.""" self.logger.info("[INFO][execute_migration][ENTER] Шаг 4/4: Выполнение миграции.") w = Whiptail(title="Выполнение миграции", backtitle="Superset Migration Tool") if not self.dashboards_to_migrate: self.logger.warning("[WARN][execute_migration][STATE] Нет дашбордов для миграции.") w.msgbox("Нет дашбордов для миграции. Завершение.") return total_dashboards = len(self.dashboards_to_migrate) self.logger.info(f"[INFO][execute_migration][STATE] Начало миграции {total_dashboards} дашбордов.") with w.gauge("Выполняется миграция...", width=60, height=10) as gauge: for i, dashboard in enumerate(self.dashboards_to_migrate): try: dashboard_id = dashboard['id'] dashboard_title = dashboard['dashboard_title'] progress = int((i / total_dashboards) * 100) self.logger.debug(f"[DEBUG][execute_migration][PROGRESS] {progress}% - Миграция: {dashboard_title}") gauge.set_text(f"Миграция: {dashboard_title} ({i+1}/{total_dashboards})") gauge.set_percent(progress) self.logger.info(f"[INFO][execute_migration][PROGRESS] Миграция дашборда: {dashboard_title} (ID: {dashboard_id})") # 1. Экспорт exported_content, _ = self.from_c.export_dashboard(dashboard_id) zip_path, unpacked_path = save_and_unpack_dashboard(exported_content, f"temp_export_{dashboard_id}", unpack=True) self.logger.info(f"[INFO][execute_migration][STATE] Дашборд экспортирован и распакован в {unpacked_path}") # 2. Обновление YAML, если нужно if self.db_config_replacement: update_yamls(db_configs=[self.db_config_replacement], path=str(unpacked_path)) self.logger.info(f"[INFO][execute_migration][STATE] YAML-файлы обновлены.") # 3. Упаковка и импорт new_zip_path = f"migrated_dashboard_{dashboard_id}.zip" create_dashboard_export(new_zip_path, [str(unpacked_path)]) self.to_c.import_dashboard(new_zip_path) self.logger.info(f"[INFO][execute_migration][SUCCESS] Дашборд {dashboard_title} успешно импортирован.") except Exception as e: self.logger.error(f"[ERROR][execute_migration][FAILURE] Ошибка при миграции дашборда {dashboard_title}: {e}", exc_info=True) error_msg = f"Не удалось смигрировать дашборд: {dashboard_title}.\n\nОшибка: {e}" w.msgbox(error_msg, width=60, height=15) gauge.set_percent(100) self.logger.info("[INFO][execute_migration][STATE] Миграция завершена.") w.msgbox("Миграция завершена!", width=40, height=8) self.logger.info("[INFO][execute_migration][EXIT] Шаг 4 завершен.") # END_FUNCTION_execute_migration # END_CLASS_Migration # [MAIN_EXECUTION_BLOCK] if __name__ == "__main__": migration = Migration() migration.run() # END_MAIN_EXECUTION_BLOCK # END_MODULE_migration_script