qq_lib.clear

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