feature/integracion-cpanel-asrecovery
This commit is contained in:
@@ -1,9 +1,13 @@
|
||||
"""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
|
||||
|
||||
|
||||
@@ -25,19 +29,30 @@ class SevenZipExtractor:
|
||||
|
||||
@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
|
||||
|
||||
"""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(
|
||||
@@ -134,35 +149,78 @@ class SevenZipExtractor:
|
||||
return result.returncode, result.stdout, result.stderr
|
||||
|
||||
@staticmethod
|
||||
def find_bak_file(extract_dir: str) -> Optional[str]:
|
||||
def find_bak_file(extract_dir: str, node_name: Optional[str] = None) -> Optional[str]:
|
||||
"""
|
||||
Busca el archivo .bak extraído en el directorio.
|
||||
|
||||
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 o None si no se encuentra
|
||||
|
||||
Raises:
|
||||
ValueError: Si hay múltiples archivos .bak
|
||||
Ruta al archivo .bak (ya renombrado si node_name) o None si no hay archivos.
|
||||
"""
|
||||
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}")
|
||||
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
|
||||
|
||||
if len(bak_files) > 1:
|
||||
# Si hay múltiples, tomar el más reciente
|
||||
|
||||
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(
|
||||
f"Múltiples archivos .bak encontrados ({len(bak_files)}), "
|
||||
"seleccionando el más reciente"
|
||||
"No se encontró archivo con extensión .bak; usando el archivo más "
|
||||
f"grande de la extracción: {largest.name}"
|
||||
)
|
||||
bak_files.sort(key=lambda p: p.stat().st_mtime, reverse=True)
|
||||
|
||||
bak_path = str(bak_files[0])
|
||||
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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user