src/liquidcode/commands/discover/middleware.py - master - liquid-g/liquid-code

src/liquidcode/commands/discover/middleware.py
253 строки10 KB
User
0.3.0-RC - дописаны тесты и мелкие исправления
04 июл 2026, 13:07
04 июл 2026, 13:0740ad8b8
"""
Команда для обнаружения middleware в проекте.

Использование: lcc discover:middleware [--directory DIR] [--output FILE] [--include-builtin] [--force] [--list]
"""

import argparse
import importlib
import inspect
import os
import pkgutil
import sys
from typing import List, Dict, Any

from ...middleware import Middleware

try:
    import yaml
except ImportError:
    yaml = None

# Импорт для конфигурации (используем абсолютный импорт для легкого мокинга)
from liquidcode.config import load_config


def configure_parser(subparser: argparse.ArgumentParser) -> None:
    """
    Добавляет аргументы для команды discover:middleware.

    Args:
        subparser: Объект ArgumentParser для команды.
    """
    subparser.add_argument(
        '--directory',
        default='src/Middleware',
        help='Директория с пользовательскими middleware (по умолчанию: src/Middleware)'
    )
    subparser.add_argument(
        '--output',
        default='config/middleware.yaml',
        help='Путь для сохранения конфига (по умолчанию: config/middleware.yaml)'
    )
    subparser.add_argument(
        '--list',
        action='store_true',
        help='Только показать список найденных middleware без генерации конфига'
    )
    subparser.add_argument(
        '--include-builtin',
        action='store_true',
        help='Включить встроенные middleware в конфиг (по умолчанию только пользовательские)'
    )
    subparser.add_argument(
        '--force',
        action='store_true',
        help='Перезаписать файл, если он существует (только с --output)'
    )


def run(args: argparse.Namespace) -> None:
    """
    Выполняет команду discover:middleware.
    """
    # 1. Собираем встроенные middleware (из liquidcode.middleware)
    builtin_middlewares = _discover_builtin_middleware()

    # 2. Собираем пользовательские middleware (из указанной директории)
    user_middlewares = _discover_user_middleware(args.directory)

    # 3. Объединяем
    all_middlewares = []
    if args.include_builtin:
        all_middlewares.extend(builtin_middlewares)
    all_middlewares.extend(user_middlewares)

    if not all_middlewares:
        print("⚠️  Middleware не найдены.")
        return

    # Сортируем по приоритету
    all_middlewares.sort(key=lambda x: x.get('priority', 0))

    # Если указан --list, показываем список
    if args.list:
        _print_list(all_middlewares, args.include_builtin)
        return

    # Если указан --output, генерируем конфиг
    if args.output:
        # Генерация конфига
        if yaml is None:
            print("❌ Для сохранения конфига требуется установить PyYAML: pip install PyYAML")
            sys.exit(1)

    # Проверка существования файла
    if os.path.exists(args.output) and not args.force:
        print(f"⚠️  Файл {args.output} уже существует.")
        response = input("Перезаписать? (y/N): ").strip().lower()
        if response != 'y':
            print("Операция отменена.")
            return

    # Генерируем и сохраняем конфиг
    _generate_config(all_middlewares, args.output, args.include_builtin)


def _discover_builtin_middleware() -> List[Dict[str, Any]]:
    """
    Сканирует пакет liquidcode.middleware и возвращает список встроенных middleware.
    """
    result = []
    try:
        import liquidcode.middleware as middleware_package
        package_path = middleware_package.__path__
        for _, module_name, is_pkg in pkgutil.iter_modules(package_path, middleware_package.__name__ + '.'):
            if is_pkg:
                continue
            try:
                module = importlib.import_module(module_name)
                for name, obj in inspect.getmembers(module, inspect.isclass):
                    if (
                            issubclass(obj, Middleware)
                            and obj is not Middleware
                            and obj.__module__.startswith('liquidcode.middleware')
                    ):
                        priority = getattr(obj, 'priority', 0)
                        result.append({
                            'name': name,
                            'full_name': f"{module_name}.{name}",
                            'priority': priority,
                            'enabled': True,
                            'builtin': True,
                        })
            except ImportError:
                continue
    except (ImportError, AttributeError):
        pass
    return result


def _discover_user_middleware(directory: str) -> List[Dict[str, Any]]:
    """
    Сканирует указанную папку и возвращает список пользовательских middleware.
    """
    result = []
    if not os.path.exists(directory):
        return result

    package_path = directory.replace(os.sep, '.')
    try:
        package = importlib.import_module(package_path)
        for _, module_name, is_pkg in pkgutil.iter_modules(package.__path__, package.__name__ + '.'):
            if is_pkg:
                continue
            try:
                module = importlib.import_module(module_name)
                for name, obj in inspect.getmembers(module, inspect.isclass):
                    if (
                            issubclass(obj, Middleware)
                            and obj is not Middleware
                            and not obj.__module__.startswith('liquidcode.middleware')
                    ):
                        priority = getattr(obj, 'priority', 0)
                        result.append({
                            'name': name,
                            'full_name': f"{module_name}.{name}",
                            'priority': priority,
                            'enabled': True,
                            'builtin': False,
                        })
            except ImportError:
                continue
    except (ImportError, ModuleNotFoundError):
        # Если нет __init__.py, сканируем файлы вручную
        for root, _, files in os.walk(directory):
            for file in files:
                if file.endswith('.py') and not file.startswith('__'):
                    rel_path = os.path.relpath(os.path.join(root, file), start='.')
                    mod_name = rel_path[:-3].replace(os.sep, '.')
                    try:
                        module = importlib.import_module(mod_name)
                        for name, obj in inspect.getmembers(module, inspect.isclass):
                            if (
                                    issubclass(obj, Middleware)
                                    and obj is not Middleware
                                    and not obj.__module__.startswith('liquidcode.middleware')
                            ):
                                priority = getattr(obj, 'priority', 0)
                                result.append({
                                    'name': name,
                                    'full_name': f"{mod_name}.{name}",
                                    'priority': priority,
                                    'enabled': True,
                                    'builtin': False,
                                })
                    except ImportError:
                        continue
    return result


def _print_list(middlewares: List[Dict[str, Any]], include_builtin: bool) -> None:
    """
    Выводит список middleware в консоль.
    """
    print("🔍 Найденные middleware:")
    for item in middlewares:
        builtin_mark = " [встроенный]" if item.get('builtin', False) else ""
        status = "✅" if item.get('enabled', True) else "❌"
        print(f"  {status} {item['name']}{builtin_mark} (приоритет: {item['priority']})")


def _generate_config(middlewares: List[Dict[str, Any]], output_path: str, include_builtin: bool) -> None:
    """
    Генерирует YAML-конфиг с учётом существующих ручных правок.
    """
    if yaml is None:
        print("❌ Для генерации YAML-конфига требуется установить PyYAML: pip install PyYAML")
        return

    # Формируем новые данные
    new_config = {
        'middleware': [
            {
                'class': item['full_name'],
                'priority': item.get('priority', 0),
                'enabled': item.get('enabled', True),
            }
            for item in middlewares
        ]
    }

    # Загружаем существующий конфиг, если есть, чтобы сохранить ручные правки
    existing_config = {}
    if os.path.exists(output_path):
        try:
            existing_config = load_config(output_path)
            existing_middleware = existing_config.get('middleware', [])
            existing_map = {item['class']: item for item in existing_middleware}
            for item in new_config['middleware']:
                if item['class'] in existing_map:
                    # Сохраняем пользовательские изменения
                    item['priority'] = existing_map[item['class']].get('priority', item['priority'])
                    item['enabled'] = existing_map[item['class']].get('enabled', item['enabled'])
        except Exception:
            pass

    # Сохраняем конфиг
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    with open(output_path, 'w', encoding='utf-8') as f:
        yaml.dump(new_config, f, default_flow_style=False, allow_unicode=True)

    print(f"✅ Конфиг middleware сохранён в {output_path}")
    print("📝 Теперь вы можете включать/отключать middleware и менять приоритеты в этом файле.")