#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