helpers

Helpers para operações comuns ao SDK PLDPro.

download(method, filename, save_path='', specific_file=False, as_zip=False)

Realiza o download de arquivos de entrada de um caso.

Parameters:

Name	Type	Description	Default
method	Callable	Função de download a ser chamada.	required
caso_id	Union[UUID, str]	Identificador único do caso.	required
filename	Union[Path, str]	Nome do arquivo a ser baixado.	required
save_path	Union[Path, str]	Caminho onde o arquivo será salvo.	''
specific_file	bool	Indica se o download é de um arquivo específico.	False
as_zip	bool	Indica se o arquivo deve ser baixado como zip.	False

Returns:

Type	Description
str	Caminho do arquivo salvo ou string vazia em caso de erro.

Source code in pldpro_sdk/core/helpers.py

def download(
    method: Callable,
    filename: Union[Path, str],
    save_path: Union[Path, str] = "",
    specific_file: bool = False,
    as_zip: bool = False,
) -> str:
    """
    Realiza o download de arquivos de entrada de um caso.

    :param method: Função de download a ser chamada.
    :type method: Callable
    :param caso_id: Identificador único do caso.
    :type caso_id: Union[UUID, str]
    :param filename: Nome do arquivo a ser baixado.
    :type filename: Union[Path, str]
    :param save_path: Caminho onde o arquivo será salvo.
    :type save_path: Union[Path, str]
    :param specific_file: Indica se o download é de um arquivo específico.
    :type specific_file: bool
    :param as_zip: Indica se o arquivo deve ser baixado como zip.
    :type as_zip: bool
    :return: Caminho do arquivo salvo ou string vazia em caso de erro.
    :rtype: str
    """
    try:
        if isinstance(caso_id, str):
            caso_id = UUID(caso_id)

        if isinstance(filename, str):
            filename = Path(filename)

        if isinstance(save_path, str):
            save_path = Path(save_path)


        if save_path == "":
            save_path = Path.cwd() / filename
        else:
            save_path = Path(save_path) / filename

        if specific_file:
            api_response = method(caso_id, filename.name)
        else:
            api_response = method(caso_id)

        if as_zip:
            save_path = save_path.with_suffix(".zip")

        with open(save_path, "wb") as f:
            f.write(api_response)

        return str(save_path)
    except (ApiException, ValueError, FileNotFoundError, OSError) as e:
        error_msg = get_error_message(e) if isinstance(e, ApiException) else str(e)
        logger.exception(
            f"Ocorreu o seguinte erro ao baixar o(s) arquivo(s) de entrada do caso {caso_id}:\n\t {error_msg}",
            exc_info=True,
        )
        return ""

download_api_pub_key(method, usuario, workdir, filename, node_uuid=None, save_path='')

Realiza o download de arquivos de entrada de um caso.

Parameters:

Name	Type	Description	Default
method	Callable	Função de download a ser chamada.	required
filename	Union[Path, str]	Nome do arquivo a ser baixado.	required
save_path	Union[Path, str]	Caminho onde o arquivo será salvo.	''

Returns:

Type	Description
str	Caminho do arquivo salvo ou string vazia em caso de erro.

Source code in pldpro_sdk/core/helpers.py

def download_api_pub_key(
    method: Callable,
    usuario:str,
    workdir:str,
    filename: Union[Path, str],
    node_uuid:Union[UUID, str]=None,
    save_path: Union[Path, str] = "",
) -> str:
    """
    Realiza o download de arquivos de entrada de um caso.

    :param method: Função de download a ser chamada.
    :type method: Callable
    :param filename: Nome do arquivo a ser baixado.
    :type filename: Union[Path, str]
    :param save_path: Caminho onde o arquivo será salvo.
    :type save_path: Union[Path, str]
    :return: Caminho do arquivo salvo ou string vazia em caso de erro.
    :rtype: str
    """
    try:

        auth_node_request_pub_key = AuthNodeRequestPubKey(
            usuario=usuario, workdir=workdir, uuidNode=node_uuid
        )

        if save_path == "":
            save_path = Path.cwd() / filename
        else:
            save_path = Path(save_path) / filename

        api_response = method(node_uuid,auth_node_request_pub_key)

        with open(save_path, "wb") as f:
            f.write(api_response)

        return str(save_path)
    except (ApiException, ValueError, OSError) as e:
        error_msg = get_error_message(e) if isinstance(e, ApiException) else str(e)
        logger.exception(
            f"Ocorreu o seguinte erro ao baixar a chave pública do usuario {usuario}:\n\t {error_msg}",
            exc_info=True,
        )
        return ""

get_error_message(api_exception)

Obtém a mensagem de erro a partir de uma exceção da API.

Parameters:

Name	Type	Description	Default
api_exception	ApiException	Exceção gerada pela API.	required

Returns:

Type	Description
str	Mensagem de erro extraída da exceção.

Source code in pldpro_sdk/core/helpers.py

def get_error_message(api_exception: ApiException) -> str:
    """
    Obtém a mensagem de erro a partir de uma exceção da API.

    :param api_exception: Exceção gerada pela API.
    :type api_exception: ApiException
    :return: Mensagem de erro extraída da exceção.
    :rtype: str
    """
    body = json.loads(str(api_exception.body))
    error_msg = body.get("message") or body.get("menssage") or str(body)
    return error_msg

get_logger(name='pldpro_sdk')

Retorna uma instância de logger configurada para o SDK. Essa função cria loggers padronizados, reutilizáveis e idempotentes.

Source code in pldpro_sdk/core/helpers.py

def get_logger(name: str = "pldpro_sdk") -> logging.Logger:
    """
    Retorna uma instância de logger configurada para o SDK.
    Essa função cria loggers padronizados, reutilizáveis e idempotentes.
    """
    logger = logging.getLogger(name)

    if not logger.handlers:
        logger.setLevel(logging.INFO)

        formatter = logging.Formatter(
            fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
            datefmt="%Y-%m-%d %H:%M:%S",
        )

        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(formatter)

        logger.addHandler(console_handler)

        logger.propagate = False

    return logger

upload(method, error_context, caso_id, files, should_zip=True, save_zip=False, use_full_path=True, *args, **kwargs)

Realiza o upload de arquivos ou diretórios para um caso.

Parameters:

Name	Type	Description	Default
method	Callable	Função de upload a ser chamada.	required
error_context	str	Contexto do erro para mensagens de log.	required
caso_id	Union[UUID, str]	Identificador único do caso.	required
files	Union[List[Path], List[str], Path, str]	Arquivo(s) ou diretório(s) a serem enviados.	required
should_zip	bool	Indica se os arquivos devem ser compactados em um zip antes do upload.	True
save_zip	bool	Indica se o arquivo zip deve ser salvo em disco.	False
use_full_path	bool	Indica se o caminho completo deve ser usado no upload.	True

Returns:

Type	Description
Union[FileUploadResponse, None]	Resposta do upload.

Source code in pldpro_sdk/core/helpers.py

def upload(
    method: Callable,
    error_context: str,
    caso_id: Union[UUID, str],
    files: Union[List[Path], List[str], Path, str],
    should_zip: bool = True,
    save_zip: bool = False,
    use_full_path: bool = True,
    *args,
    **kwargs,
) -> Union[FileUploadResponse, None]:
    """
    Realiza o upload de arquivos ou diretórios para um caso.

    :param method: Função de upload a ser chamada.
    :type method: Callable
    :param error_context: Contexto do erro para mensagens de log.
    :type error_context: str
    :param caso_id: Identificador único do caso.
    :type caso_id: Union[UUID, str]
    :param files: Arquivo(s) ou diretório(s) a serem enviados.
    :type files: Union[List[Path], List[str], Path, str]
    :param should_zip: Indica se os arquivos devem ser compactados em um zip antes do upload.
    :type should_zip: bool
    :param save_zip: Indica se o arquivo zip deve ser salvo em disco.
    :type save_zip: bool
    :param use_full_path: Indica se o caminho completo deve ser usado no upload.
    :type use_full_path: bool
    :return: Resposta do upload.
    :rtype: Union[FileUploadResponse, None]
    """

    def zip_directory_to_memory(dir_path: Path) -> bytes:
        """Compacta um diretório em um arquivo zip na memória e retorna os bytes."""
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
            for root, dirs, files in os.walk(dir_path):
                for file in files:
                    file_path = os.path.join(root, file)
                    rel_path = os.path.relpath(file_path, start=dir_path)
                    zip_file.write(file_path, rel_path)
        zip_buffer.seek(0)
        return zip_buffer.read()

    def zip_list_of_files_to_memory(file_paths: List[Path]) -> bytes:
        """Compacta uma lista de arquivos em um arquivo zip na memória e retorna os bytes."""
        zip_buffer = io.BytesIO()
        with zipfile.ZipFile(zip_buffer, "w", zipfile.ZIP_DEFLATED) as zip_file:
            for file_path in file_paths:
                zip_file.write(file_path, file_path.name)
        zip_buffer.seek(0)
        return zip_buffer.read()

    def save_zip_to_disk(zip_bytes: bytes, zip_path: Path) -> None:
        """Salva os bytes do arquivo zip em disco."""
        with open(zip_path, "wb") as f:
            f.write(zip_bytes)

    def list_of_files_to_upload(file_paths: List[Path]) -> List[tuple]:
        """Prepara uma lista de tuplas (nome_arquivo, bytes_arquivo) para upload."""
        upload_files = []
        for file_path in file_paths:
            with open(file_path, "rb") as f:
                file_data = f.read()
            upload_files.append((file_path.name, file_data))
        return upload_files

    try:
        if isinstance(caso_id, str):
            caso_id = UUID(caso_id)

        # Se for uma lista de arquivos
        if isinstance(files, list):
            if all(isinstance(f, str) for f in files):
                file_paths = [Path(f) for f in files]
            elif all(isinstance(f, Path) for f in files):
                file_paths: List[Path] = files  # type: ignore
            else:
                raise ValueError(
                    "A lista de arquivos deve conter apenas strings ou objetos Path."
                )

            if should_zip:
                zip_data = zip_list_of_files_to_memory(file_paths)
                # Salvando o arquivo zip em disco
                if save_zip:
                    first_file_path = file_paths[0]
                    zip_disk_path = first_file_path.parent / "upload_files.zip"
                    save_zip_to_disk(zip_data, zip_disk_path)
                ret = method(caso_id, [("upload_files.zip", zip_data)], *args, **kwargs)
            else:
                # NOTE: Alguns métodos aceitam múltiplos arquivos sem zip
                upload_files = list_of_files_to_upload(file_paths)
                logger.debug("Uploading files: %s", upload_files)
                ret = method(caso_id, upload_files, *args, **kwargs)
            return ret

        # Se for um único Path ou string
        elif isinstance(files, str):
            file_path = Path(files)

            if file_path.is_dir():
                file_data = zip_directory_to_memory(file_path)
                filename = f"{file_path.name}.zip"

                # FIXME: Verificar se é necessário validar diretório!
                # if not validate_filename_for_method(filename, method.__name__):
                #     raise ValueError(f"O diretório {file_path} não é válido para o método {method.__name__}.")

                # Salvando o arquivo zip em disco
                if save_zip:
                    save_zip_to_disk(file_data, file_path.with_suffix(".zip"))
                ret = method(caso_id, [(filename, file_data)], *args, **kwargs)
                return ret

            elif file_path.is_file():
                # Validar nome do arquivo
                if not validate_filename_for_method(file_path.name, method.__name__):
                    raise ValueError(
                        f"O arquivo {file_path.name} não é válido para o método {method.__name__}."
                    )

                # Para os endpoints que aceitam upload direto sem zip
                if not should_zip:
                    with open(file_path, "rb") as f:
                        file_data = f.read()

                # Se já for um .zip, envia direto
                elif file_path.suffix.lower() == ".zip":
                    with open(file_path, "rb") as f:
                        file_data = f.read()

                else:
                    # Qualquer outro arquivo, compacta antes de enviar
                    file_data = zip_list_of_files_to_memory([file_path])

                    # salva zip localmente se for necessário
                    if save_zip:
                        zip_disk_path = file_path.with_suffix(".zip")
                        save_zip_to_disk(file_data, zip_disk_path)

                # Executa o método de upload
                upload_target = str(file_path) if use_full_path else file_path.name
                ret = method(caso_id, (upload_target, file_data), *args, **kwargs)
                return ret

            else:
                raise ValueError(
                    f"{file_path} não é um arquivo nem um diretório válido."
                )

    except (ApiException, ValueError, OSError) as e:
        error_msg = get_error_message(e) if isinstance(e, ApiException) else str(e)
        logger.exception(
            f"Ocorreu o seguinte erro durante o upload {error_context}:\n\t {error_msg}",
            exc_info=True,
        )
    return None

upload_license(method, error_context, files, should_zip=True, save_zip=False, use_full_path=True, *args, **kwargs)

Realiza o upload de arquivos ou diretórios para endpoints sem caso_id.

Parameters:

Name	Type	Description	Default
method	Callable	Função de upload a ser chamada.	required
error_context	str	Contexto do erro para mensagens de log.	required
files	Union[List[Path], List[str], Path, str]	Arquivo(s) ou diretório(s) a serem enviados.	required
should_zip	bool	Indica se os arquivos devem ser compactados em um zip antes do upload.	True
save_zip	bool	Indica se o arquivo zip deve ser salvo em disco.	False
use_full_path	bool	Indica se o caminho completo deve ser usado no upload.	True

Returns:

Type	Description
Union[FileUploadResponse, None]	Resposta do upload.

Source code in pldpro_sdk/core/helpers.py

def upload_license(
    method: Callable,
    error_context: str,
    files: Union[List[Path], List[str], Path, str],
    should_zip: bool = True,
    save_zip: bool = False,
    use_full_path: bool = True,
    *args,
    **kwargs,
) -> Union[FileUploadResponse, None]:
    """
    Realiza o upload de arquivos ou diretórios para endpoints sem `caso_id`.

    :param method: Função de upload a ser chamada.
    :type method: Callable
    :param error_context: Contexto do erro para mensagens de log.
    :type error_context: str
    :param files: Arquivo(s) ou diretório(s) a serem enviados.
    :type files: Union[List[Path], List[str], Path, str]
    :param should_zip: Indica se os arquivos devem ser compactados em um zip antes do upload.
    :type should_zip: bool
    :param save_zip: Indica se o arquivo zip deve ser salvo em disco.
    :type save_zip: bool
    :param use_full_path: Indica se o caminho completo deve ser usado no upload.
    :type use_full_path: bool
    :return: Resposta do upload.
    :rtype: Union[FileUploadResponse, None]
    """
    return _execute_upload(
        method=method,
        error_context=error_context,
        files=files,
        should_zip=should_zip,
        save_zip=save_zip,
        use_full_path=use_full_path,
        upload_callback=lambda payload, *callback_args, **callback_kwargs: method(
            payload, *callback_args, **callback_kwargs
        ),
        *args,
        **kwargs,
    )

validate_filename_for_method(filename, method_name)

Valida se o nome do arquivo é compatível com o método de upload.

Parameters:

Name	Type	Description	Default
filename	str	Nome do arquivo a ser validado.	required
method_name	str	Nome do método de upload.	required

Returns:

Type	Description
bool	True se o nome do arquivo for válido para o método, False caso contrário.

Source code in pldpro_sdk/core/helpers.py

def validate_filename_for_method(filename: str, method_name: str) -> bool:
    """
    Valida se o nome do arquivo é compatível com o método de upload.

    :param filename: Nome do arquivo a ser validado.
    :type filename: str
    :param method_name: Nome do método de upload.
    :type method_name: str
    :return: True se o nome do arquivo for válido para o método, False caso contrário.
    :rtype: bool
    """
    method_config = methods_config.get(method_name)

    if not method_config:
        return True  # Sem configuração específica, aceita qualquer arquivo

    valid_filenames = method_config.get("file_names", [])
    if filename in valid_filenames:
        return True

    valid_file_patterns = method_config.get("file_patterns", [])
    for pattern in valid_file_patterns:
        if re.fullmatch(pattern, filename):
            return True
    return False