descarga_xml_sunat/smb_manager.py

import os
import re
import platform
import subprocess
from pathlib import Path
from dotenv import load_dotenv
from logger_manager import get_logger

# ============================================================================
load_dotenv()
logger = get_logger(__name__)

SMB_SERVER = os.getenv("SMB_SERVER", "").strip()
SMB_SHARE = os.getenv("SMB_SHARE", "").strip()
SMB_USER = os.getenv("SMB_USER", "").strip()
SMB_PASSWORD = os.getenv("SMB_PASSWORD", "").strip()
SMB_SUBFOLDER = os.getenv("SMB_SUBFOLDER", "").strip()
SMB_MOUNT_MAC = os.getenv("SMB_MOUNT_MAC", "").strip()
SMB_MOUNT_LINUX = os.getenv("SMB_MOUNT_LINUX", "").strip()
SMB_MOUNT_WINDOWS = os.getenv("SMB_MOUNT_WINDOWS", "").strip()

# ============================================================================
def get_current_os():
    system = platform.system().lower()
    if system == "darwin":
        return "mac"
    if system == "windows":
        return "windows"
    return "linux"

# ============================================================================
def get_mount_point():
    current_os = get_current_os()
    mapping = {
        "mac": SMB_MOUNT_MAC,
        "linux": SMB_MOUNT_LINUX,
        "windows": SMB_MOUNT_WINDOWS,
    }
    return mapping.get(current_os, "")

# ============================================================================
def esta_montado(mount_point):
    if not mount_point:
        return False
    if get_current_os() == "windows":
        return os.path.exists(mount_point)
    return os.path.ismount(mount_point) or os.path.exists(
        os.path.join(mount_point, ".mounted_check")
    )

# ============================================================================
def montar_smb():
    mount_point = get_mount_point()
    if not mount_point:
        logger.warning("No hay punto de montaje configurado para este SO.")
        return False

    if esta_montado(mount_point):
        return True

    current_os = get_current_os()

    if current_os != "windows" and not os.path.exists(mount_point):
        os.makedirs(mount_point)

    try:
        if current_os == "mac":
            cmd = (
                f"mount_smbfs //'{SMB_USER}':'{SMB_PASSWORD}'"
                f"@{SMB_SERVER}/{SMB_SHARE} {mount_point}"
            )
            subprocess.run(cmd, shell=True, check=True, capture_output=True)
        elif current_os == "linux":
            cmd = (
                f"sudo mount -t cifs //{SMB_SERVER}/{SMB_SHARE} {mount_point}"
                f" -o username={SMB_USER},password={SMB_PASSWORD}"
            )
            subprocess.run(cmd, shell=True, check=True, capture_output=True)
        elif current_os == "windows":
            ruta_smb = f"\\\\{SMB_SERVER}\\{SMB_SHARE}"
            cmd = [
                "net", "use", mount_point, ruta_smb,
                SMB_PASSWORD, f"/user:{SMB_USER}", "/persistent:no",
            ]
            subprocess.run(cmd, check=True, capture_output=True)

        logger.info(f"SMB montado en: {mount_point}")
        return True
    except Exception as e:
        logger.error(f"Error montando SMB: {e}")
        return False

# ============================================================================
def sincronizar_a_smb(ruta_local):
    if not os.path.isdir(ruta_local):
        logger.warning(f"La carpeta local no existe: {ruta_local}")
        return False

    if not montar_smb():
        logger.error("No se pudo sincronizar: SMB no montado.")
        return False

    mount_point = get_mount_point()
    current_os = get_current_os()

    if current_os == "windows" and mount_point.endswith(":") and len(mount_point) == 2:
        mount_point += "\\"

    subfolder = SMB_SUBFOLDER.lstrip("/")
    if subfolder:
        ruta_destino = os.path.normpath(os.path.join(mount_point, subfolder))
    else:
        ruta_destino = mount_point

    if not os.path.exists(ruta_destino):
        os.makedirs(ruta_destino)
        logger.info(f"Carpeta creada: {ruta_destino}")

    source = os.path.abspath(ruta_local)
    destination = os.path.abspath(ruta_destino)

    logger.info(f"Sincronizando a {current_os.upper()}: {destination}")

    try:
        if current_os in ("mac", "linux"):
            cmd = (
                f"find {source} -maxdepth 1 \\( -name '*.pdf' -o -name '*.xml' \\) "
                f"-exec rsync -av --remove-source-files {{}} {destination}/ \\;"
            )
            subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
            logger.info("Archivos copiados a NAS")
            subprocess.run(
                f"find {source} -type f \\( -name '*.json' -o -name '*.zip' \\) -delete",
                shell=True, check=True, capture_output=True, text=True,
            )
            logger.info("Limpieza de temporales completada")
        elif current_os == "windows":
            import shutil
            for archivo in os.listdir(source):
                if not archivo.lower().endswith((".pdf", ".xml")):
                    continue
                src_path = os.path.join(source, archivo)
                dst_path = os.path.join(destination, archivo)
                if os.path.isfile(src_path):
                    if os.path.exists(dst_path):
                        os.remove(dst_path)
                    shutil.move(src_path, dst_path)
            logger.info("Archivos copiados a NAS")
            for archivo in os.listdir(source):
                if archivo.lower().endswith((".json", ".zip")):
                    os.remove(os.path.join(source, archivo))
            logger.info("Limpieza de temporales completada")

        return True
    except Exception as e:
        logger.error(f"Error en sincronización: {e}")
        return False
# ============================================================================