Librería ABM Servidor

abm_servidor.py Librería genérica para la gestión de operaciones SFTP entre servidores.

Este módulo proporciona la clase ABMServidor, que permite conectar, listar, copiar, borrar y verificar archivos entre múltiples servidores configurados en un archivo JSON. Está diseñado para ser importado y reutilizado en distintas aplicaciones o scripts de automatización.

Ejemplo de uso:

from abm_servidor import ABMServidor

abm = ABMServidor("abm_servidor.json")
archivos = abm.listar_todos_archivos("servidor_fuente")
Dependencias:

  • paramiko

  • json

  • os

  • sys

class abm_servidor.ABMServidor(config_file: str = 'abm_servidor.json')

Bases: object

Gestor genérico de operaciones SFTP.

Centraliza la configuración, conexión y manipulación de archivos en servidores remotos definidos en un archivo JSON.

archivo_existe(nombre_servidor: str, ruta: str) → Dict[str, Any] | None

Verifica si un archivo existe en un servidor remoto.

Parámetros:

  • nombre_servidor (str) – Identificador del servidor en el JSON.

  • ruta (str) – Ruta absoluta del archivo en el servidor.

Devuelve:

Diccionario con tamaño (int) y

mtime (float) si el archivo existe y es un archivo regular. Retorna None si no existe o si es un directorio.

Tipo del valor devuelto:

Optional[Dict[str, Any]]

borrar_archivo(nombre_servidor: str, ruta_archivo: str, callback_progreso: Callable[[str, int], None] | None = None) → Dict[str, Any]

Borra un archivo de un servidor remoto.

Parámetros:

  • nombre_servidor (str) – Nombre del servidor en el JSON.

  • ruta_archivo (str) – Ruta absoluta del archivo a borrar.

  • callback_progreso (Callable, optional) – Función (mensaje: str, porcentaje: int) -> None.

Devuelve:

Diccionario con el resultado de la operación:

  • exito (bool): True si se borró correctamente.

  • archivo (str): Ruta del archivo.

  • servidor (str): Nombre del servidor.

  • mensaje (str): Descripción del resultado.

  • accion (str): 'borrado', 'cancelado' o 'error'.

  • timestamp (str): Fecha y hora de la operación.

Tipo del valor devuelto:

Dict[str, Any]

conectar(nombre_servidor: str) → Tuple[SSHClient, SFTPClient]

Establece una conexión SSH y abre un canal SFTP con el servidor indicado.

Parámetros:

nombre_servidor (str) – Identificador del servidor en el JSON.

Devuelve:

Tupla con el cliente

SSH y el cliente SFTP abiertos.

Tipo del valor devuelto:

Tuple[paramiko.SSHClient, paramiko.SFTPClient]

Muestra:

  • ValueError – Si la contraseña no está configurada.

  • paramiko.AuthenticationException – Si las credenciales son incorrectas.

  • paramiko.SSHException – Si hay un error de conexión SSH.

copiar_archivo(servidor_origen: str, servidor_destino: str, ruta_origen: str, ruta_destino: str | None = None, forzar: bool = False, callback_confirmar: Callable[[Dict[str, Any], Dict[str, Any]], bool] | None = None, callback_progreso: Callable[[str, int], None] | None = None) → Dict[str, Any]

Copia un archivo entre dos servidores (en cualquier dirección).

Parámetros:

  • servidor_origen (str) – Nombre del servidor origen.

  • servidor_destino (str) – Nombre del servidor destino.

  • ruta_origen (str) – Ruta absoluta del archivo en el origen.

  • ruta_destino (str, optional) – Ruta absoluta en el destino. Si es None, se calcula automáticamente usando las ruta_base.

  • forzar (bool) – Si es True, sobrescribe sin pedir confirmación.

  • callback_confirmar (Callable, optional) – Función (info_origen, info_destino) -> bool para confirmar sobrescritura si el archivo ya existe.

  • callback_progreso (Callable, optional) – Función (mensaje: str, porcentaje: int) -> None.

Devuelve:

Diccionario con el resultado de la operación:

  • exito (bool): True si la operación terminó sin errores.

  • origen (str): Ruta de origen.

  • destino (str): Ruta de destino.

  • mensaje (str): Descripción del resultado.

  • accion (str): 'copiado', 'omitido' o 'cancelado'.

  • timestamp (str): Fecha y hora de la operación.

Tipo del valor devuelto:

Dict[str, Any]

listar_todos_archivos(nombre_servidor: str, callback: Callable[[str, int], None] | None = None) → List[Dict[str, Any]]

Lista recursivamente todos los archivos del servidor indicado.

Parámetros:

  • nombre_servidor (str) – Identificador del servidor en el JSON.

  • callback (Callable, optional) – Función para reportar progreso con la firma (mensaje: str, progreso: int) -> None.

Devuelve:

Lista de diccionarios, donde cada uno contiene:

  • ruta_relativa (str): Ruta relativa a la ruta_base.

  • ruta_absoluta (str): Ruta completa en el servidor.

  • tamaño (int): Tamaño en bytes.

  • mtime (float): Timestamp de última modificación.

Tipo del valor devuelto:

List[Dict[str, Any]]

obtener_servidor(nombre: str) → Dict[str, Any]

Obtiene la configuración de un servidor por su nombre.

Parámetros:

nombre (str) – Identificador del servidor en el JSON.

Devuelve:

Diccionario con la configuración del servidor

(host, port, username, password, ruta_base).

Tipo del valor devuelto:

Dict[str, Any]

Muestra:

ValueError – Si el nombre del servidor no existe en la configuración.