Le blog technique

### azure_list_blobs_properties.py ###
##### Script Python pour lister les proprietes de blobs dans un container Azure Blob Storage #####

# PARAMETERES D'ENTREE DU SCRIPT :
# --storage_account_name : Nom du compte de stockage Azure (default: mystorageaccount)
# --Cledestockage : Clé de stockage Azure
# --nomcontainer : Nom du container de stockage Azure (default: mycontainer)
# --MaxResults : Nombre maximum de blobs à récupérer par lot (default: 10000)
# --export_all_blobs_list : Exporter la liste de tous les blobs dans un fichier CSV (True/False) (default: False)
# --exportfolder : Chemin du dossier pour les fichiers CSV (default: .)
# --fichierrapport : Chemin du fichier de rapport (default: myreport.txt)
# --help : Afficher l'aide du script

# USAGE :
# WARNING: Ne pas pas positionner a True le parametre export_all_blobs_list si le container cible contiens beaucoup de blobs. Le fichier csv generé sera très volumineux (1 ligne / blob).
# python.exe .\azure_list_blobs_properties.py --storage_account_name <nom_du_compte_de_stockage> --Cledestockage <clé_de_stockage> --nomcontainer <nom_du_container> --export_all_blobs_list True/False



import os
import argparse
from datetime import datetime, timezone
from azure.storage.blob import BlobServiceClient
import csv


if __name__ == '__main__':

        ###########################################
        # Arguments du script
        ###########################################
        
        parser = argparse.ArgumentParser()

        parser.add_argument(
        '--storage_account_name',
        type=str,
        default='mystorageaccount',
        help='Nom du compte de stockage Azure'
        )


        parser.add_argument(
        '--Cledestockage',
        type=str,
        help='Clé de stockage Azure'
        )

        parser.add_argument(
        '--nomcontainer',
        type=str,
        default='mycontainer',
        help='nom du container de stockage Azure'
        )

        parser.add_argument(
        '--MaxResults',
        type=int,
        default=10000,
        help='Nombre maximum de blobs à récupérer par lot (par défaut: 10000)'
        )

        parser.add_argument(
        '--export_all_blobs_list',
        choices=['True', 'False'],
        default='False',
        help='Exporter la liste de tous les blobs dans un fichier CSV'
        )

        # export folder
        parser.add_argument(
        '--exportfolder',
        type=str,
        default='.',
        help='Chemin du dossier pour les fichiers CSV'
        )

        # fichier de rapport
        parser.add_argument(
        '--fichierrapport',
        type=str,
        default='myreport.txt',
        help='Nom du fichier de rapport'
        )

        # On parse les arguments
        args = parser.parse_args()

        # AUTRES PARAMETRES
        # URL du compte de stockage Azure
        account_url = f"https://{args.storage_account_name}.blob.core.windows.net"

        # On signifie a Python que l'on utilise PAS de proxy pour les connexions vers Azure
        os.environ['NO_PROXY'] = '.blob.core.windows.net,login.microsoftonline.com,management.azure.com'

        
        

        ### FONCTION POUR ECRIRE UN MESSAGE DANS UN FICHIER DE RAPPORT
        def write_to_report_file(message, filename):
            """
            Écrit un message dans un fichier de rapport.

            :param message: Message à écrire dans le fichier.
            :param
            filename: Nom du fichier de rapport.
            """
            try:
                with open(args.fichierrapport, 'a', encoding='utf-8') as fichier:
                    fichier.write(message + '\n')
                print(f"Message ecrit dans le fichier de rapport '{args.fichierrapport}'.")
            except Exception as e:
                print(f"Erreur lors de l'ecriture dans le fichier {args.fichierrapport}: {e}")
        # Fin de la fonction write_to_report_file



         ## FONCTION POUR TESTER L'ACCES AU CONTAINER DE STOCKAGE AZURE
        def test_access_to_azure_storage_container(account_url, credential, container_name):
            """
            Teste l'accès à un container de stockage Azure Blob.
            
            :param account_url: URL du compte de stockage Azure.
            :param credential: Clé d'accès ou SAS token pour le compte de stockage.
            :param container_name: Nom du container à tester.
            """
            try:
                blob_service_client = BlobServiceClient(account_url=account_url, credential=credential)
                container_client = blob_service_client.get_container_client(container_name)
                
                # Vérification de la connexion
                container_properties = container_client.get_container_properties()
                print(f"Connexion reussie au container '{container_name}'.")
                
                return True
            except Exception as e:
                print(f"Erreur lors de l'accès au container '{container_name}' : {e}")
                return False
        # Fin de la fonction test_access_to_azure_storage_container
        


        ## FONCTION POUR LISTER TOUT LES BLOBS DANS UN CONTAINER

        def list_all_blobs_in_container_by_lot(container_client, max_results):
            """
            Liste les blobs dans un conteneur Azure Blob Storage par lot avec une limite.

            :param container_name: Nom du conteneur.
            :param max_results: Nombre maximum de blobs à récupérer par lot.
            :return: Un générateur qui retourne les blobs par lot.
            """
            try:
                

                # Liste les blobs par lot
                blob_iterator = container_client.list_blobs(results_per_page=max_results).by_page()
                for blob_page in blob_iterator:
                    yield list(blob_page)  # Retourne un lot de blobs sous forme de liste

            except Exception as e:
                print(f"Erreur lors de la récupération des blobs : {e}")





        # FONCTION POUR ECRIRE UN CSV
        def write_csv(data, filename):
            """
            Écrit les données dans un fichier CSV.

            :param data: Liste de dictionnaires contenant les données à écrire.
                        Chaque dictionnaire représente une ligne, avec les clés comme en-têtes.
            :param filename: Nom du fichier CSV à créer.
            """
            try:
                # Vérifier si la liste de données n'est pas vide
                if not data:
                    print("Aucune donnee a ecrire dans le fichier CSV.")
                    return

                # Ouvrir le fichier en mode écriture
                with open(filename, mode='w', newline='', encoding='utf-8') as csvfile:
                    # Créer un objet writer
                    writer = csv.DictWriter(csvfile, fieldnames=data[0].keys())

                    # Écrire l'en-tête
                    writer.writeheader()

                    # Écrire les lignes
                    writer.writerows(data)

                print(f"Les donnees ont ete ecrites avec succes dans le fichier '{filename}'.")
            except Exception as e:
                print(f"Erreur lors de l'ecriture dans le fichier CSV : {e}")

    

        #Debut du traitement
        print(f"Debut du traitement pour le container {args.nomcontainer}.")



        # TEST DE L'ACCES AU CONTAINER DE STOCKAGE AZURE
        print("Test de l'acces au container de stockage Azure...")
        test_access_to_azure_storage_container(
            account_url=account_url,
            credential=args.Cledestockage,
            container_name=args.nomcontainer
        )
        # Fin du test de l'accès au container de stockage Azure
        # #################################################
  
        # On recupere la date d'aujourd'hui
        today = datetime.now().replace(tzinfo=timezone.utc)

        # Instanciation du client de service Blob
        container = BlobServiceClient(
            account_url=account_url,
            credential=args.Cledestockage
        ).get_container_client(args.nomcontainer)



        all_blobs_generator = list_all_blobs_in_container_by_lot(
            container_client=container,
            max_results=args.MaxResults # Nombre maximum de blobs à récupérer par lot
        )

        # On verifie que le generator n'est pas vide
        if not all_blobs_generator:
            print(f"Aucun blob trouve dans le container '{args.nomcontainer}'.")
            exit(1)  # Sortie du script si aucun blob n'est trouvé
        # Fin de la récupération des blobs dans le container      



        # Initialisation de la liste pour stocker les infos de tout les blobs (utilisé uniquement si l'option export_all_blobs_list est a True)
        all_blobs_info = []
        
         

        # On parcourt tous les blobs du container
        print("\n"*2)
        print("#"* 80)
        print(f"Analyse et traitement des blobs dans le container {args.nomcontainer}...")
        print("#"* 80)
       
        # initialisation de l'index pour afficher le numero du lot et le nombre total de lot a la fin du traitement
        lotcount = 0
        # initialisation de l'index pour incrementer le nombre de blobs
        blobcount = 0

        try:
            
                for blob_lot in all_blobs_generator:
                    lotcount += 1
                    print(f"Nombre de blobs dans le lot {lotcount}: {len(blob_lot)}")
                    for blob in blob_lot:
                            # On incremente le nombre de blobs
                            blobcount += 1
                            # On recupere le client du blob
                            blob_client = container.get_blob_client(blob.name)                        
                            # Récupération des propriétés du blob
                            blob_properties = blob_client.get_blob_properties()

                            
                            # On alimente all_blobs_info
                            all_blobs_info.append({
                                    'Container': blob.container,
                                    'Blob_Name': blob.name,
                                    'Blob_Size': blob.size,
                                    'is_deleted': blob.deleted,
                                    'remaining_retention_days': blob.remaining_retention_days,
                                    'is_current_version': blob.is_current_version,
                                    'last_modified': blob.last_modified,
                                    'expiration': blob_properties.immutability_policy.expiry_time if blob_properties.immutability_policy else None
                                })
            

        except StopIteration:
                print("Fin de l'iterateur, tous les lots de blobs ont ete analyses et traites.")
        except Exception as e:
                print(f"Erreur lors de la recuperation des blobs : {e}")
            # Fin de la récupération des blobs dans le container


        # affichage du nombre lots analyses
        print("\n")
        print("#"* 80)
        print(f"Nombre total de lots de blobs analyses: {lotcount}")
        print("#"* 80)



        # STATISTIQUES SUR LES BLOBS
        print("\n")
        print("#"* 80)
        print(f"Nombre total de blobs dans le container {args.nomcontainer}: {blobcount}")
        print("#"* 80)


        # On affiche certaines proprietes de all_blobs_info
        print("\n")
        print("#"* 80)
        print(f"Contenu de la liste de tous les blobs du container {args.nomcontainer}:")
        for blob_info in all_blobs_info:
            print(f"Blob Name: {blob_info['Blob_Name']}, Size: {blob_info['Blob_Size']} octets, Expiration: {blob_info['expiration']}, Last Modified: {blob_info['last_modified']}")
        print("#"* 80)


        # ECRIRE LE CONTENU DE all_blobs_info DANS UN CSV (Si l'option export_all_blobs_list est a True)
        if args.export_all_blobs_list == 'True':
            print("\n")
            print("#"* 80)
            print(f"Ecriture de la liste de tout les blobs du container dans le fichier CSV avec la date d'expiration...")
            csv_filepath = f"{args.exportfolder}\\{args.nomcontainer}_all_blobs_info.csv"
            write_csv(data=all_blobs_info, filename=csv_filepath)
            print("#"* 80)

        


        # On ajoute un message de fin dans le fichier de rapport avec toutes les statistiques
        print("\n")
        print("#"* 80)
        msg = f"Fin du traitement pour le container {args.nomcontainer}\nLe traitement s'est opere en {lotcount} lots\nNombre total de blobs: {blobcount}\nFichier CSV des blobs: {csv_filepath if args.export_all_blobs_list == 'True' else 'Non exporté'}"
        print(msg)

        # Ecriture dans le fichier de rapport
        write_to_report_file(
        message = f"\n### {today.isoformat()} ###\n{msg}\n#########################################",
        filename=f"{args.exportfolder}\\{args.fichierrapport}"
        )
        

        print("#"* 80)
        print("\n")
        # Fin du script
        print("#"* 40 + " FIN DU SCRIPT " + "#"*40)