SysDevCleaner/sysdevcleaner/core/devscan.py

"""Рекурсивный поиск артефактов сборки и зависимостей внутри проектов."""
from __future__ import annotations

import fnmatch
import os
from typing import Callable, Iterable

from .model import Category, Finding
from .rules import NEVER_DESCEND, PROJECT_RULES, SKIP_DIRS, ProjectRule
from .util import Cancel, dir_stats, human

# имя каталога -> список правил с таким именем
_BY_NAME: dict[str, list[ProjectRule]] = {}
for _r in PROJECT_RULES:
    _BY_NAME.setdefault(_r.name.lower(), []).append(_r)

# файлы-маркеры, по которым каталог считается корнем проекта
PROJECT_MARKERS = (
    "package.json", "pyproject.toml", "setup.py", "requirements.txt", "Cargo.toml",
    "go.mod", "pom.xml", "build.gradle", "build.gradle.kts", "composer.json", "Gemfile",
    "pubspec.yaml", "CMakeLists.txt", "Makefile", ".git", "*.sln", "*.csproj",
    "ProjectSettings", "*.uproject",
)


def _matches_marker(entries: set[str], markers: Iterable[str]) -> bool:
    """Есть ли рядом файл, подтверждающий тип проекта."""
    for m in markers:
        if "*" in m or "?" in m:
            if any(fnmatch.fnmatch(e, m.lower()) for e in entries):
                return True
        elif m.lower() in entries:
            return True
    return False


def _has_inner(path: str, inner: Iterable[str]) -> bool:
    """Есть ли внутри каталога подтверждающий файл (pyvenv.cfg и т.п.)."""
    try:
        names = {e.name.lower() for e in os.scandir(path)}
    except (OSError, ValueError):
        return False
    return any(m.lower() in names for m in inner)


def scan_projects(
    roots: Iterable[str],
    cancel: Cancel | None = None,
    progress: Callable[[str], None] | None = None,
    max_depth: int = 9,
    min_size: int = 1024 * 1024,
    include_hidden: bool = False,
) -> list[Finding]:
    """Обойти ``roots`` и найти node_modules, target, build, __pycache__ и т.п.

    Найденный артефакт не раскрывается вглубь — его размер считается целиком,
    а вложенные артефакты не дублируются в отчёте.
    """
    findings: list[Finding] = []
    seen: set[str] = set()
    scanned = 0

    for root in roots:
        root = os.path.abspath(root)
        if not os.path.isdir(root):
            continue
        stack: list[tuple[str, int]] = [(root, 0)]
        while stack:
            if cancel is not None and cancel.cancelled:
                return findings
            cur, depth = stack.pop()
            if depth > max_depth:
                continue
            try:
                entries = list(os.scandir(cur))
            except (OSError, ValueError):
                continue

            scanned += 1
            if progress is not None and scanned % 200 == 0:
                progress(cur)

            names = {e.name.lower() for e in entries}
            subdirs: list[os.DirEntry] = []
            for e in entries:
                try:
                    if not e.is_dir(follow_symlinks=False):
                        continue
                except OSError:
                    continue
                subdirs.append(e)

            for e in subdirs:
                low = e.name.lower()
                if low in SKIP_DIRS:
                    continue
                if not include_hidden and low.startswith(".") and low not in _BY_NAME:
                    # скрытые каталоги без собственного правила пропускаем
                    continue

                rules = _BY_NAME.get(low)
                matched: ProjectRule | None = None
                if rules:
                    for rule in rules:
                        if rule.marker and not _matches_marker(names, rule.marker):
                            continue
                        if rule.inner and not _has_inner(e.path, rule.inner):
                            continue
                        matched = rule
                        break

                if matched is not None:
                    key = os.path.normcase(e.path)
                    if key in seen:
                        continue
                    seen.add(key)
                    size, nfiles, ndirs, mtime = dir_stats(e.path, cancel)
                    if size < min_size:
                        continue
                    project = os.path.basename(cur) or cur
                    findings.append(Finding(
                        path=e.path,
                        title=f"{matched.title} — {project}",
                        category=Category.DEV_PROJECT,
                        risk=matched.risk,
                        size=size, files=nfiles, dirs=ndirs, mtime=mtime,
                        detail=(f"{matched.detail}\nПроект: {cur}\n"
                                f"Экосистема: {matched.ecosystem or '—'} · "
                                f"{nfiles:,} файлов · {human(size)}".replace(",", " ")),
                        group=matched.title,
                        extra={"project": cur, "ecosystem": matched.ecosystem,
                               "artifact": matched.name},
                    ))
                    if matched.stop:
                        continue  # внутрь не идём

                if low in NEVER_DESCEND:
                    continue      # чужие зависимости — своих проектов внутри нет
                stack.append((e.path, depth + 1))

    findings.sort(key=lambda f: -f.size)
    return findings


def find_project_roots(
    roots: Iterable[str],
    cancel: Cancel | None = None,
    max_depth: int = 6,
) -> list[str]:
    """Быстро найти корни проектов — чтобы показать пользователю, где он программирует."""
    found: list[str] = []
    for root in roots:
        root = os.path.abspath(root)
        if not os.path.isdir(root):
            continue
        stack = [(root, 0)]
        while stack:
            if cancel is not None and cancel.cancelled:
                return found
            cur, depth = stack.pop()
            if depth > max_depth:
                continue
            try:
                entries = list(os.scandir(cur))
            except (OSError, ValueError):
                continue
            names = {e.name.lower() for e in entries}
            if _matches_marker(names, PROJECT_MARKERS):
                found.append(cur)
                continue  # вложенные подпроекты не ищем
            for e in entries:
                try:
                    if e.is_dir(follow_symlinks=False) and e.name.lower() not in SKIP_DIRS \
                            and not e.name.startswith("."):
                        stack.append((e.path, depth + 1))
                except OSError:
                    continue
    return found


def summarize_by_ecosystem(findings: list[Finding]) -> dict[str, tuple[int, int]]:
    """{'JavaScript': (размер, количество)}"""
    out: dict[str, tuple[int, int]] = {}
    for f in findings:
        eco = f.extra.get("ecosystem") or "Прочее"
        size, cnt = out.get(eco, (0, 0))
        out[eco] = (size + f.size, cnt + 1)
    return dict(sorted(out.items(), key=lambda kv: -kv[1][0]))