ss-tools/backup_script.py

import logging
from datetime import datetime
import shutil
import keyring
import os
from pathlib import Path
from superset_tool.models import SupersetConfig, DatabaseConfig
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


def setup_clients(logger: SupersetLogger):
    """Инициализация клиентов для разных окружений"""
    clients = {}
    try:
        # Конфигурация для Dev
        dev_config = SupersetConfig(
            base_url="https://devta.bi.dwh.rusal.com/api/v1",
            auth={
                "provider": "db",
                "username": "migrate_user",
                "password": keyring.get_password("system", "dev migrate"),
                "refresh": True
            },
            logger=logger,
            verify_ssl=False
        )

        # Конфигурация для Prod
        prod_config = SupersetConfig(
            base_url="https://prodta.bi.dwh.rusal.com/api/v1",
            auth={
                "provider": "db",
                "username": "migrate_user",
                "password": keyring.get_password("system", "prod migrate"),
                "refresh": True
            },
            logger=logger,
            verify_ssl=False
        )

        # Конфигурация для Sandbox
        sandbox_config = SupersetConfig(
            base_url="https://sandboxta.bi.dwh.rusal.com/api/v1",
            auth={
                "provider": "db",
                "username": "migrate_user",
                "password": keyring.get_password("system", "sandbox migrate"),
                "refresh": True
            },
            logger=logger,
            verify_ssl=False
        )

        clients['dev'] = SupersetClient(dev_config)
        clients['sbx'] = SupersetClient(sandbox_config)
        clients['prod'] = SupersetClient(prod_config)
        logger.info("Клиенты для окружений успешно инициализированы")
        return clients
    except Exception as e:
        logger.error(f"Ошибка инициализации клиентов: {str(e)}")
        raise

def backup_dashboards(client, env_name, backup_root, logger):
    """Выполнение бэкапа дашбордов с детальным логированием ошибок"""
    try:
        dashboard_count, dashboard_meta = client.get_dashboards()
        if dashboard_count == 0:
            logger.warning(f"Нет дашбордов для экспорта в {env_name}")
            return True

        success = 0
        errors = []
        
        for db in dashboard_meta:
            if not db.get('slug'):
                continue

            try:
                dashboard_title = db['dashboard_title']
                dashboard_dir = Path(backup_root) / env_name / sanitize_filename(dashboard_title)
                dashboard_dir.mkdir(parents=True, exist_ok=True)
                
                zip_content, filename = client.export_dashboard(db['id'])
                save_and_unpack_dashboard(
                    zip_content=zip_content,
                    original_filename=filename,
                    output_dir=dashboard_dir,
                    unpack=False
                )
                
                # Архивирование старых бэкапов
                try:
                    archive_exports(dashboard_dir)
                except Exception as cleanup_error:
                    logger.warning(f"Ошибка очистки архива: {cleanup_error}")
                
                success += 1
                
            except Exception as db_error:
                error_info = {
                    'dashboard': db.get('dashboard_title'),
                    'error': str(db_error),
                    'env': env_name
                }
                errors.append(error_info)
                logger.error("Ошибка экспорта дашборда", extra=error_info)

        if errors:
            logger.error(f"Итоги экспорта для {env_name}",
                       extra={'success': success, 'errors': errors, 'total': dashboard_count})
            
        return len(errors) == 0
        
    except Exception as e:
        logger.critical(f"Фатальная ошибка бэкапа {env_name}: {str(e)}", exc_info=True)
        return False

def main():
    # Инициализация логгера
    log_dir = Path("P:\\Superset\\010 Бекапы\\Logs")
    logger = SupersetLogger(
        log_dir=log_dir,
        level=logging.INFO,
        console=True
    )    
    """Основная функция выполнения бэкапа"""
    logger.info("="*50)
    logger.info("Запуск процесса бэкапа Superset")
    logger.info("="*50)
    
    try:
        clients = setup_clients(logger)
        superset_backup_repo = Path("P:\\Superset\\010 Бекапы")
        
        # Бэкап для DEV
        dev_success = backup_dashboards(
            clients['dev'], 
            "DEV", 
            superset_backup_repo,
            logger=logger
        )
        
        #Бэкап для Sandbox
        sbx_success = backup_dashboards(
            clients['sbx'], 
            "SBX", 
            superset_backup_repo,
            logger=logger
        )
        #Бэкап для Прода
        prod_success = backup_dashboards(
            clients['prod'], 
            "PROD", 
            superset_backup_repo,
            logger=logger
        )
        
        # Итоговый отчет
        logger.info("="*50)
        logger.info("Итоги выполнения бэкапа:")
        logger.info(f"DEV: {'Успешно' if dev_success else 'С ошибками'}")
        logger.info(f"SBX: {'Успешно' if sbx_success else 'С ошибками'}")
        logger.info(f"PROD: {'Успешно' if prod_success else 'С ошибками'}")
        logger.info(f"Полный лог доступен в: {log_dir}")
        
    except Exception as e:
        logger.critical(f"Фатальная ошибка выполнения скрипта: {str(e)}", exc_info=True)
        return 1
    
    logger.info("Процесс бэкапа завершен")
    return 0

if __name__ == "__main__":
    exit_code = main()
    exit(exit_code)