"""Gestión de extracción de archivos con 7-Zip.""" import subprocess from pathlib import Path from typing import Optional, Tuple import time from ..utils.logger import app_logger class SevenZipExtractor: """Gestor de extracción con 7-Zip.""" def __init__(self, seven_zip_path: Optional[str] = None): """ Inicializa el extractor. Args: seven_zip_path: Ruta a 7z.exe (se auto-detecta si es None) """ self.seven_zip_path = seven_zip_path or self._auto_detect_7zip() if not self.seven_zip_path or not Path(self.seven_zip_path).exists(): raise FileNotFoundError( "No se encontró 7-Zip. Instálalo o especifica la ruta en la configuración." ) @staticmethod def _auto_detect_7zip() -> Optional[str]: """Auto-detecta la ubicación de 7z.exe.""" possible_paths = [ r"C:\Program Files\7-Zip\7z.exe", r"D:\Program Files\7-Zip\7z.exe", r"C:\Program Files (x86)\7-Zip\7z.exe", r"D:\Program Files (x86)\7-Zip\7z.exe" ] for path in possible_paths: if Path(path).exists(): app_logger.info(f"7-Zip auto-detectado en: {path}") return path return None def extract( self, zip_path: str, output_dir: str, timeout_minutes: int = 30 ) -> Tuple[int, str, str]: """ Extrae un archivo ZIP (incluyendo multipart) usando 7-Zip. Para archivos multipart (.zip.001), 7-Zip detecta automáticamente las demás partes si están en la misma carpeta. Args: zip_path: Ruta al archivo ZIP (o .zip.001 para multipart) output_dir: Directorio de salida timeout_minutes: Timeout en minutos Returns: Tupla (exit_code, stdout, stderr) Raises: subprocess.TimeoutExpired: Si se excede el timeout FileNotFoundError: Si no existe el archivo ZIP """ zip_path_obj = Path(zip_path) if not zip_path_obj.exists(): raise FileNotFoundError(f"Archivo ZIP no encontrado: {zip_path}") # Crear directorio de salida output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) # Comando 7-Zip: extraer con rutas completas, sobrescribir sin preguntar cmd = [ str(self.seven_zip_path), "x", # Extract with full paths f"-o{output_dir}", # Output directory "-y", # Answer yes to all prompts str(zip_path) ] app_logger.info(f"Ejecutando extracción: {' '.join(cmd)}") start_time = time.time() try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=timeout_minutes * 60, encoding='utf-8', errors='replace' ) elapsed = time.time() - start_time app_logger.info( f"Extracción completada en {elapsed:.2f}s - Exit code: {result.returncode}" ) return result.returncode, result.stdout, result.stderr except subprocess.TimeoutExpired as e: elapsed = time.time() - start_time app_logger.error(f"Timeout de extracción después de {elapsed:.2f}s") raise def list_contents(self, zip_path: str) -> Tuple[int, str, str]: """ Lista el contenido de un archivo ZIP. Args: zip_path: Ruta al archivo ZIP Returns: Tupla (exit_code, stdout, stderr) """ cmd = [ str(self.seven_zip_path), "l", # List contents str(zip_path) ] result = subprocess.run( cmd, capture_output=True, text=True, encoding='utf-8', errors='replace' ) return result.returncode, result.stdout, result.stderr @staticmethod def find_bak_file(extract_dir: str) -> Optional[str]: """ Busca el archivo .bak extraído en el directorio. Args: extract_dir: Directorio donde se extrajo Returns: Ruta al archivo .bak o None si no se encuentra Raises: ValueError: Si hay múltiples archivos .bak """ extract_path = Path(extract_dir) bak_files = list(extract_path.rglob("*.bak")) if not bak_files: app_logger.error(f"No se encontró archivo .bak en {extract_dir}") return None if len(bak_files) > 1: # Si hay múltiples, tomar el más reciente app_logger.warning( f"Múltiples archivos .bak encontrados ({len(bak_files)}), " "seleccionando el más reciente" ) bak_files.sort(key=lambda p: p.stat().st_mtime, reverse=True) bak_path = str(bak_files[0]) app_logger.info(f"Archivo .bak encontrado: {bak_path}") return bak_path @staticmethod def is_multipart(zip_path: str) -> bool: """ Verifica si un archivo es parte de un ZIP multipart. Args: zip_path: Ruta al archivo Returns: True si es multipart (.zip.001, .zip.002, etc.) """ path = Path(zip_path) # Buscar patrón .zip.NNN if path.suffix.lower() in ['.zip']: # Verificar si hay un .001 al final del stem if path.stem.endswith('.001') or path.stem.endswith('.002'): return True # Verificar extensiones como .zip.001 name_lower = path.name.lower() return '.zip.' in name_lower and any( name_lower.endswith(f'.zip.{i:03d}') for i in range(1, 100) ) @staticmethod def get_first_part(zip_path: str) -> str: """ Si es un archivo multipart, retorna la ruta al primer archivo (.001). Args: zip_path: Ruta a cualquier parte del multipart Returns: Ruta al primer archivo (.001) """ path = Path(zip_path) name_lower = path.name.lower() # Si ya es .001, retornar tal cual if name_lower.endswith('.zip.001'): return str(path) # Si es otra parte, buscar el .001 if '.zip.' in name_lower: base_name = name_lower.split('.zip.')[0] first_part = path.parent / f"{base_name}.zip.001" if first_part.exists(): return str(first_part) return str(path)