""" Everything API Wrapper Integración con Everything para búsqueda rápida de archivos .s7p """ import os import ctypes from pathlib import Path from typing import List, Optional, Tuple import logging class EverythingSearcher: """Wrapper para la API de Everything""" def __init__(self, dll_path: str = None): self.logger = logging.getLogger(__name__) self.dll = None self.is_initialized = False if dll_path: self.load_dll(dll_path) def load_dll(self, dll_path: str) -> bool: """Cargar la DLL de Everything""" try: if not os.path.exists(dll_path): self.logger.error(f"DLL de Everything no encontrada: {dll_path}") return False self.dll = ctypes.WinDLL(dll_path) # Definir funciones de la API self._setup_api_functions() # Inicializar Everything self.is_initialized = self._initialize_everything() if self.is_initialized: self.logger.info("Everything API inicializada correctamente") else: self.logger.error("Error inicializando Everything API") return self.is_initialized except Exception as e: self.logger.error(f"Error cargando DLL de Everything: {e}") return False def _setup_api_functions(self): """Configurar las funciones de la API""" try: # Everything_SetSearchW - Establecer consulta de búsqueda self.dll.Everything_SetSearchW.argtypes = [ctypes.c_wchar_p] self.dll.Everything_SetSearchW.restype = None # Everything_SetMatchPath - Habilitar coincidencia de ruta completa self.dll.Everything_SetMatchPath.argtypes = [ctypes.c_bool] self.dll.Everything_SetMatchPath.restype = None # Everything_SetMatchCase - Habilitar coincidencia de mayúsculas/minúsculas self.dll.Everything_SetMatchCase.argtypes = [ctypes.c_bool] self.dll.Everything_SetMatchCase.restype = None # Everything_QueryW - Ejecutar consulta self.dll.Everything_QueryW.argtypes = [ctypes.c_bool] self.dll.Everything_QueryW.restype = ctypes.c_bool # Everything_GetNumResults - Obtener número de resultados self.dll.Everything_GetNumResults.restype = ctypes.c_uint # Everything_GetResultFullPathNameW - Obtener ruta completa del resultado self.dll.Everything_GetResultFullPathNameW.argtypes = [ ctypes.c_uint, ctypes.c_wchar_p, ctypes.c_uint ] self.dll.Everything_GetResultFullPathNameW.restype = ctypes.c_uint # Everything_GetLastError - Obtener último error self.dll.Everything_GetLastError.restype = ctypes.c_uint self.logger.debug("Funciones API configuradas correctamente") except Exception as e: self.logger.error(f"Error configurando funciones API: {e}") raise def _initialize_everything(self) -> bool: """Inicializar Everything con configuraciones por defecto""" try: # Configurar opciones de búsqueda self.dll.Everything_SetMatchPath(True) self.dll.Everything_SetMatchCase(False) return True except Exception as e: self.logger.error(f"Error inicializando Everything: {e}") return False def search_s7p_files(self, search_directories: List[str]) -> List[str]: """ Buscar archivos .s7p en los directorios especificados Retorna lista de rutas completas a archivos .s7p encontrados """ if not self.is_initialized: self.logger.error("Everything no está inicializado") return [] try: all_s7p_files = [] for directory in search_directories: s7p_files = self._search_s7p_in_directory(directory) all_s7p_files.extend(s7p_files) # Eliminar duplicados manteniendo el orden unique_files = [] seen = set() for file_path in all_s7p_files: if file_path not in seen: seen.add(file_path) unique_files.append(file_path) self.logger.info(f"Encontrados {len(unique_files)} archivos .s7p únicos") return unique_files except Exception as e: self.logger.error(f"Error buscando archivos S7P: {e}") return [] def _search_s7p_in_directory(self, directory: str) -> List[str]: """Buscar archivos .s7p en un directorio específico""" try: # Normalizar la ruta del directorio directory = os.path.normpath(directory) # Construir consulta de búsqueda # Buscar archivos .s7p en el directorio y subdirectorios search_query = f'"{directory}" ext:s7p' self.logger.debug(f"Búsqueda en Everything: {search_query}") # Establecer consulta self.dll.Everything_SetSearchW(search_query) # Ejecutar búsqueda if not self.dll.Everything_QueryW(True): error_code = self.dll.Everything_GetLastError() self.logger.error(f"Error en consulta Everything: código {error_code}") return [] # Obtener número de resultados num_results = self.dll.Everything_GetNumResults() self.logger.debug(f"Everything encontró {num_results} resultados para {directory}") if num_results == 0: return [] # Extraer rutas de los resultados results = [] buffer_size = 4096 # Tamaño del buffer para rutas for i in range(num_results): # Buffer para almacenar la ruta path_buffer = ctypes.create_unicode_buffer(buffer_size) # Obtener ruta completa chars_copied = self.dll.Everything_GetResultFullPathNameW( i, path_buffer, buffer_size ) if chars_copied > 0: file_path = path_buffer.value if file_path and os.path.exists(file_path): results.append(file_path) self.logger.debug(f"Archivo S7P encontrado: {file_path}") return results except Exception as e: self.logger.error(f"Error buscando en directorio {directory}: {e}") return [] def search_files_by_pattern(self, pattern: str, max_results: int = 1000) -> List[str]: """ Buscar archivos usando un patrón específico """ if not self.is_initialized: self.logger.error("Everything no está inicializado") return [] try: self.logger.debug(f"Búsqueda por patrón: {pattern}") # Establecer consulta self.dll.Everything_SetSearchW(pattern) # Ejecutar búsqueda if not self.dll.Everything_QueryW(True): error_code = self.dll.Everything_GetLastError() self.logger.error(f"Error en consulta Everything: código {error_code}") return [] # Obtener número de resultados num_results = self.dll.Everything_GetNumResults() actual_results = min(num_results, max_results) self.logger.debug(f"Procesando {actual_results} de {num_results} resultados") results = [] buffer_size = 4096 for i in range(actual_results): path_buffer = ctypes.create_unicode_buffer(buffer_size) chars_copied = self.dll.Everything_GetResultFullPathNameW( i, path_buffer, buffer_size ) if chars_copied > 0: file_path = path_buffer.value if file_path and os.path.exists(file_path): results.append(file_path) return results except Exception as e: self.logger.error(f"Error buscando patrón {pattern}: {e}") return [] def is_everything_available(self) -> bool: """Verificar si Everything está disponible y funcionando""" return self.is_initialized def get_project_directory_from_s7p(self, s7p_file_path: str, observation_directory: str) -> Tuple[str, str]: """ Determinar el directorio del proyecto y la ruta relativa desde un archivo .s7p Retorna: (directorio_proyecto, ruta_relativa) """ try: s7p_path = Path(s7p_file_path) obs_path = Path(observation_directory) # El directorio del proyecto es el padre del archivo .s7p project_dir = s7p_path.parent # Calcular ruta relativa desde el directorio de observación try: relative_path = project_dir.relative_to(obs_path) return str(project_dir), str(relative_path) except ValueError: # Si no se puede calcular la ruta relativa, usar ruta absoluta self.logger.warning(f"No se puede calcular ruta relativa para {s7p_file_path}") return str(project_dir), str(project_dir) except Exception as e: self.logger.error(f"Error determinando directorio de proyecto para {s7p_file_path}: {e}") return "", "" # Funciones de utilidad para crear instancias def create_everything_searcher(dll_path: str) -> Optional[EverythingSearcher]: """Crear una instancia de EverythingSearcher con manejo de errores""" try: searcher = EverythingSearcher(dll_path) if searcher.is_everything_available(): return searcher else: return None except Exception as e: logging.getLogger(__name__).error(f"Error creando EverythingSearcher: {e}") return None