Repository: apache/airflow
File: airflow-core/src/airflow/dag_processing/manager.py (Python)

Code before:
                    changed_recently.add(file)
            except FileNotFoundError:
                self.log.warning("Skipping processing of missing file: %s", file)
                self._file_stats.pop(file, None)
                continue
        file_infos = [info for info, ts in sorted(files_with_mtime.items(), key=itemgetter(1), reverse=True)]
        return file_infos, changed_recently

    def processed_recently(self, now, file):
        last_time = self._file_stats[file].last_finish_time
        if not last_time:
            return False
        elapsed_ss = (now - last_time).total_seconds()
        if elapsed_ss < self._file_process_interval:
            return True
        return False

    def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
        """
        Scan dags dir to generate more file paths to process.

        Note this method is only called when the file path queue is empty
        """
        # We only emit metrics after processing all files in the queue. If `self._parsing_start_time` is None
        # when this method is called, no files have yet been added to the queue so we shouldn't emit metrics.
        if self._parsing_start_time:
            emit_metrics(
                parse_time=time.perf_counter() - self._parsing_start_time,
                stats=list(self._file_stats.values()),
            )
            self._parsing_start_time = None

        # If the file path is already being processed, or if a file was
        # processed recently, wait until the next batch
        in_progress = set(self._processors)
        now = timezone.utcnow()

        # Sort the file paths by the parsing order mode
        recently_processed = set()
        files = []

        for bundle_files in known_files.values():
            for file in bundle_files:
                files.append(file)
                if self.processed_recently(now, file):
                    recently_processed.add(file)

        changed_recently: set[DagFileInfo] = set()
        if self._file_parsing_sort_mode == "modified_time":
            files, changed_recently = self._sort_by_mtime(files=files)
        elif self._file_parsing_sort_mode == "alphabetical":

Review comment (suggestion):
```suggestion
        if self._parsing_start_time is not None:
```

to not have to worry about 0.0 and such

Code after:
                continue
        file_infos = [info for info, ts in sorted(files_with_mtime.items(), key=itemgetter(1), reverse=True)]
        return file_infos, changed_recently

    def processed_recently(self, now, file):
        last_time = self._file_stats[file].last_finish_time
        if not last_time:
            return False
        elapsed_ss = (now - last_time).total_seconds()
        if elapsed_ss < self._file_process_interval:
            return True
        return False

    def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
        """
        Scan dags dir to generate more file paths to process.

        Note this method is only called when the file path queue is empty
        """
        # We only emit metrics after processing all files in the queue. If `self._parsing_start_time` is None
        # when this method is called, no files have yet been added to the queue so we shouldn't emit metrics.
        if self._parsing_start_time is not None:
            emit_metrics(
                parse_time=time.perf_counter() - self._parsing_start_time,
                stats=list(self._file_stats.values()),
            )
            self._parsing_start_time = None

        # If the file path is already being processed, or if a file was
        # processed recently, wait until the next batch
        in_progress = set(self._processors)
        now = timezone.utcnow()

        # Sort the file paths by the parsing order mode
        recently_processed = set()
        files = []

        for bundle_files in known_files.values():
            for file in bundle_files:
                files.append(file)
                if self.processed_recently(now, file):
                    recently_processed.add(file)

        changed_recently: set[DagFileInfo] = set()
        if self._file_parsing_sort_mode == "modified_time":
            files, changed_recently = self._sort_by_mtime(files=files)
        elif self._file_parsing_sort_mode == "alphabetical":
            files.sort(key=attrgetter("rel_path"))
        elif self._file_parsing_sort_mode == "random_seeded_by_host":
            # Shuffle the list seeded by hostname so multiple DAG processors can work on different
            # set of files. Since we set the seed, the sort order will remain same per host