132 lines
5.5 KiB
Python
132 lines
5.5 KiB
Python
"""
|
||
Модуль 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:
|
||
"""
|
||
Собирает файловую структуру проекта и формирует первичное
|
||
представление модулей.
|
||
|
||
Стратегия разбивки на модули: каждая директория первого уровня
|
||
является отдельным модулем. Файлы непосредственно в корне
|
||
помещаются в псевдомодуль "<root>".
|
||
"""
|
||
|
||
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:
|
||
"""
|
||
Определяет имя модуля для файла.
|
||
Файлы в корне → "<root>".
|
||
Файлы в поддиректории → относительный путь к директории первого уровня.
|
||
"""
|
||
try:
|
||
rel = file.relative_to(root)
|
||
except ValueError:
|
||
return "<root>"
|
||
|
||
parts = rel.parts
|
||
if len(parts) == 1:
|
||
return "<root>"
|
||
# Берём путь относительно корня до файла (вся директория как модуль)
|
||
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 == "" — первый файл
|