graph_builder.py - main - hyperruss/ku2

graph_builder.py
293 строки10 KB
Игорь Грибунов
finish project
08 ноя 2025, 12:50
08 ноя 2025, 12:50487b335
"""
Этап 3: Основные операции
Построение графа зависимостей алгоритмом BFS с рекурсией
Работа с тестовым репозиторием и реальным NuGet API
"""

from typing import Dict, List, Set, Optional, Union
from collections import deque
import sys


class DepGraphBuilder:
    """Построитель графа зависимостей"""

    def __init__(self, fetcher=None):
        """
        Инициализирует построитель графа

        Args:
            fetcher: Объект для получения данных о пакетах (NuGetFetcher или None для тестов)
        """
        self.fetcher = fetcher
        self.visited = set()
        self.cycles = set()

    def build_graph_bfs_recursive(
            self,
            root: str,
            repo: Union[Dict, object],
            max_depth: Optional[int] = None
    ) -> Dict[str, List[str]]:
        """
        Строит граф зависимостей с использованием BFS с рекурсией

        Args:
            root: Корневой пакет
            repo: Репозиторий (словарь для тестов или NuGetFetcher)
            max_depth: Максимальная глубина анализа (None = без ограничений)

        Returns:
            Граф в виде словаря {пакет: [зависимости]}
        """
        self.visited.clear()
        self.cycles.clear()
        graph = {}

        def bfs_recursive(queue: deque, current_depth: int) -> None:
            """Рекурсивная реализация BFS"""

            # Базовый случай: очередь пуста или достигнута макс. глубина
            if not queue or (max_depth is not None and current_depth > max_depth):
                return

            # Обрабатываем пакеты на текущем уровне
            next_queue = deque()
            level_size = len(queue)

            for _ in range(level_size):
                package = queue.popleft()

                # Пропускаем, если уже посещали
                if package in self.visited:
                    continue

                self.visited.add(package)

                # Получаем зависимости
                deps = self._get_dependencies(package, repo)
                graph[package] = deps

                # Добавляем новые зависимости в очередь
                for dep in deps:
                    if dep not in self.visited and dep not in self.cycles:
                        next_queue.append(dep)

            # Рекурсивный вызов для следующего уровня
            bfs_recursive(next_queue, current_depth + 1)

        # Запускаем BFS с корневого узла
        bfs_recursive(deque([root]), 0)

        return graph

    def _get_dependencies(
            self,
            package: str,
            repo: Union[Dict, object]
    ) -> List[str]:
        """
        Получает зависимости пакета

        Args:
            package: Имя пакета
            repo: Репозиторий

        Returns:
            Список зависимостей
        """
        try:
            # Для тестового репозитория (словарь)
            if isinstance(repo, dict):
                return repo.get(package, [])

            # Для реального NuGet API
            else:
                package_info = repo.fetch_package_info(package)
                if package_info:
                    return list(package_info.get('dependencies', {}).keys())
                return []

        except Exception as e:
            print(f"️  Ошибка при получении зависимостей для {package}: {e}")
            return []

    def detect_cycles(self, graph: Dict[str, List[str]]) -> Dict[str, List[str]]:
        """
        Обнаруживает циклические зависимости

        Args:
            graph: Граф зависимостей

        Returns:
            Словарь с обнаруженными циклами
        """
        cycles = {}
        visited = set()
        rec_stack = set()

        def dfs_cycle(node: str, path: List[str]) -> None:
            """DFS для обнаружения циклов"""
            visited.add(node)
            rec_stack.add(node)
            path.append(node)

            for neighbor in graph.get(node, []):
                if neighbor not in visited:
                    dfs_cycle(neighbor, path.copy())
                elif neighbor in rec_stack:
                    # Найден цикл
                    cycle_start = path.index(neighbor)
                    cycle = path[cycle_start:] + [neighbor]
                    cycles[f"cycle_{len(cycles)+1}"] = cycle

            rec_stack.remove(node)

        for node in graph:
            if node not in visited:
                dfs_cycle(node, [])

        return cycles


def load_test_repo(file_path: str) -> Dict[str, List[str]]:
    """
    Загружает тестовый граф из файла

    Формат файла:
    A: B C
    B: D
    C: D E
    D:

    Args:
        file_path: Путь к файлу

    Returns:
        Словарь {пакет: [зависимости]}
    """
    repo = {}
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            for line_num, line in enumerate(f, 1):
                line = line.strip()

                # Пропускаем пустые строки и комментарии
                if not line or line.startswith('#'):
                    continue

                try:
                    # Парсим формат "package: dep1 dep2"
                    if ':' not in line:
                        print(f"⚠️  Предупреждение: строка {line_num} не содержит ':'")
                        continue

                    package, deps = line.split(':', 1)
                    package = package.strip()

                    if not package:
                        print(f"⚠️  Предупреждение: строка {line_num} имеет пустое имя пакета")
                        continue

                    deps_list = [d.strip() for d in deps.split() if d.strip()]
                    repo[package] = deps_list

                except ValueError as e:
                    print(f"⚠️  Ошибка парсинга строки {line_num}: {e}")
                    continue

        if not repo:
            print(f" Ошибка: файл {file_path} пуст или не содержит валидных зависимостей")
            return {}

        return repo

    except FileNotFoundError:
        print(f" Ошибка: файл {file_path} не найден")
        return {}
    except Exception as e:
        print(f" Ошибка при чтении файла: {e}")
        return {}


def print_graph(graph: Dict[str, List[str]], title: str = "ГРАФ ЗАВИСИМОСТЕЙ") -> None:
    """
    Выводит граф в красивом формате

    Args:
        graph: Граф зависимостей
        title: Заголовок для вывода
    """
    print("\n" + "="*70)
    print(f" {title}")
    print("="*70)

    if not graph:
        print("     Граф пуст")
    else:
        print(f"   Всего узлов: {len(graph)}\n")

        for package, deps in sorted(graph.items()):
            if deps:
                deps_str = ", ".join(deps)
                print(f"   {package:30} → {deps_str}")
            else:
                print(f"   {package:30} → (нет зависимостей)")

    print("="*70 + "\n")


def demonstrate_stage_3():
    """Функция для демонстрации Этапа 3"""
    print("\n" + "="*70)
    print(" ДЕМОНСТРАЦИЯ ЭТАПА 3: ПОСТРОЕНИЕ ГРАФА ЗАВИСИМОСТЕЙ")
    print("="*70 + "\n")

    # Создаем тестовый репозиторий в памяти
    test_repo = {
        'A': ['B', 'C'],
        'B': ['D', 'E'],
        'C': ['E', 'F'],
        'D': ['G'],
        'E': ['G'],
        'F': [],
        'G': [],
        'H': ['I'],
        'I': ['A']  # Это создаст цикл: I -> A -> B -> ... и потом A -> ... -> I
    }

    print(" Тестовый репозиторий загружен:\n")
    for pkg, deps in test_repo.items():
        deps_str = ", ".join(deps) if deps else "(нет зависимостей)"
        print(f"   {pkg}: {deps_str}")

    print("\n" + "-"*70)

    # Строим граф
    builder = DepGraphBuilder()

    print("\n Построение графа зависимостей для пакета 'A' (max_depth=3)...")
    graph = builder.build_graph_bfs_recursive('A', test_repo, max_depth=3)

    print_graph(graph, "ПОЛНЫЙ ГРАФ ЗАВИСИМОСТЕЙ (максимальная глубина=3)")

    # Обнаруживаем циклы
    print(" Проверка на циклические зависимости...")
    cycles = builder.detect_cycles(graph)

    if cycles:
        print("️  НАЙДЕНЫ ЦИКЛИЧЕСКИЕ ЗАВИСИМОСТИ:\n")
        for cycle_name, cycle_path in cycles.items():
            cycle_str = " → ".join(cycle_path)
            print(f"   {cycle_name}: {cycle_str}")
    else:
        print(" Циклические зависимости не обнаружены\n")

    # Тестируем с лимитом глубины
    print(" Построение графа с ограничением max_depth=1...")
    graph_shallow = builder.build_graph_bfs_recursive('A', test_repo, max_depth=1)
    print_graph(graph_shallow, "ГРАФ С ОГРАНИЧЕНИЕМ ГЛУБИНЫ (max_depth=1)")


if __name__ == '__main__':
    demonstrate_stage_3()