voici la meme recherche avec:



def scan_directory(self, directory, results, criteria_lower):



    """Scan the directory for files matching the criteria."""



    for root, dirs, files in os.walk(directory):



        for file in files:



            fullPath = os.path.join(root, file)



            if not self.extension_pattern.match(fullPath):



                continue  # Filtrer par extension



            self.process_file(fullPath, file, results, criteria_lower)



2024-11-18 13:51:41,015 - DEBUG - End of launch_delayed_search



2024-11-18 13:51:42,029 - DEBUG - Function launchSearch start



2024-11-18 13:51:42,029 - DEBUG - Search criteria: bumper, criteria_type: filename



2024-11-18 13:51:42,030 - DEBUG - Current root path: Y:/



2024-11-18 13:51:42,030 - DEBUG - WATCHED_DIR: Y:\DocSoufiane



2024-11-18 13:51:42,030 - DEBUG - Search criteria set: bumper (type: filename)



2024-11-18 13:51:42,030 - DEBUG - Date range set: 2024-11-18 - 2024-11-18



2024-11-18 13:51:42,030 - DEBUG - Base path for search set: Y:\



2024-11-18 13:51:42,031 - DEBUG - Function launchSearch end



2024-11-18 13:51:42,031 - INFO - Starting search.



2024-11-18 13:51:42,031 - DEBUG - Root path for search: Y:\



2024-11-18 13:51:42,032 - INFO - Using file system search.



2024-11-18 13:52:00,173 - INFO - Finalizing search.



2024-11-18 13:52:00,174 - INFO - Search completed with 1561 results.



et avec :



def scan_directory(self, directory, results, criteria_lower):



        """Recursively scan the directory for files matching the criteria."""



        for entry in os.scandir(directory):



            if entry.is_dir():



                self.scan_directory(entry.path, results, criteria_lower)  # Recursive call for subdirectories



            elif entry.is_file():



                self.process_file(entry.path, entry.name, results, criteria_lower)







2024-11-18 13:53:57,237 - DEBUG - Function launchSearch start



2024-11-18 13:53:57,237 - DEBUG - Search criteria: bumper, criteria_type: filename



2024-11-18 13:53:57,237 - DEBUG - Current root path: Y:/



2024-11-18 13:53:57,237 - DEBUG - WATCHED_DIR: Y:\DocSoufiane



2024-11-18 13:53:57,237 - DEBUG - Search criteria set: bumper (type: filename)



2024-11-18 13:53:57,238 - DEBUG - Date range set: 2024-11-18 - 2024-11-18



2024-11-18 13:53:57,238 - DEBUG - Base path for search set: Y:\



2024-11-18 13:53:57,238 - DEBUG - Function launchSearch end



2024-11-18 13:53:57,238 - INFO - Starting search.



2024-11-18 13:53:57,238 - DEBUG - Root path for search: Y:\



2024-11-18 13:53:57,239 - INFO - Using file system search.



2024-11-18 13:54:08,794 - INFO - Finalizing search.



2024-11-18 13:54:08,795 - INFO - Search completed with 1561 results.

En regardant attentivement les logs, on constate que :



os.walk: La recherche a pris environ 18 secondes (de 13:51:42 à 13:52:00).

os.scandir: La recherche a pris environ 11 secondes (de 13:53:57 à 13:54:08).



comment encore optimiser la rapidité de la recherche dans fichier system dans la class:
# Background search worker class
class SearchWorker(QObject):
    searchFinished = pyqtSignal(bool, list)  # Signal to indicate search success

    def __init__(self, indexer, index_dir):
        super().__init__()
        self.running = False
        self.running_mutex = QMutex()  # Mutex to protect self.running
        self.rootPath = ""
        self.criteria = ""
        self.criteria_type = "filename"
        self.start_date = None
        self.end_date = None
        self.allowed_extensions = ALLOWED_EXTENSIONS
        self.extension_pattern = re.compile(r".*(" + "|".join([re.escape(ext) + "$" for ext in self.allowed_extensions]) + ")", re.IGNORECASE)
        self.indexer = indexer  
        self.index_dir = os.path.normpath(index_dir)
        logger.debug(f"SearchWorker initialized with index_dir: {self.index_dir}")

    def stop(self):
        """Stop the ongoing search gracefully."""
        self.running_mutex.lock()
        running = self.running
        self.running_mutex.unlock()

        if not running:
            logger.info("Search interrupted by user.")
            return  # Exit if the search was not running

        logger.info("Search stopped by user.")
        self.running = False  # This will cause the search loop to exit

    def setCriteria(self, criteria, criteria_type="filename"):
        """Sets the search criteria and its type."""
        self.criteria = criteria.lower()  # Convert criteria to lowercase
        self.criteria_type = criteria_type
        logger.debug(f"Search criteria set: {criteria} (type: {criteria_type})")

    def setBasePath(self, path):
        self.rootPath = os.path.normpath(path)
        logger.debug(f"Base path for search set: {self.rootPath}")

    def setDateRange(self, start_date, end_date):
        self.start_date = start_date.toPyDate()
        self.end_date = end_date.toPyDate()
        logger.debug(f"Date range set: {self.start_date} - {self.end_date}")

        # Pre-calculate timestamps for date comparison
        self.start_date_timestamp = datetime.combine(self.start_date, datetime.min.time()).timestamp()
        self.end_date_timestamp = datetime.combine(self.end_date, datetime.max.time()).timestamp()

    def performSearch(self, retry=False):
        logger.info("Starting search.")
        logger.debug(f"Root path for search: {self.rootPath}")

        # Verify root path
        if not self.rootPath or not os.path.exists(self.rootPath):
            logger.error("Base path not set or does not exist.")
            self.searchFinished.emit(False, [])
            return

        # Validate dates
        if self.start_date and self.end_date and self.start_date > self.end_date:
            logger.error("Start date must be before end date.")
            self.searchFinished.emit(False, [])
            return

        self.running_mutex.lock()  # Lock before modifying
        self.running = True
        self.running_mutex.unlock()  # Unlock after modification

        results = []
        criteria_lower = self.criteria.lower().strip()
        if not criteria_lower:
            logger.error("Search criteria is empty after processing. Aborting search.")
            self.searchFinished.emit(False, [])
            return

        try:
            # Check if within watched directory
            normalized_root_path = os.path.normcase(os.path.normpath(self.rootPath))
            normalized_watched_dir = os.path.normcase(os.path.normpath(WATCHED_DIR))

            if normalized_root_path.startswith(normalized_watched_dir):
                logger.info("Using index-based search.")
                self.index_based_search(results, criteria_lower, retry)
            else:
                logger.info("Using file system search.")
                self.scan_directory(normalized_root_path, results, criteria_lower)

            #logger.debug(f"Final search results: {results}")
        except Exception as e:
            logger.error(f"Error during search: {e}")
            self.searchFinished.emit(False, [])
            return
        finally:
            logger.info("Finalizing search.")
            self.running_mutex.lock()
            self.running = False
            self.running_mutex.unlock()
            self.searchFinished.emit(True, results)
            logger.info(f"Search completed with {len(results)} results.")

    def index_based_search(self, results, criteria_lower, retry):
        try:
            logger.info("Performing index-based search.")
            ix = open_dir(self.index_dir)
            with ix.searcher() as searcher:
                query_parts = []

                # Build query based on criteria type
                if self.criteria_type == "filename":
                    if not criteria_lower.strip():
                        raise ValueError("Search criteria for filename is empty.")
                    query_part = Prefix("name", criteria_lower)
                    query_parts.append(query_part)
                elif self.criteria_type in ("creation_date", "modification_date"):
                    date_field = "creation_time" if self.criteria_type == "creation_date" else "modification_time"
                    query_parts.append(DateRange(date_field, self.start_date, self.end_date))

                # Filter by extensions using regex
                if self.allowed_extensions:  # Check if extensions are specified
                    query_parts.append(self.extension_pattern)

                # Filter by path
                query_parts.append(Prefix("path", os.path.normcase(os.path.normpath(self.rootPath))))
                query = And(query_parts)

                # Execute the search
                hits = searcher.search(query, limit=None)
                for hit in hits:
                    self.running_mutex.lock()  # Protect access to self.running
                    running = self.running
                    self.running_mutex.unlock()
                    if not running:
                        break
                    results.append(hit["path"])
        except EmptyIndexError:
            if not retry:
                logger.warning("Index directory is empty, creating index...")
                self.indexer.buildIndex()
                return self.performSearch(retry=True)  # Retry after rebuilding index
            else:
                logger.error("Index is still empty after rebuilding.")
                self.searchFinished.emit(False, [])
        except Exception as e:
            logger.error(f"Error during Whoosh search: {e}")
            self.searchFinished.emit(False, [])

    def scan_directory(self, directory, results, criteria_lower):
        """Recursively scan the directory for files matching the criteria."""
        for entry in os.scandir(directory):
            if entry.is_dir():
                self.scan_directory(entry.path, results, criteria_lower)  # Recursive call for subdirectories
            elif entry.is_file() :
                self.process_file(entry.path, entry.name, results, criteria_lower)

    def process_file(self, fullPath, file, results, criteria_lower):
        try:
            file_lower = file.lower()
            _, extension = os.path.splitext(fullPath)

            # Check extension using regex
            if not self.extension_pattern.match(fullPath):
                return

            # Check criteria
            if self.criteria_type == "filename" and criteria_lower in file_lower:
                results.append(fullPath)
            elif self.criteria_type in ("creation_date", "modification_date"):
                date_to_check = os.path.getctime(fullPath) if self.criteria_type == "creation_date" else os.path.getmtime(fullPath)
                if self.compare_dates(date_to_check) :
                    if criteria_lower in file_lower:
                        results.append(fullPath)
        except Exception as e:
            logger.error(f"Error processing file {fullPath}: {e}")

    def compare_dates(self, file_timestamp):
        return self.start_date_timestamp <= file_timestamp <= self.end_date_timestamp