Projeto 1: Análise de Vendas

Código Python

import requests
import pandas as pd
import time
from io import BytesIO
from config import (
    AZURE_TENANT_ID,
    AZURE_CLIENT_ID,
    AZURE_CLIENT_SECRET,
    API_AUTH_ID,
    API_AUTH_PASSWORD,
    SHAREPOINT_UPLOAD_URL,
    API_TOKEN_URL,
    API_VENDAS_URL
)

# 1. Função para obter o token OAuth (Azure)
def obter_token_oauth():
    try:
        url = f'https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/v2.0/token'
        data = {
            'grant_type': 'client_credentials',
            'client_id': AZURE_CLIENT_ID,
            'client_secret': AZURE_CLIENT_SECRET,
            'scope': 'https://graph.microsoft.com/.default'
        }

        response = requests.post(url, data=data, timeout=10)
        response.raise_for_status()
        
        return response.json().get('access_token')
    
    except Exception as e:
        print(f'Erro ao obter token OAuth: {str(e)}')
        return None

# 2. Função para upload seguro no SharePoint
def upload_consolidado_para_sharepoint(output, token_oauth):
    try:
        headers = {
            'Authorization': f'Bearer {token_oauth}',
            'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
        }

        response = requests.put(
            SHAREPOINT_UPLOAD_URL,
            headers=headers,
            data=output.getvalue(),
            timeout=15
        )
        
        response.raise_for_status()
        print("Arquivo consolidado enviado com sucesso para o SharePoint.")
        return True
    
    except Exception as e:
        print(f"Erro no upload: {str(e)}")
        return False

# Função principal com tratamento de erros aprimorado
def main():
    try:
        # 3. Autenticação na API de Vendas
        body_token = {
            "id": API_AUTH_ID,
            "scope": "['DASH','DASH','PPR1.180610.011','1.0']",
            "password": API_AUTH_PASSWORD,
            "Role": "Admin"
        }

        # Primeira requisição com timeout
        response_token = requests.post(
            API_TOKEN_URL,
            json=body_token,
            timeout=15
        )
        response_token.raise_for_status()
        access_token = response_token.json().get("access_token")

        # Intervalo entre requisições
        time.sleep(2)

        # 4. Busca de dados de vendas
        vendas_params = {
            "user": API_AUTH_ID,
            "dataini": "11-01-2024",
            "datafim": "12-31-2024"
        }

        headers_vendas = {
            "Authorization": f"Bearer {access_token}",
            "User": str(API_AUTH_ID)
        }

        response_vendas = requests.get(
            API_VENDAS_URL,
            headers=headers_vendas,
            params=vendas_params,
            timeout=15
        )
        response_vendas.raise_for_status()

        # 5. Processamento dos dados
        vendas_data = response_vendas.json()
        linhas = []

        for loja in vendas_data:
            for pedido in loja.get("Pedidos", []):
                for item in pedido.get("Itens", []):
                    linha = {
                        "Loja": loja["Loja"],
                        "Nome": loja["Nome"],
                        "Total": loja["Total"],
                        "Pedido": pedido["Pedido"],
                        # ... (mantido o mesmo processamento de dados)
                    }
                    linhas.append(linha)

        df_itens = pd.DataFrame(linhas)

        # 6. Criação do arquivo em memória
        output_itens = BytesIO()
        with pd.ExcelWriter(output_itens, engine='xlsxwriter') as writer:
            df_itens.to_excel(writer, index=False, sheet_name='Itens')
        output_itens.seek(0)

        # 7. Upload seguro para SharePoint
        if token_oauth := obter_token_oauth():
            upload_consolidado_para_sharepoint(output_itens, token_oauth)

    except requests.exceptions.RequestException as e:
        print(f"Erro de comunicação: {str(e)}")
    except Exception as e:
        print(f"Erro inesperado: {str(e)}")
        

if __name__ == "__main__":
    main()
                

Fluxo de Dados