office365-rest-python-client - gerenciador básico de arquivos no sharepoint

sharepoint_gerenciar.py

import logging
import os

from datetime import datetime

from office365.runtime.auth.user_credential import UserCredential
from office365.sharepoint.client_context import ClientContext


class SharePointGerenciar(Configuracoes):

    def __init__(self):
        self.prosseguir = True
        self.error = ""
        self.file_name = None
        self.file_time_created = None
        self.file_last_modified = None
        self.client_context = ClientContext("https://seu-server.sharepoint.com/sites/SeuSite").with_credentials(UserCredential("sharepoint_username", "sharepoint_password"))

    @staticmethod
    def logging_upload_progress(offset):
        logging.info("upload {} bytes".format(offset))

    def sharepoint_upload(self,
                          directory_filename,
                          list_title_sharepoint="arquivos_api",
                          directory_sharepoint="",
                          subdirectory_sharepoint=""):
        """
        :param directory_filename: caminho completo do diretório + arquivo que será feito o upload
        :type directory_filename: str
        :param list_title_sharepoint: nome da lista em que será feito o upload dos dados
        :type list_title_sharepoint: str
        :param directory_sharepoint: nome do diretório em que será feito o upload dos dados
        :type directory_sharepoint: str
        :param subdirectory_sharepoint: nome do subdiretório em que será feito o upload dos dados
        :type subdirectory_sharepoint: str
        """

        try:
            target_url = list_title_sharepoint

            if directory_sharepoint != "":
                target_url = os.path.join(target_url, directory_sharepoint)

                if subdirectory_sharepoint != "":
                    target_url = os.path.join(target_url, subdirectory_sharepoint)

            target_folder = self.client_context.web.get_folder_by_server_relative_url(target_url)

            size_chunk = 1000000
            file_size = os.path.getsize(directory_filename)

            logging.info("upload: {}".format(directory_filename))

            if file_size > size_chunk:
                target_file = target_folder.files.create_upload_session(directory_filename, size_chunk, self.logging_upload_progress)

            else:
                with open(directory_filename, "rb") as content_file:
                    file_content = content_file.read()
                basename = os.path.basename(directory_filename)

                target_file = target_folder.upload_file(basename, file_content)

            self.client_context.execute_query()

        except Exception as exception:
            self.prosseguir = False
            self.error = str(exception)
            logging.critical(exception)

        finally:
            pass

    def sharepoint_download(self,
                            directory_download,
                            filename_download,
                            list_title_sharepoint="arquivos_api",
                            directory_sharepoint="",
                            subdirectory_sharepoint=""):
        """
        :param directory_download: caminho completo do diretório onde será direcionado o download
        :type directory_download: str
        :param filename_download: nome.extensão do arquivo que deverá ser feito o download
        :type filename_download: str
        :param list_title_sharepoint: nome da lista em que será feito o upload dos dados
        :type list_title_sharepoint: str
        :param directory_sharepoint: nome do diretório em que será feito o upload dos dados
        :type directory_sharepoint: str
        :param subdirectory_sharepoint: nome do subdiretório em que será feito o upload dos dados
        :type subdirectory_sharepoint: str
        """

        try:
            target_url = list_title_sharepoint

            if directory_sharepoint != "":
                target_url = os.path.join(target_url, directory_sharepoint)

                if subdirectory_sharepoint != "":
                    target_url = os.path.join(target_url, subdirectory_sharepoint)

            target_url = os.path.join(target_url, filename_download)

            download_path = os.path.join(directory_download, os.path.basename(target_url))

            target_url = os.path.join("https://seu-server.sharepoint.com/sites/SeuSite", target_url)
            user_credentials = UserCredential("sharepoint_username", "sharepoint_password")

            logging.info("baixando: {}".format(target_url))

            with open(download_path, "wb") as local_file:
                File.from_url(target_url).with_credentials(user_credentials).download(local_file).execute_query()

            self.sharepoint_download_directory_filename = download_path

        except Exception as exception:
            self.prosseguir = False
            self.error = str(exception)
            logging.critical(exception)

        finally:
            pass

    def sharepoint_procurar(self,
                            part_filename,
                            list_title_sharepoint="arquivos_api",
                            directory_sharepoint="",
                            subdirectory_sharepoint=""):
        """
        :param part_filename: parte do nome do arquivo que deverá ser encontrado
        :type part_filename: str
        :param list_title_sharepoint: nome da lista de onde será feito o download dos dados
        :type list_title_sharepoint: str
        :param directory_sharepoint: nome do diretório em que será feito o upload dos dados
        :type directory_sharepoint: str
        :param subdirectory_sharepoint: nome do subdiretório em que será feito o upload dos dados
        :type subdirectory_sharepoint: str
        """

        self.file_name = None

        try:
            target_url = list_title_sharepoint

            if directory_sharepoint != "":
                target_url = os.path.join(target_url, directory_sharepoint)

                if subdirectory_sharepoint != "":
                    target_url = os.path.join(target_url, subdirectory_sharepoint)

            target_folder = self.client_context.web.get_folder_by_server_relative_url(target_url)

            list_items = target_folder.files
            self.client_context.load(list_items)
            self.client_context.execute_query()

            logging.info("procurando: {}\\{}".format(target_url, part_filename))

            for item in list_items:
                if part_filename.lower() in item.properties["Name"].lower():
                    self.file_name = item.properties["Name"]
                    self.file_time_created = datetime.strptime(item.properties["TimeCreated"], '%Y-%m-%dT%H:%M:%SZ')
                    self.file_last_modified = datetime.strptime(item.properties["TimeLastModified"], '%Y-%m-%dT%H:%M:%SZ')
                    break

            if self.file_name is None:
                raise ValueError("arquivo não encontrado")

        except Exception as exception:
            self.prosseguir = False
            self.error = str(exception)
            logging.critical(exception)

        finally:
            pass


if __name__ == "__main__":
    pass
    # Configuracoes().set_info(diretorio_arquivo=os.path.realpath(__file__).replace("\\", "/"))
    # 
    # classe_sharepoint_gerenciar = SharePointGerenciar()
    # classe_sharepoint_gerenciar.sharepoint_upload(directory_filename="D:/programacao/python/regionalspc/test/archive/no_file_download.txt",
    #                                               directory_sharepoint="in")
    # classe_sharepoint_gerenciar.sharepoint_encontrar(filename="no_file_download.txt")
    # classe_sharepoint_gerenciar.sharepoint_download(directory_download="D:/programacao/python/regionalspc/test/archive",
    #                                           filename_download="no_file_download.txt")