Initial commit: CloudRestoreAS v1.0.0 - Aplicación completa de restauración automática SQL Server
This commit is contained in:
218
app/extract/seven_zip.py
Normal file
218
app/extract/seven_zip.py
Normal file
@@ -0,0 +1,218 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user