""" Модуль Collector — обход файловой структуры проекта и сбор исходных данных. Отвечает за: - рекурсивный обход дерева каталогов - идентификацию модулей и их границ - подготовку списков файлов для MetricsEngine и DependencyAnalyzer """ import logging from pathlib import Path from typing import Dict, List, Tuple from config import AnalyzerConfig, DEFAULT_CONFIG from models import ModuleMetrics logger = logging.getLogger(__name__) class Collector: """ Собирает файловую структуру проекта и формирует первичное представление модулей. Стратегия разбивки на модули: каждая директория первого уровня является отдельным модулем. Файлы непосредственно в корне помещаются в псевдомодуль "". """ def __init__(self, config: AnalyzerConfig = DEFAULT_CONFIG): self.config = config self._cpp_exts = set(config.cpp_extensions) self._js_exts = set(config.js_extensions) self._ignored = set(config.ignored_dirs) # ------------------------------------------------------------------ # Публичный API # ------------------------------------------------------------------ def collect(self, project_root: str) -> Dict[str, ModuleMetrics]: """ Обходит дерево проекта и возвращает словарь модулей. :param project_root: абсолютный или относительный путь к корню проекта :returns: {module_name: ModuleMetrics} с заполненными полями files и language """ root = Path(project_root).resolve() if not root.exists(): raise FileNotFoundError(f"Корень проекта не найден: {root}") logger.info("Начало сбора данных из: %s", root) modules: Dict[str, ModuleMetrics] = {} self._walk(root, root, modules) # Убираем пустые модули (нет анализируемых файлов) modules = {k: v for k, v in modules.items() if v.files} logger.info("Обнаружено модулей: %d", len(modules)) for name, mod in modules.items(): logger.debug(" %s: %d файл(ов) [%s]", name, len(mod.files), mod.language) return modules # ------------------------------------------------------------------ # Внутренние методы # ------------------------------------------------------------------ def _walk(self, root: Path, current: Path, modules: Dict[str, ModuleMetrics]) -> None: """Рекурсивный обход директории.""" try: entries = sorted(current.iterdir()) except PermissionError: logger.warning("Нет прав на чтение: %s", current) return for entry in entries: if entry.name.startswith("."): continue if entry.name in self._ignored: logger.debug("Пропускаем директорию: %s", entry) continue if entry.is_dir(): self._walk(root, entry, modules) elif entry.is_file(): lang = self._detect_language(entry) if lang is None: continue module_name = self._resolve_module_name(root, entry) if module_name not in modules: modules[module_name] = ModuleMetrics(name=module_name) modules[module_name].files.append(str(entry)) modules[module_name].language = self._merge_language( modules[module_name].language, lang ) def _detect_language(self, path: Path) -> str | None: """Возвращает 'cpp', 'js' или None если файл не анализируется.""" suffix = path.suffix.lower() if suffix in self._cpp_exts: return "cpp" if suffix in self._js_exts: return "js" return None def _resolve_module_name(self, root: Path, file: Path) -> str: """ Определяет имя модуля для файла. Файлы в корне → "". Файлы в поддиректории → относительный путь к директории первого уровня. """ try: rel = file.relative_to(root) except ValueError: return "" parts = rel.parts if len(parts) == 1: return "" # Берём путь относительно корня до файла (вся директория как модуль) return str(Path(*parts[:-1])) @staticmethod def _merge_language(existing: str, new_lang: str) -> str: """Объединяет языки при добавлении нового файла в модуль.""" if existing == "mixed" or existing == new_lang: return existing if existing != "mixed" else "mixed" if existing in ("cpp", "js") and existing != new_lang: return "mixed" return new_lang # existing == "" — первый файл