Spaces:
Running
Running
File size: 14,175 Bytes
c272101 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 | #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
|