pyclean.modern

Complexity

Total Complexity

83

Size/Duplication

Total Lines	505
Duplicated Lines	0 %

Importance

Changes

0

19 Functions

Rating	Name	Duplication	Size	Complexity
A	remove_directory()	0	9	2
A	print_filename()	0	4	1
A	remove_file()	0	9	2
A	print_dirname()	0	4	1
A	normalize()	0	8	1
B	should_ignore()	0	29	8
A	confirm()	0	8	2
C	delete_filesystem_objects()	0	34	10
A	detect_debris_in_directory()	0	18	5
A	recursive_delete_debris()	0	23	5
A	build_git_clean_command()	0	25	4
A	run_git_clean()	0	31	3
B	descend_and_clean()	0	19	7
B	remove_empty_directories()	0	25	6
A	remove_freeform_targets()	0	19	2
B	execute_git_clean()	0	36	5
A	remove_debris_for()	0	8	1
C	pyclean()	0	40	10
A	suggest_debris_option()	0	24	4

# SPDX-FileCopyrightText: 2020 Peter Bittner <django@bittner.it>
#
# SPDX-License-Identifier: GPL-3.0-or-later

"""
Modern, cross-platform, pure-Python pyclean implementation.
"""

import logging
import os
import subprocess
from pathlib import Path

BYTECODE_FILES = ['.pyc', '.pyo']
BYTECODE_DIRS = ['__pycache__']
DEBRIS_TOPICS = {
    'cache': [
        '.cache/**/*',
        '.cache/',
    ],
    'coverage': [
        '.coverage',
        'coverage.json',
        'coverage.lcov',
        'coverage.xml',
        'htmlcov/**/*',
        'htmlcov/',
    ],
    'jupyter': [
        '.ipynb_checkpoints/**/*',
        '.ipynb_checkpoints/',
    ],
    'mypy': [
        '.mypy_cache/**/*',
        '.mypy_cache/',
    ],
    'package': [
        'build/bdist.*/**/*',
        'build/bdist.*/',
        'build/lib/**/*',
        'build/lib/',
        'build/',
        'dist/**/*',
        'dist/',
        'sdist/**/*',
        'sdist/',
        '*.egg-info/**/*',
        '*.egg-info/',
    ],
    'pytest': [
        '.pytest_cache/**/*',
        '.pytest_cache/',
        'pytestdebug.log',
    ],
    'ruff': [
        '.ruff_cache/**/*',
        '.ruff_cache/',
    ],
    'tox': [
        '.tox/**/*',
        '.tox/',
    ],
}


class CleanupRunner:
    """Module-level configuration and value store."""

    def __init__(self):
        """Cleanup runner with optional dry-run behavior."""
        self.unlink = None
        self.rmdir = None
        self.ignore = None
        self.unlink_count = None
        self.unlink_failed = None
        self.rmdir_count = None
        self.rmdir_failed = None

    def configure(self, args):
        """Set up runner according to command line options."""
        self.unlink = print_filename if args.dry_run else remove_file
        self.rmdir = print_dirname if args.dry_run else remove_directory
        self.ignore = args.ignore
        self.unlink_count = 0
        self.unlink_failed = 0
        self.rmdir_count = 0
        self.rmdir_failed = 0


log = logging.getLogger(__name__)
Runner = CleanupRunner()


def normalize(path_pattern: str) -> str:
    """
    Normalize path separators in a pattern for cross-platform support.

    On Windows, both forward slash and backslash are valid path separators.
    On Unix/Posix, only forward slash is valid (backslash can be part of filename).
    """
    return path_pattern.replace(os.sep, os.altsep or os.sep)


def should_ignore(path: Path, ignore_patterns: list[str]) -> bool:
    """
    Check if a path should be ignored based on ignore patterns.

    Patterns can be:
    - Simple names like 'bar': matches any directory with that name
    - Paths like 'foo/bar': matches 'bar' directory inside 'foo' directory
      and also ignores everything inside that directory
    """
    if not ignore_patterns:
        return False

    for pattern in ignore_patterns:
        # Check if pattern has multiple components (is a path with separators)
        pattern_parts = Path(normalize(pattern)).parts
        if len(pattern_parts) > 1:
            # Pattern contains path separator - match relative path
            # Path must have at least as many parts as the pattern
            if len(path.parts) < len(pattern_parts):
                continue
            # Check if pattern matches anywhere in the path hierarchy
            for i in range(len(path.parts) - len(pattern_parts) + 1):
                path_slice = path.parts[i : i + len(pattern_parts)]
                if path_slice == pattern_parts:
                    return True
        # Simple name - match the directory name anywhere
        elif path.name == pattern:
            return True
    return False


def remove_file(fileobj):
    """Attempt to delete a file object for real."""
    log.debug('Deleting file: %s', fileobj)
    try:
        fileobj.unlink()
        Runner.unlink_count += 1
    except OSError as err:
        log.debug('File not deleted. %s', err)
        Runner.unlink_failed += 1


def remove_directory(dirobj):
    """Attempt to remove a directory object for real."""
    log.debug('Removing directory: %s', dirobj)
    try:
        dirobj.rmdir()
        Runner.rmdir_count += 1
    except OSError as err:
        log.debug('Directory not removed. %s', err)
        Runner.rmdir_failed += 1


def print_filename(fileobj):
    """Only display the file name, used with --dry-run."""
    log.debug('Would delete file: %s', fileobj)
    Runner.unlink_count += 1


def print_dirname(dirobj):
    """Only display the directory name, used with --dry-run."""
    log.debug('Would delete directory: %s', dirobj)
    Runner.rmdir_count += 1


def pyclean(args):
    """Cross-platform cleaning of Python bytecode."""
    Runner.configure(args)

    for dir_name in args.directory:
        dir_path = Path(dir_name)

        log.info('Cleaning directory %s', dir_path)
        descend_and_clean(dir_path, BYTECODE_FILES, BYTECODE_DIRS)

        for topic in args.debris:
            remove_debris_for(topic, dir_path)

        remove_freeform_targets(dir_path, args.erase, args.yes, args.dry_run)

        if args.folders:
            log.debug('Removing empty directories...')
            remove_empty_directories(dir_path)

        if args.git_clean:
            execute_git_clean(dir_path, args)

    log.info(
        'Total %d files, %d directories %s.',
        Runner.unlink_count,
        Runner.rmdir_count,
        'would be removed' if args.dry_run else 'removed',
    )

    if Runner.unlink_failed or Runner.rmdir_failed:
        log.debug(
            '%d files, %d directories %s not be removed.',
            Runner.unlink_failed,
            Runner.rmdir_failed,
            'would' if args.dry_run else 'could',
        )

    # Suggest --debris option if it wasn't used
    if not args.debris:
        suggest_debris_option(args)


def descend_and_clean(directory, file_types, dir_names):
    """
    Walk and descend a directory tree, cleaning up files of a certain type
    along the way. Only delete directories if they are empty, in the end.
    """
    for child in sorted(directory.iterdir()):
        if child.is_file():
            if child.suffix in file_types:
                Runner.unlink(child)
        elif child.is_dir():
            if should_ignore(child, Runner.ignore):
                log.debug('Skipping %s', child)
            else:
                descend_and_clean(child, file_types, dir_names)

            if child.name in dir_names:
                Runner.rmdir(child)
        else:
            log.debug('Ignoring %s (neither a file nor a folder)', child)


def remove_debris_for(topic, directory):
    """
    Clean up debris for a specific topic.
    """
    log.debug('Scanning for debris of %s ...', topic.title())

    patterns = DEBRIS_TOPICS[topic]
    recursive_delete_debris(directory, patterns)


def remove_empty_directories(directory):
    """
    Recursively remove empty directories in the given directory tree.

    This walks the directory tree in post-order (bottom-up), attempting to
    remove directories that are empty.
    """
    try:
        subdirs = [
            Path(entry.path) for entry in os.scandir(directory) if entry.is_dir()
        ]
    except (OSError, PermissionError) as err:
        log.warning('Cannot access directory %s: %s', directory, err)
        return

    for subdir in subdirs:
        if should_ignore(subdir, Runner.ignore):
            log.debug('Skipping %s', subdir)
        else:
            remove_empty_directories(subdir)  # recurse down the hierarchy
            try:
                if next(subdir.iterdir(), None) is None:
                    Runner.rmdir(subdir)
            except (OSError, PermissionError) as err:
                log.debug('Cannot check or remove directory %s: %s', subdir, err)


def remove_freeform_targets(directory, glob_patterns, yes, dry_run=False):
    """
    Remove free-form targets using globbing.

    This is **potentially dangerous** since users can delete everything
    anywhere in their file system, including the entire project they're
    working on. For this reason, the implementation imposes the following
    (user experience-related) restrictions:

    - Deleting (directories) is not recursive, directory contents must be
      explicitly specified using globbing (e.g. ``dirname/**/*``).
    - The user is responsible for the deletion order, so that a directory
      is empty when it is attempted to be deleted.
    - A confirmation prompt for the deletion of every single file system
      object is shown (unless the ``--yes`` option is used, in addition).
    """
    for path_glob in glob_patterns:
        log.debug('Erase file system objects matching: %s', path_glob)
        delete_filesystem_objects(directory, path_glob, prompt=not yes, dry_run=dry_run)


def recursive_delete_debris(directory, patterns):
    """
    Recursively delete debris matching any of the given patterns.

    This function walks the directory tree once and applies all patterns
    at each level, avoiding redundant directory scans.
    """
    for pattern in patterns:
        delete_filesystem_objects(directory, pattern)

    try:
        subdirs = (
            Path(entry.path) for entry in os.scandir(directory) if entry.is_dir()

The variable entry does not seem to be defined.

        )
    except (OSError, PermissionError) as err:
        log.warning('Cannot access directory %s: %s', directory, err)
        return

    for subdir in subdirs:
        if should_ignore(subdir, Runner.ignore):
            log.debug('Skipping %s', subdir)
        else:
            recursive_delete_debris(subdir, patterns)


def delete_filesystem_objects(directory, path_glob, prompt=False, dry_run=False):
    """
    Identifies all pathnames matching a specific glob pattern, and attempts
    to delete them in the proper order, optionally asking for confirmation.

    Implementation Note: We sort the file system objects in *reverse order*
    and first delete *all files* before removing directories. This way we
    make sure that the directories that are deepest down in the hierarchy
    are empty (for both files & directories) when we attempt to remove them.
    """
    all_names = sorted(directory.glob(path_glob), reverse=True)
    dirs = (name for name in all_names if name.is_dir() and not name.is_symlink())

The variable name does not seem to be defined.

    files = (name for name in all_names if not name.is_dir() or name.is_symlink())

    for file_object in files:
        file_type = 'symlink' if file_object.is_symlink() else 'file'
        if (
            not dry_run
            and prompt
            and not confirm('Delete %s %s' % (file_type, file_object))
        ):
            Runner.unlink_failed += 1
            continue
        Runner.unlink(file_object)

    for dir_object in dirs:
        if (
            not dry_run
            and prompt
            and not confirm('Remove empty directory %s' % dir_object)
        ):
            Runner.rmdir_failed += 1
            continue
        Runner.rmdir(dir_object)


def confirm(message):
    """An interactive confirmation prompt."""
    try:
        answer = input('%s? ' % message)
        return answer.strip().lower() in ['y', 'yes']
    except KeyboardInterrupt:
        msg = 'Aborted by user.'
        raise SystemExit(msg)


def detect_debris_in_directory(directory):
    """
    Scan a directory for debris artifacts and return a list of detected topics.
    """
    detected_topics = []

    for topic, patterns in DEBRIS_TOPICS.items():
        for pattern in patterns:
            # Skip patterns that are for recursive cleanup (contain **)
            if '**' in pattern:
                continue
            # Check if the pattern matches anything in the directory
            matches = list(directory.glob(pattern))
            if matches:
                detected_topics.append(topic)
                break  # Found at least one match for this topic, move to next

    return detected_topics


def suggest_debris_option(args):
    """
    Suggest using the --debris option when it wasn't used.
    Optionally provide targeted suggestions based on detected artifacts.
    """
    # Collect all detected debris topics across all directories
    all_detected = set()
    for dir_name in args.directory:
        dir_path = Path(dir_name)
        if dir_path.exists():
            detected = detect_debris_in_directory(dir_path)
            all_detected.update(detected)

    if all_detected:
        # Provide targeted suggestion
        topics_str = ' '.join(sorted(all_detected))
        log.info(
            'Hint: Use --debris to also clean up build artifacts. Detected: %s',
            topics_str,
        )
    else:
        # Provide general suggestion
        log.info(
            'Hint: Use --debris to also clean up build artifacts '
            'from common Python development tools.',
        )


def build_git_clean_command(ignore_patterns, dry_run=False, force=False):
    """
    Build the git clean command with appropriate flags.

    Args:
        ignore_patterns: List of patterns to exclude from git clean
        dry_run: If True, only show what would be deleted
        force: If True, delete without prompting (otherwise interactive)

    Returns:
        List of command arguments
    """
    cmd = ['git', 'clean', '-dx']

    for pattern in ignore_patterns:
        cmd.extend(['-e', pattern])

    if dry_run:
        cmd.append('-n')
    elif force:
        cmd.append('-f')
    else:
        cmd.append('-i')

    return cmd


def execute_git_clean(directory, args):
    """
    Execute git clean in the specified directory.

    Args:
        directory: Path to the directory to clean
        args: Command line arguments with ignore, dry_run, yes flags
    """
    cmd = build_git_clean_command(
        args.ignore,
        dry_run=args.dry_run,
        force=args.yes
    )

    log.debug('Running: %s in %s', ' '.join(cmd), directory)

    result = subprocess.run(
        cmd,
        cwd=directory,
        capture_output=True,
        text=True,
        check=False,
    )

    if result.stdout:
        log.info(result.stdout.rstrip())
    if result.stderr:
        log.warning(result.stderr.rstrip())

    if result.returncode == 128:
        log.warning(
            'Directory %s is not under version control. Skipping git clean.',
            directory,
        )
    elif result.returncode != 0:
        raise SystemExit(result.returncode)


def run_git_clean(directory, ignore_patterns, dry_run=False, force=False):
    """
    Run git clean in the specified directory with appropriate flags.

    Args:
        directory: Path to the directory to clean
        ignore_patterns: List of patterns to exclude from git clean
        dry_run: If True, only show what would be deleted
        force: If True, delete without prompting (otherwise interactive)

    Returns:
        Exit code from git clean
    """
    cmd = build_git_clean_command(ignore_patterns, dry_run, force)

    log.debug('Running: %s in %s', ' '.join(cmd), directory)

    result = subprocess.run(
        cmd,
        cwd=directory,
        capture_output=True,
        text=True,
        check=False,
    )

    if result.stdout:
        log.info(result.stdout.rstrip())
    if result.stderr:
        log.warning(result.stderr.rstrip())

    return result.returncode