Automatisation & Python : se connecter à un serveur sFTP et télécharger des fichiers

La programmation permet d’automatiser moult tâches du quotidien y compris la collecte de fichiers disséminés sur divers serveurs.

Collecte FEC 2024 bis

Quotidiennement, nous avons besoin de données qui proviennent de systèmes divers (comptabilité, gestion commerciale…) et qui sont accessibles sur un réseau local ou sur internet par le biais d’une requête ou d’une connexion à un service (API, http, sFTP…).

Une des raisons qui m’ont poussé à m’intéresser à Python est que le VBA ne permet pas aisément de se connecter à un serveur sFTP…

Installer la bibliothèque externe Paramiko :

La connexion à un serveur sFTP nécessite la bibliothèque externe Paramiko. Toutes les infos sur le site de l’éditeur : https://docs.paramiko.org/en/latest/.

Pour l’installer, ouvrir une invite de commandes (CMD.exe) et entrez la séquence :

pip install paramiko

pip install paramiko

Après validation (touche Entrée), l’installation de la bibliothèque Paramiko se lance :

PYTHON Install Paramiko

La bibliothèque étant installée (“Sucessfully installed…”), elle peut être appelée par un script Python à l’aide de la commande import paramiko.

Fonctionnement du script fourni ci-après :

Ce script ajoute une fonction telechargement_sftp(hostname, port, username, password, remote_path, local_path) qui rapatrie tous les fichiers contenus dans le dossier désigné par le paramètre remote_path (serveur sFTP) vers le dossier local_path (PC) après connexion sécurisée au serveur (hostname, port, username, password).

Cette fonction s’occupe passer en revue le dossier remote_path : de créer tous les sous-dossiers le cas échéant (en utilisant les propriétés de récursivité du langage Python) et de copier les fichiers contenus dans ces sous-dossiers.

Code source du script Connexion sFTP.py :

# Téléchargement de fichiers situés sur un serveur sFTP

# v1.0 11/2023

import os

import time

import paramiko # bibliothèque externe sFTP

def telecharge_fichiers_sftp_recursive(sftp, remote_path, local_path):

try:

# Liste des fichiers et dossiers dans le répertoire distant

remote_files = sftp.listdir(remote_path)

# Téléchargement de chaque fichier du répertoire distant

for remote_file in remote_files:

remote_file_path = f"{remote_path}/{remote_file}"

local_file_path = f"{local_path}\\{remote_file}"

# Vérifie si l'élément distant est un répertoire

if is_directory(sftp, remote_file_path):

# Crée le sous-dossier local si nécessaire

if not os.path.exists(local_file_path):

os.makedirs(local_file_path)

# Appelle récursivement la fonction pour traiter le sous-dossier distant

telecharge_fichiers_sftp_recursive(sftp, remote_file_path, local_file_path)

else:

# Téléchargement du fichier

sftp.get(remote_file_path, local_file_path)

# Obtient la date de modification du fichier distant

mtime = sftp.stat(remote_file_path).st_mtime

# Définit la date de modification locale du fichier téléchargé

os.utime(local_file_path, (time.time(), mtime))

print(f"\tTéléchargé : {remote_file}")

except FileNotFoundError as e:

print(f"\tErreur : Répertoire distant introuvable. {e}")

except Exception as e:

print(f"\tErreur : {e}")

def is_directory(sftp, remote_path):

try:

# Liste des fichiers et dossiers dans le répertoire distant

files = sftp.listdir(remote_path)

# Si le chemin distant est un répertoire, la liste contiendra des fichiers et des sous-répertoires

return True

except IOError:

# Si le chemin distant est un fichier, la liste sera vide et IOError sera levée

return False

def telechargement_sftp(hostname, port, username, password, remote_path, local_path):

# Affichage des dossiers distant et local

print("-" * 85)

print(f"Transfert des fichiers du serveur {hostname}:{port}:{remote_path} vers {local_path} :")

# Crée une instance de transport

transport = paramiko.Transport((hostname, port))

transport.connect(username=username, password=password)

# Crée une instance SFTP

sftp = paramiko.SFTPClient.from_transport(transport)

# Appelle la fonction récursive pour télécharger les fichiers SFTP de manière récursive

telecharge_fichiers_sftp_recursive(sftp, remote_path, local_path)

# Ferme la connexion SFTP

sftp.close()

# Ferme le transport

transport.close()

print()

print("-" * 85)

print("Transfert de fichiers sFTP :")

print()

# Téléchargement des balances et fiches tiers

# Paramètres de connexion

hostname = 'adresse_serveur.fr'

port = 2022

username = 'user1'

password = 'mdp1'

remote_path = '/projects/benoit'

local_path = 'C:\\COLLECTE FEC & BG\\SOURCE\\benoit'

telechargement_sftp(hostname, port, username, password, remote_path, local_path)

# Téléchargement des FEC

# Paramètres de connexion

hostname = 'adresse_serveur.fr'

port = 2022

username = 'user2'

password = 'mdp2'

remote_path = '/projects/fecfiles'

local_path = 'C:\\COLLECTE FEC & BG\\SOURCE\\fecfiles'

telechargement_sftp(hostname, port, username, password, remote_path, local_path)

# Téléchargement de fichiers situés sur un serveur sFTP # v1.0 11/2023 import os import time import paramiko # bibliothèque externe sFTP def telecharge_fichiers_sftp_recursive(sftp, remote_path, local_path): try: # Liste des fichiers et dossiers dans le répertoire distant remote_files = sftp.listdir(remote_path) # Téléchargement de chaque fichier du répertoire distant for remote_file in remote_files: remote_file_path = f"{remote_path}/{remote_file}" local_file_path = f"{local_path}\\{remote_file}" # Vérifie si l'élément distant est un répertoire if is_directory(sftp, remote_file_path): # Crée le sous-dossier local si nécessaire if not os.path.exists(local_file_path): os.makedirs(local_file_path) # Appelle récursivement la fonction pour traiter le sous-dossier distant telecharge_fichiers_sftp_recursive(sftp, remote_file_path, local_file_path) else: # Téléchargement du fichier sftp.get(remote_file_path, local_file_path) # Obtient la date de modification du fichier distant mtime = sftp.stat(remote_file_path).st_mtime # Définit la date de modification locale du fichier téléchargé os.utime(local_file_path, (time.time(), mtime)) print(f"\tTéléchargé : {remote_file}") except FileNotFoundError as e: print(f"\tErreur : Répertoire distant introuvable. {e}") except Exception as e: print(f"\tErreur : {e}") def is_directory(sftp, remote_path): try: # Liste des fichiers et dossiers dans le répertoire distant files = sftp.listdir(remote_path) # Si le chemin distant est un répertoire, la liste contiendra des fichiers et des sous-répertoires return True except IOError: # Si le chemin distant est un fichier, la liste sera vide et IOError sera levée return False def telechargement_sftp(hostname, port, username, password, remote_path, local_path): # Affichage des dossiers distant et local print("-" * 85) print(f"Transfert des fichiers du serveur {hostname}:{port}:{remote_path} vers {local_path} :") # Crée une instance de transport transport = paramiko.Transport((hostname, port)) transport.connect(username=username, password=password) # Crée une instance SFTP sftp = paramiko.SFTPClient.from_transport(transport) # Appelle la fonction récursive pour télécharger les fichiers SFTP de manière récursive telecharge_fichiers_sftp_recursive(sftp, remote_path, local_path) # Ferme la connexion SFTP sftp.close() # Ferme le transport transport.close() print() print("-" * 85) print("Transfert de fichiers sFTP :") print() # Téléchargement des balances et fiches tiers # Paramètres de connexion hostname = 'adresse_serveur.fr' port = 2022 username = 'user1' password = 'mdp1' remote_path = '/projects/benoit' local_path = 'C:\\COLLECTE FEC & BG\\SOURCE\\benoit' telechargement_sftp(hostname, port, username, password, remote_path, local_path) # Téléchargement des FEC # Paramètres de connexion hostname = 'adresse_serveur.fr' port = 2022 username = 'user2' password = 'mdp2' remote_path = '/projects/fecfiles' local_path = 'C:\\COLLECTE FEC & BG\\SOURCE\\fecfiles' telechargement_sftp(hostname, port, username, password, remote_path, local_path)

# Téléchargement de fichiers situés sur un serveur sFTP
# v1.0 11/2023

import os
import time
import paramiko # bibliothèque externe sFTP

def telecharge_fichiers_sftp_recursive(sftp, remote_path, local_path):
    try:
        # Liste des fichiers et dossiers dans le répertoire distant
        remote_files = sftp.listdir(remote_path)

        # Téléchargement de chaque fichier du répertoire distant
        for remote_file in remote_files:
            remote_file_path = f"{remote_path}/{remote_file}"
            local_file_path = f"{local_path}\\{remote_file}"
            
            # Vérifie si l'élément distant est un répertoire
            if is_directory(sftp, remote_file_path):
                # Crée le sous-dossier local si nécessaire
                if not os.path.exists(local_file_path):
                    os.makedirs(local_file_path)

                # Appelle récursivement la fonction pour traiter le sous-dossier distant
                telecharge_fichiers_sftp_recursive(sftp, remote_file_path, local_file_path)
            else:
                # Téléchargement du fichier
                sftp.get(remote_file_path, local_file_path)

                # Obtient la date de modification du fichier distant
                mtime = sftp.stat(remote_file_path).st_mtime

                # Définit la date de modification locale du fichier téléchargé
                os.utime(local_file_path, (time.time(), mtime))

                print(f"\tTéléchargé : {remote_file}")

    except FileNotFoundError as e:
        print(f"\tErreur : Répertoire distant introuvable. {e}")
    except Exception as e:
        print(f"\tErreur : {e}")

def is_directory(sftp, remote_path):
    try:
        # Liste des fichiers et dossiers dans le répertoire distant
        files = sftp.listdir(remote_path)

        # Si le chemin distant est un répertoire, la liste contiendra des fichiers et des sous-répertoires
        return True
    except IOError:
        # Si le chemin distant est un fichier, la liste sera vide et IOError sera levée
        return False

def telechargement_sftp(hostname, port, username, password, remote_path, local_path):
    # Affichage des dossiers distant et local
    print("-" * 85)
    print(f"Transfert des fichiers du serveur {hostname}:{port}:{remote_path} vers {local_path} :")

    # Crée une instance de transport
    transport = paramiko.Transport((hostname, port))
    transport.connect(username=username, password=password)

    # Crée une instance SFTP
    sftp = paramiko.SFTPClient.from_transport(transport)

    # Appelle la fonction récursive pour télécharger les fichiers SFTP de manière récursive
    telecharge_fichiers_sftp_recursive(sftp, remote_path, local_path)

    # Ferme la connexion SFTP
    sftp.close()

    # Ferme le transport
    transport.close()

    print()


print("-" * 85)
print("Transfert de fichiers sFTP :")
print()

# Téléchargement des balances et fiches tiers

# Paramètres de connexion
hostname = 'adresse_serveur.fr'
port = 2022
username = 'user1'
password = 'mdp1'
remote_path = '/projects/benoit'
local_path = 'C:\\COLLECTE FEC & BG\\SOURCE\\benoit'

telechargement_sftp(hostname, port, username, password, remote_path, local_path)

# Téléchargement des FEC

# Paramètres de connexion
hostname = 'adresse_serveur.fr'
port = 2022
username = 'user2'
password = 'mdp2'
remote_path = '/projects/fecfiles'
local_path = 'C:\\COLLECTE FEC & BG\\SOURCE\\fecfiles'

telechargement_sftp(hostname, port, username, password, remote_path, local_path)

Copie écran de l’exécution du programme :

Le nom des fichiers téléchargés su serveur sFTP vers le PC est affiché à l’écran (pour des raisons de confidentialité, le nom des serveurs a été anonymisé) :

Exécution Connexion sFTP.py

Ce script évite à l’utilisateur de télécharger manuellement les fichiers lui faisant économiser un temps précieux.

Filezilla

Tous les articles en rapport avec le langage Python

___

Pour approfondir le sujet : se former à la programmation en langage Python pour automatiser ses tâches

Bio
Derniers articles

Benoît RIVIERE

Après seize années passées en cabinet d’expertise-comptable et de commissariat aux comptes (où j’ai exercé comme expert-comptable et chef de mission audit), j’ai pris le poste de directeur comptable d’un groupe de distribution automobile en novembre 2014. Au cours de ma carrière, j’ai acquis une expérience significative en audit et en exploitation des systèmes d’information (analyse de données, automatisation des tâches, programmation informatique) au service de la production des comptes annuels et consolidés. C’est cette expérience personnelle et ma passion pour l’informatique que je partage sur ce blog. Mon CV / Réalisations personnelles et projets informatiques / Ma collection / Me contacter

Audit & Systèmes d'Information

Pistes de réflexion sur l’audit et les systèmes d’information

Automatisation & Python : se connecter à un serveur sFTP et télécharger des fichiers

Benoît RIVIERE

Derniers articles parBenoît RIVIERE (voir tous)

Un commentaire

Laisser un commentaire