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 == "" — первый файл
|