Spaces:
Running
Running
| #Modulos | |
| import Google_Sheets | |
| #Bibliotecas externas ou internas | |
| import os | |
| import requests | |
| from qlik_sdk import AuthType, Config, Apps | |
| import pandas as pd | |
| import time | |
| import traceback # Nova biblioteca para rastrear erros linha por linha! | |
| # --- 1. CONFIGURAÇÕES --- | |
| TENANT_URL = os.environ.get("TENANT_URL") | |
| APP_ID = os.environ.get("APP_ID") | |
| CLIENT_ID = os.environ.get("CLIENT_ID") | |
| CLIENT_SECRET = os.environ.get("CLIENT_SECRET") | |
| # ========================================================= | |
| def extrair_dados_engine(): | |
| try: | |
| print("==========================================") | |
| print("Iniciando extração... Buscando um NOVO token M2M no Qlik...") | |
| # Fazemos o pedido oficial do Token DENTRO da função | |
| resposta = requests.post( | |
| f"{TENANT_URL}/oauth/token", | |
| json={ | |
| "client_id": CLIENT_ID, | |
| "client_secret": CLIENT_SECRET, | |
| "grant_type": "client_credentials" | |
| }, | |
| headers={"Accept": "application/json", "Content-Type": "application/json"} | |
| ) | |
| if resposta.status_code != 200: | |
| print(f"❌ Erro na autenticação com o Qlik: {resposta.text}") | |
| return None, None | |
| token_seguro = resposta.json().get("access_token") | |
| print("✔️ Token gerado com sucesso!") | |
| # --- CORREÇÃO 1: O TRUQUE DO RELÓGIO --- | |
| print("⏳ Aguardando 3 segundos para sincronia de relógio entre os servidores...") | |
| time.sleep(3) | |
| # --------------------------------------- | |
| # --- CORREÇÃO 2: A LIMPEZA DA URL --- | |
| # O .strip("/") garante que, mesmo que você tenha colado a URL no | |
| # Hugging Face com uma barra no final, o Python tira ela. | |
| url_limpa = TENANT_URL.strip("/") | |
| config = Config( | |
| host=url_limpa, | |
| auth_type=AuthType.APIKey, | |
| api_key=token_seguro | |
| ) | |
| apps = Apps(config) | |
| print("Conectando ao App do Qlik...") | |
| # --- NOVO: SISTEMA DE TENTATIVAS PARA ACORDAR O QLIK --- | |
| max_tentativas = 5 | |
| #Cria um loop que irá verificar se esta retornado memso depois de um tempo | |
| for tentativa in range(1, max_tentativas + 1): | |
| try: | |
| app = apps.get(APP_ID) | |
| with app.open(): | |
| print(f"App '{app.attributes.name}' aberto! Montando a tabela virtual...") | |
| campos_desejados = ["Chave_CC_PRO", "SAFRA", "UF", "CÓDIGO DO PRODUTO", "DERIVAÇÃO", "DESCRIÇÃO PRODUTO", | |
| "MARCA", "CUSTO ATUAL", "DATA VENC ATUAL","NOVO CUSTO","DATA DE VENC NOVA", | |
| "% BARTER", "QUANTIDADE"] | |
| limite_linhas = int(10000 / len(campos_desejados)) | |
| qDimensions = [{"qDef": {"qFieldDefs": [campo]}} for campo in campos_desejados] | |
| properties = { | |
| "qInfo": {"qType": "StraightTable"}, | |
| "qHyperCubeDef": { | |
| "qDimensions": qDimensions, | |
| "qInitialDataFetch": [ | |
| {"qTop": 0, "qLeft": 0, "qHeight": limite_linhas, "qWidth": len(campos_desejados)}] | |
| } | |
| } | |
| obj = app.create_session_object(properties) | |
| layout = obj.get_layout() | |
| total_linhas_qlik = layout.qHyperCube.qSize.qcy | |
| print(f"O Qlik encontrou um total de {total_linhas_qlik} linhas. Iniciando extração...") | |
| dados_limpos = [] | |
| for offset in range(0, total_linhas_qlik, limite_linhas): | |
| paginas = obj.get_hyper_cube_data( | |
| qPath="/qHyperCubeDef", | |
| qPages=[{"qTop": offset, "qLeft": 0, "qHeight": limite_linhas, "qWidth": len(campos_desejados)}] | |
| ) | |
| matriz_pagina = paginas[0].qMatrix | |
| for linha in matriz_pagina: | |
| registro = {} | |
| for i, campo in enumerate(campos_desejados): | |
| registro[campo] = linha[i].qText if linha[i].qText is not None else "" | |
| dados_limpos.append(registro) | |
| print(f"Extração do Qlik concluída! {len(dados_limpos)} registros baixados.") | |
| df = pd.DataFrame(dados_limpos) | |
| df["Valor de reposição?"] = "" | |
| df["Atualizar no Senior?"] = "" | |
| df["DUPLICAR"] = "" | |
| df["Salvar alteração"] = "" | |
| print("Puxando dados do Google Sheets...") | |
| hist = Google_Sheets.ler_historico() | |
| if not hist.empty: | |
| print("Lógica Passo a Passo: Verificando produtos novos e variações...") | |
| hist['DATA REGISTRADA'] = pd.to_datetime(hist['DATA REGISTRADA'], format="%d/%m/%Y %H:%M:%S") | |
| falsos_vazios = ["nan", "NaN", "None", "none", "null", "-"] | |
| for col in ['Chave_CC_PRO', 'SAFRA', 'UF']: | |
| df[col] = df[col].astype(str).str.strip().replace(falsos_vazios, "") | |
| hist[col] = hist[col].astype(str).str.strip().replace(falsos_vazios, "") | |
| # 2. DEDUPLICAÇÃO: Pega só o registro mais recente de cada Chave+Safra+UF | |
| hist_recente = hist.sort_values(by='DATA REGISTRADA', ascending=False).drop_duplicates( | |
| subset=['Chave_CC_PRO', 'SAFRA', 'UF'], keep='first') | |
| # Arruma os nomes do Sheets | |
| hist_recente = hist_recente.rename( | |
| columns={"BARTER %": "% BARTER", "DATA VENC NOVA": "DATA DE VENC NOVA"}) | |
| # O Qlik cria as colunas estritamente como texto. Se o Sheets trouxer um número (ex: 0), o Pandas dá erro. | |
| # Aqui nós forçamos tudo que vem do Sheets a virar texto puro antes do loop! | |
| colunas_valores = ['DATA DE VENC NOVA', 'NOVO CUSTO', '% BARTER',"DERIVAÇÃO",'QUANTIDADE'] | |
| for col_val in colunas_valores: | |
| if col_val in hist_recente.columns: | |
| # Converte para string e limpa os "nan" que o Pandas cria ao converter células vazias | |
| hist_recente[col_val] = hist_recente[col_val].astype(str).replace("nan", "") | |
| hist_recente[col_val] = hist_recente[col_val].replace("nan.0","") # Limpa casos de decimais vazios | |
| # Lista para guardar as variações que precisarem ser adicionadas no final | |
| linhas_para_adicionar = [] | |
| # ============================================================== | |
| # 3. A SUA LÓGICA: VERIFICAÇÃO LINHA A LINHA (VERSÃO BLINDADA) | |
| # ============================================================== | |
| # O arsenal anti-lixo do Qlik | |
| vazios = ["", " ", "-", "nan", "NaN", "None", "none", "null"] | |
| for index, linha_sht in hist_recente.iterrows(): | |
| # Garantimos que a chave de busca não tenha espaços invisíveis | |
| chave = str(linha_sht['Chave_CC_PRO']).strip() | |
| safra_sht = str(linha_sht['SAFRA']).strip() | |
| uf_sht = str(linha_sht['UF']).strip() | |
| # Procura essa chave no df do Qlik (também limpando os espaços) | |
| mascara_chave = df['Chave_CC_PRO'].astype(str).str.strip() == chave | |
| if mascara_chave.any(): | |
| # REGRA 1: Já existe essa EXATA variação? | |
| mascara_exata = mascara_chave & (df['SAFRA'].astype(str).str.strip() == safra_sht) & ( | |
| df['UF'].astype(str).str.strip() == uf_sht) | |
| if mascara_exata.any(): | |
| df.loc[mascara_exata, 'DATA DE VENC NOVA'] = linha_sht['DATA DE VENC NOVA'] | |
| df.loc[mascara_exata, 'NOVO CUSTO'] = linha_sht['NOVO CUSTO'] | |
| df.loc[mascara_exata, '% BARTER'] = linha_sht['% BARTER'] | |
| df.loc[mascara_exata, 'QUANTIDADE'] = linha_sht['QUANTIDADE'] | |
| else: | |
| # REGRA 2 BLINDADA: Verifica se a linha do Qlik contém qualquer tipo de "lixo" vazio | |
| # O .isin() olha pra nossa lista de vazios e dá o veredito final | |
| safra_qlik_limpa = df['SAFRA'].astype(str).str.strip() | |
| uf_qlik_limpa = df['UF'].astype(str).str.strip() | |
| mascara_vazia = mascara_chave & safra_qlik_limpa.isin(vazios) & uf_qlik_limpa.isin(vazios) | |
| print(f'Essa é a mascara as cahes que vai cair no if das vazias {mascara_vazia}') | |
| if mascara_vazia.any(): | |
| # SOBRESCREVE A LINHA VAZIA! | |
| idx_vazia = df[mascara_vazia].index[0] | |
| print(f'Index para ser rescrita:{idx_vazia}') | |
| df.loc[idx_vazia, 'SAFRA'] = safra_sht | |
| df.loc[idx_vazia, 'UF'] = uf_sht | |
| df.loc[idx_vazia, 'DERIVAÇÃO'] = f"00{linha_sht['DERIVAÇÃO']}" | |
| df.loc[idx_vazia, 'NOVO CUSTO'] = linha_sht['NOVO CUSTO'] | |
| df.loc[idx_vazia, '% BARTER'] = linha_sht['% BARTER'] | |
| df.loc[idx_vazia, 'QUANTIDADE'] = linha_sht['QUANTIDADE'] | |
| print(f'Index para ser rescrita:{linha_sht}') | |
| else: | |
| # REGRA 3: O PRODUTO DUPLICADO (Nova Variação) | |
| linha_nova = df[mascara_chave].iloc[0].copy() | |
| linha_nova['SAFRA'] = safra_sht | |
| linha_nova['UF'] = uf_sht | |
| linha_nova['DERIVAÇÃO'] = f"00{linha_sht['DERIVAÇÃO']}" | |
| linha_nova['QUANTIDADE'] = linha_sht['QUANTIDADE'] | |
| linha_nova['DATA DE VENC NOVA'] = linha_sht['DATA DE VENC NOVA'] | |
| linha_nova['NOVO CUSTO'] = linha_sht['NOVO CUSTO'] | |
| linha_nova['% BARTER'] = linha_sht['% BARTER'] | |
| linha_nova['QUANTIDADE'] = linha_sht['QUANTIDADE'] | |
| linha_nova['Valor de reposição?'] = "" | |
| linha_nova['Atualizar no Senior?'] = "" | |
| linha_nova['DUPLICAR'] = "" | |
| linha_nova['Salvar alteração'] = "" | |
| linhas_para_adicionar.append(linha_nova) | |
| else: | |
| # Se a chave do Google Sheets não existir de forma alguma no Qlik, | |
| # podemos rastrear aqui no terminal se houve erro de digitação | |
| print(f"⚠️ Atenção: A chave {chave} está no Sheets mas não foi encontrada no Qlik!") | |
| # 4. ADICIONA AS DUPLICADAS NO FINAL DA TABELA | |
| if linhas_para_adicionar: | |
| df_novas = pd.DataFrame(linhas_para_adicionar) | |
| df = pd.concat([df, df_novas], ignore_index=True) | |
| # 5. MAQUIAGEM PARA O FLET (Evita o erro de NaN na tela) | |
| df = df.fillna("") | |
| else: | |
| print("Planilha vazia. Criando colunas em branco.") | |
| df["DATA DE VENC NOVA"] = "" | |
| df["NOVO CUSTO"] = "" | |
| df["% BARTER"] = "" | |
| df["QUANTIDADE"] = "" | |
| ordem_colunas = df.columns.tolist() | |
| ordem_colunas.remove("DUPLICAR") | |
| ordem_colunas.insert(0, "DUPLICAR") | |
| df = df[ordem_colunas] | |
| print("Tudo pronto! Retornando os dados.") | |
| return df, hist | |
| except Exception as e: | |
| print("❌ ERRO FATAL DENTRO DA FUNÇÃO DE EXTRAÇÃO:") | |
| print(traceback.format_exc()) # Isso vai nos mostrar EXATAMENTE onde o código está quebrando! | |
| if tentativa < max_tentativas: | |
| time.sleep(10) | |
| else: | |
| # Só desiste se falhar as 5 vezes | |
| # Agora ele vai obrigatoriamente esperar e tentar de novo! | |
| print( | |
| f"⚠️ Qlik rejeitou a conexão. O servidor está acordando (Tentativa {tentativa}/5). Aguardando 10s...") | |
| raise e | |
| except Exception as e: | |
| print("❌ ERRO FATAL DENTRO DA FUNÇÃO DE EXTRAÇÃO:") | |
| print(traceback.format_exc()) # Isso vai nos mostrar EXATAMENTE onde o código está quebrando! | |
| return None, None | |