kanboard-api/main.py

#!/usr/bin/env -S uv run --with-requirements requirements.txt

import os
import sys
import click
from dotenv import load_dotenv

from lib.kanboard_api import KanboardAPI
from lib.interactive import KanboardInteractive
from lib.users import UsersMigrator
from lib.projects import ProjectsMigrator
from lib.tags import TagsMigrator
from lib.tasks import TasksMigrator

load_dotenv()

MAPPING_FILES = {
    "user": "user_mapping.json",
    "project": "project_mapping.json", 
    "column": "column_mapping.json",
    "tag": "tag_mapping.json",
}

SCOPES_ORDER = ["users", "projects", "tags", "tasks"]

def get_env_var(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise EnvironmentError(f"Не найдена переменная окружения: {name}")
    return value

def setup_apis():
    try:
        source = KanboardAPI(
            url=get_env_var("SOURCE_URL"),
            user=get_env_var("SOURCE_USER"),
            token=get_env_var("SOURCE_TOKEN")
        )
        target = KanboardAPI(
            url=get_env_var("TARGET_URL"),
            user=get_env_var("TARGET_USER"),
            token=get_env_var("TARGET_TOKEN")
        )
        return source, target
    except EnvironmentError as e:
        print(e)
        sys.exit(1)

def create_migrator(scope_name, source_api, target_api):
    """Создание одного мигратора для указанной области."""
    if scope_name == "users":
        return UsersMigrator(source_api, target_api, MAPPING_FILES["user"])
    elif scope_name == "projects":
        return ProjectsMigrator(source_api, target_api, MAPPING_FILES["project"], MAPPING_FILES["user"], MAPPING_FILES["column"])
    elif scope_name == "tags":
        return TagsMigrator(source_api, target_api, MAPPING_FILES["tag"], MAPPING_FILES["project"])
    elif scope_name == "tasks":
        return TasksMigrator(source_api, target_api, MAPPING_FILES["project"], MAPPING_FILES["user"], MAPPING_FILES["tag"], MAPPING_FILES["column"])
    else:
        raise ValueError(f"Неизвестный scope: {scope_name}")

def create_plan(scope, reverse_order=False):
    """
    Создание плана выполнения на основе области.
    
    Args:
        scope: Область выполнения (None для полного выполнения)
        reverse_order: Если True, используется обратный порядок для полного выполнения
    
    Returns:
        Список областей для выполнения
    """
    if scope is None:
        # Полное выполнение
        order = list(reversed(SCOPES_ORDER)) if reverse_order else SCOPES_ORDER
        return order
    else:
        # Выполнение только указанной области
        return [scope]

#
# Основные операции
#

def run_interactive(source, target, instance):
    if instance not in ["source", "target"]:
        print(f"Неизвестный экземпляр: '{instance}'. Должен быть 'source' или 'target'")
        sys.exit(1)
    api = source if instance == "source" else target
    interactive = KanboardInteractive(api)
    interactive.run()

def run_migrate(source_api, target_api, scope):
    """Запуск миграции с созданием миграторов по мере выполнения."""
    
    if scope and scope not in SCOPES_ORDER:
        click.echo(f"Неизвестный scope: '{scope}'. Доступные: {', '.join(SCOPES_ORDER)}", err=True)
        sys.exit(1)
    
    migration_plan = create_plan(scope, reverse_order=False)
    
    for scope_name in migration_plan:
        # Создаем мигратор непосредственно перед выполнением
        migrator = create_migrator(scope_name, source_api, target_api)
        try:
            click.echo(f"=== Выполнение миграции: {scope_name} ===")
            migrator.migrate()
            click.echo(f"=== Миграция {scope_name} завершена ===\n")
        except Exception as e:
            click.echo(f"Ошибка при миграции {scope_name}: {e}", err=True)
            sys.exit(1)

def run_delete(source_api, target_api, scope):
    """Удаление данных из целевого экземпляра с созданием миграторов по мере выполнения."""
    
    if scope and scope not in SCOPES_ORDER:
        click.echo(f"Неизвестный scope: '{scope}'. Доступные: {', '.join(SCOPES_ORDER)}", err=True)
        sys.exit(1)
    
    deletion_plan = create_plan(scope, reverse_order=True)
    
    for scope_name in deletion_plan:
        # Создаем мигратор непосредственно перед выполнением
        deleter = create_migrator(scope_name, source_api, target_api)
        try:
            click.echo(f"=== Выполнение удаления: {scope_name} ===")
            deleter.delete_all()
            click.echo(f"=== Удаление {scope_name} завершено ===\n")
        except Exception as e:
            click.echo(f"Ошибка при удалении {scope_name}: {e}", err=True)
            sys.exit(1)

@click.group()
def cli():
    """Скрипт для миграции Kanboard. Используйте 'command --help' для деталей."""

@cli.command()
@click.argument('instance',
                type=click.Choice(['source', 'target']))
def interactive(instance):
    """Интерактивный режим: укажите source или target."""
    source, target = setup_apis()
    run_interactive(source, target, instance)

@cli.command()
@click.option('--scope',
              type=click.Choice(SCOPES_ORDER),
              default=None,
              help=f"Scope: {', '.join(SCOPES_ORDER)} (пример: --scope projects)")
def migrate(scope):
    """Миграция данных: укажите --scope или используйте для полного режима."""
    source, target = setup_apis()
    run_migrate(source, target, scope)

@cli.command()
@click.option('--scope',
              type=click.Choice(SCOPES_ORDER),
              default=None,
              help=f"Scope: {', '.join(SCOPES_ORDER)} (пример: --scope tags)")
def delete(scope):
    """Удаление данных: укажите --scope или используйте для полного режима."""
    source, target = setup_apis()
    run_delete(source, target, scope)

if __name__ == '__main__':
    cli()