qq_lib.clear

 30class Clearer:
 31    """
 32    Handles detection and removal of qq runtime files from a directory.
 33    """
 34
 35    def __init__(self, directories: list[Path]):
 36        """
 37        Initialize a Clearer for one or more directories.
 38
 39        Args:
 40            directories (list[Path]): The directories to clear qq runtime files from.
 41        """
 42        self._directories = directories
 43
 44    def clear(self, force: bool = False) -> None:
 45        """
 46        Remove all qq runtime files from all directories that are safe to be removed.
 47
 48        Directories are cleared in parallel. Only qq files that do **not**
 49        correspond to an active or successfully finished job will be removed,
 50        unless `force` is set to True. A combined summary is logged at the end.
 51
 52        Args:
 53            force (bool): If True, remove all qq runtime files, even if unsafe.
 54        """
 55        results: list[_ClearResult] = []
 56        lock = threading.Lock()
 57
 58        def clear_single(directory: Path) -> None:
 59            result = Clearer._clear_directory(directory, force)
 60            with lock:
 61                results.append(result)
 62
 63        with ThreadPoolExecutor(
 64            max_workers=CFG.parallelization_options.clear_max_threads
 65        ) as executor:
 66            for directory in self._directories:
 67                executor.submit(clear_single, directory)
 68
 69        total_deleted = sum(r.deleted for r in results)
 70        total_excluded = sum(r.excluded for r in results)
 71
 72        if total_deleted == 0 and total_excluded == 0:
 73            logger.info("Nothing to clear.")
 74            return
 75
 76        if total_deleted > 0:
 77            logger.info(
 78                f"Removed {total_deleted} qq file{'s' if total_deleted > 1 else ''}."
 79            )
 80
 81        if total_excluded > 0:
 82            logger.info(
 83                f"{total_excluded} qq file{'s' if total_excluded > 1 else ''} could not be safely cleared. "
 84                f"Rerun as '{CFG.binary_name} clear --force' to clear them forcibly."
 85            )
 86
 87    @staticmethod
 88    def _clear_directory(directory: Path, force: bool) -> _ClearResult:
 89        """
 90        Clear qq runtime files from a single directory and return the result.
 91
 92        Args:
 93            directory (Path): The directory to clear.
 94            force (bool): If True, remove all qq runtime files, even if unsafe.
 95
 96        Returns:
 97            _ClearResult: The number of deleted and excluded files.
 98        """
 99        files = Clearer._collect_runtime_files(directory)
100        logger.debug(f"All qq runtime files in '{directory}': {files}.")
101        if not files:
102            return _ClearResult()
103
104        excluded = Clearer._collect_excluded_files(directory) if not force else set()
105        logger.debug(f"Files excluded from clearing in '{directory}': {excluded}.")
106
107        to_delete = files - excluded
108        logger.debug(f"Files to delete in '{directory}': {to_delete}.")
109
110        if to_delete:
111            Clearer._delete_files(to_delete)
112
113        return _ClearResult(deleted=len(to_delete), excluded=len(excluded))
114
115    @staticmethod
116    def _collect_runtime_files(directory: Path) -> set[Path]:
117        """
118        Collect all qq runtime files in the directory.
119
120        Returns:
121            set[Path]: Paths to all files matching qq-specific suffixes.
122        """
123        return set(get_runtime_files(directory))
124
125    @staticmethod
126    def _collect_excluded_files(directory: Path) -> set[Path]:
127        """
128        Collect qq runtime files that should **not** be deleted.
129
130        Runtime files corresponding to active or successfully finished jobs are included.
131
132        Returns:
133            set[Path]: Paths to qq runtime files that should not be deleted.
134        """
135        excluded = []
136
137        # iterate through info files
138        for file in get_info_files(directory):
139            try:
140                informer = Informer.from_file(file)
141                state = informer.get_real_state()
142                logger.debug(f"Job state: {str(state)}.")
143            except QQError:
144                # ignore the file if it cannot be read
145                continue
146
147            if state not in [
148                RealState.KILLED,
149                RealState.FAILED,
150                RealState.IN_AN_INCONSISTENT_STATE,
151            ]:
152                excluded.append(file)  # qq info file
153                excluded.append(directory / informer.info.stdout_file)  # script stdout
154                excluded.append(directory / informer.info.stderr_file)  # script stderr
155                excluded.append(
156                    (directory / informer.info.job_name).with_suffix(
157                        CFG.suffixes.qq_out
158                    )
159                )  # qq out file
160
161        return set(excluded)
162
163    @staticmethod
164    def _delete_files(files: Iterable[Path]) -> None:
165        """
166        Delete all specified files.
167
168        Args:
169            files (Iterable[Path]): The list of files to delete.
170        """
171        for file in files:
172            logger.debug(f"Removing file '{file}'.")
173            file.unlink()