descarga_cpe_sovos/smb_transfer.py

import os
import sys
import subprocess
from pathlib import Path
from dotenv import load_dotenv
from logger_manager import get_logger

# ============================================================================
load_dotenv()

logger = get_logger("smb")

# ============================================================================
def get_mount_base() -> str:
    """Base de montaje SMB según la plataforma (.env)."""
    if sys.platform == "darwin":
        return os.getenv("SMB_MOUNT_MAC", "/Volumes/FileSAP")
    return os.getenv("SMB_MOUNT_LINUX", "/mnt/filesap")


def _creds() -> dict:
    return {
        "server": os.getenv("SMB_SERVER", ""),
        "share": os.getenv("SMB_SHARE", ""),
        "user": os.getenv("SMB_USER", ""),
        "password": os.getenv("SMB_PASSWORD", ""),
    }

# ============================================================================
def detectar_montaje(mount_base: str) -> bool:
    """Verifica que el punto de montaje exista y esté montado de verdad."""
    p = Path(mount_base)
    if not p.is_dir():
        return False
    if os.path.ismount(mount_base):
        return True
    try:
        return any(p.iterdir())
    except PermissionError:
        return False

# ============================================================================
def convertir_unc_a_local(ruta_unc: str, mount_base: str | None = None) -> str | None:
    """Convierte una ruta UNC (\\servidor\\recurso\\carpeta\\archivo.pdf) a ruta
    local sobre el montaje SMB actual."""
    ruta_unc = (ruta_unc or "").strip()
    if len(ruta_unc) < 5:
        return None
    base = mount_base or get_mount_base()
    if not base:
        return None
    path = ruta_unc.replace("\\", "/").lstrip("/")
    partes = path.split("/")
    if len(partes) < 2:
        return None
    ruta_relativa = "/".join(partes[2:])
    return os.path.join(base, ruta_relativa)

# ============================================================================
def montar_smb(mount_base: str) -> tuple[bool, str]:
    """Intenta montar la unidad SMB sin sudo (usuario con privilegios)."""
    creds = _creds()
    if not creds["server"] or not creds["share"]:
        return False, "Faltan SMB_SERVER/SMB_SHARE en .env."

    Path(mount_base).mkdir(parents=True, exist_ok=True)

    try:
        if sys.platform == "darwin":
            url = f"//{creds['user']}:{creds['password']}@{creds['server']}/{creds['share']}"
            result = subprocess.run(
                ["mount_smbfs", url, mount_base],
                capture_output=True, text=True,
            )
        else:
            opts = f"username={creds['user']},password={creds['password']}"
            result = subprocess.run(
                ["mount", "-t", "cifs", f"//{creds['server']}/{creds['share']}", mount_base, "-o", opts],
                capture_output=True, text=True,
            )
    except FileNotFoundError:
        return False, "Comando de montaje no disponible en el sistema."

    if result.returncode != 0:
        return False, result.stderr.strip() or "Error montando SMB (código %d)." % result.returncode

    return True, f"Unidad montada en {mount_base}."

# ============================================================================
def _rsync_cmd(destino: str) -> list[str]:
    return [
        "rsync", "-rtv", "--progress",
        "--remove-source-files",
        "--include=*.pdf",
        "--include=*.xml",
        "--exclude=*",
        "./",
        f"{destino}/",
    ]

# ============================================================================
def transferir_empresa(empresa: dict, mount_base: str, auto_mount: bool = False) -> dict:
    """Transfiere downloads/<nombre>/ hacia <mount_base>/<smb_subfolder>/ con rsync.

    Si el SMB no está montado: si auto_mount está habilitado se intenta montar;
    si falla (o no está habilitado), se conservan los archivos localmente.
    """
    nombre = empresa["nombre"]
    origen = Path("downloads") / nombre

    if not origen.is_dir():
        return {"empresa": nombre, "estado": "sin_descargas", "movidos": 0}

    if not detectar_montaje(mount_base):
        if auto_mount:
            ok, msg = montar_smb(mount_base)
            if ok:
                logger.info(f"   🔌 {msg}")
            else:
                logger.warning(f"   ⚠️  No se pudo montar SMB ({mount_base}): {msg}. "
                               f"Archivos conservados en {origen}.")
                return {"empresa": nombre, "estado": "smb_no_montado", "movidos": 0}
        else:
            logger.warning(f"   ⚠️  Empresa '{nombre}': SMB NO montado ({mount_base}). "
                           f"Transferencia omitida, archivos conservados en {origen}.")
            return {"empresa": nombre, "estado": "smb_no_montado", "movidos": 0}

    destino = Path(mount_base) / empresa["smb_subfolder"]
    try:
        destino.mkdir(parents=True, exist_ok=True)
    except Exception as e:
        logger.warning(f"   ⚠️  Empresa '{nombre}': no se pudo crear {destino}: {e}. "
                       f"Transferencia omitida, archivos conservados.")
        return {"empresa": nombre, "estado": "error_destino", "movidos": 0}

    logger.info(f"   🔁 Empresa '{nombre}': rsync {origen}/ -> {destino}/")
    try:
        result = subprocess.run(
            _rsync_cmd(str(destino)),
            cwd=str(origen),
            capture_output=True,
            text=True,
        )
    except FileNotFoundError:
        logger.error("   ❌ rsync no está instalado en el sistema.")
        return {"empresa": nombre, "estado": "rsync_faltante", "movidos": 0}

    if result.returncode != 0:
        logger.error(f"   ❌ Error rsync ({nombre}):\n{result.stderr.strip()}")
        return {"empresa": nombre, "estado": "error_rsync", "movidos": 0}

    restantes = [f for f in origen.iterdir() if f.is_file()]
    movidos = len(restantes) == 0
    estado = "ok" if movidos else "parcial"
    logger.info(f"      ✅ Transferencia {estado} ({nombre}). Quedan {len(restantes)} archivos locales.")
    return {"empresa": nombre, "estado": estado, "movidos": movidos}

# ============================================================================
def transferir_a_smb(empresas: list[dict], mount_base: str, auto_mount: bool = False) -> list[dict]:
    resultados = []
    for emp in empresas:
        if emp.get("locked", False):
            logger.info(f"   ⏭️  Empresa '{emp['nombre']}' bloqueada (LOCKED), omitiendo transferencia.")
            continue
        resultados.append(transferir_empresa(emp, mount_base, auto_mount))
    return resultados

# ============================================================================
if __name__ == "__main__":
    import json

    base = get_mount_base()
    if "--mount" in sys.argv:
        ok, msg = montar_smb(base)
        print(f"{'✅' if ok else '❌'} {msg}")
    else:
        cfg = json.load(open(Path(__file__).parent / "config.json", encoding="utf-8"))
        activas = [e for e in cfg.get("empresas", []) if not e.get("locked", False)]
        auto = os.getenv("SMB_AUTO_MOUNT", "false").lower() == "true"
        for r in transferir_a_smb(activas, base, auto):
            print(r)
# ============================================================================