"""Рекурсивный поиск артефактов сборки и зависимостей внутри проектов.""" 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]))