277 lines
9.5 KiB
Python
277 lines
9.5 KiB
Python
"""Gestión de extracción de archivos con 7-Zip."""
|
|
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple
|
|
import time
|
|
|
|
from ..constants import BUNDLED_7ZIP_EXE, BUNDLED_7ZIP_FALLBACK, IS_WINDOWS
|
|
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]:
|
|
"""Prioriza 7-Zip embebido; en desarrollo busca en el sistema."""
|
|
for candidate in (BUNDLED_7ZIP_EXE, BUNDLED_7ZIP_FALLBACK):
|
|
if candidate.exists():
|
|
app_logger.info(f"7-Zip embebido: {candidate}")
|
|
return str(candidate)
|
|
|
|
if not getattr(sys, "frozen", False):
|
|
if IS_WINDOWS:
|
|
possible_paths = [
|
|
r"C:\Program Files\7-Zip\7z.exe",
|
|
r"C:\Program Files (x86)\7-Zip\7z.exe",
|
|
]
|
|
else:
|
|
possible_paths = []
|
|
for name in ("7z", "7zz", "7za"):
|
|
found = shutil.which(name)
|
|
if found:
|
|
possible_paths.append(found)
|
|
|
|
for path in possible_paths:
|
|
if Path(path).exists():
|
|
app_logger.info(f"7-Zip 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, node_name: Optional[str] = None) -> Optional[str]:
|
|
"""
|
|
Busca el backup extraído y, si se indica node_name, lo normaliza a
|
|
'<node_name>.bak'.
|
|
|
|
Tolera terminaciones extra en el nombre (el backup puede venir como
|
|
'nodo.KNOWNWORLD.bak', 'nodo.bak.unknownworld' o incluso sin extensión
|
|
'.bak'). Estrategia de búsqueda en orden:
|
|
1. archivos que terminan exactamente en '.bak'
|
|
2. si no hay, archivos cuyo nombre contiene '.bak' en cualquier parte
|
|
3. si no hay, el archivo más grande de la extracción (el backup suele serlo)
|
|
|
|
Args:
|
|
extract_dir: Directorio donde se extrajo
|
|
node_name: Nombre del nodo; si se indica, el backup se renombra a
|
|
'<node_name>.bak'
|
|
|
|
Returns:
|
|
Ruta al archivo .bak (ya renombrado si node_name) o None si no hay archivos.
|
|
"""
|
|
extract_path = Path(extract_dir)
|
|
all_files = [p for p in extract_path.rglob("*") if p.is_file()]
|
|
|
|
if not all_files:
|
|
app_logger.error(f"No se encontró ningún archivo en {extract_dir}")
|
|
return None
|
|
|
|
bak_files = [p for p in all_files if p.suffix.lower() == ".bak"]
|
|
if not bak_files:
|
|
bak_files = [p for p in all_files if ".bak" in p.name.lower()]
|
|
if not bak_files:
|
|
largest = max(all_files, key=lambda p: p.stat().st_size)
|
|
app_logger.warning(
|
|
"No se encontró archivo con extensión .bak; usando el archivo más "
|
|
f"grande de la extracción: {largest.name}"
|
|
)
|
|
bak_files = [largest]
|
|
|
|
if len(bak_files) > 1:
|
|
# Preferir el que empiece con el nombre del nodo; si no, el más reciente.
|
|
if node_name:
|
|
prefixed = [
|
|
p for p in bak_files
|
|
if p.name.lower().startswith(node_name.lower())
|
|
]
|
|
if prefixed:
|
|
bak_files = prefixed
|
|
if len(bak_files) > 1:
|
|
app_logger.warning(
|
|
f"Múltiples backups encontrados ({len(bak_files)}), "
|
|
"seleccionando el más reciente"
|
|
)
|
|
bak_files.sort(key=lambda p: p.stat().st_mtime, reverse=True)
|
|
|
|
bak_file = bak_files[0]
|
|
|
|
# Normalizar el nombre a '<node_name>.bak'.
|
|
if node_name:
|
|
target = bak_file.parent / f"{node_name}.bak"
|
|
if bak_file.name != target.name:
|
|
try:
|
|
if target.exists() and not bak_file.samefile(target):
|
|
target.unlink()
|
|
bak_file = bak_file.rename(target)
|
|
app_logger.info(f"Backup normalizado a: {bak_file.name}")
|
|
except OSError as e:
|
|
# No es fatal para el RESTORE: se continúa con el nombre original.
|
|
app_logger.error(
|
|
f"No se pudo renombrar el backup a '{target.name}': {e}"
|
|
)
|
|
|
|
bak_path = str(bak_file)
|
|
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)
|