darlin.cunha/Scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
871738536aea8d9ede964b3548020176a6c5fc13
Scripts/draft_mar/draft_mar_v2.py
T

593 lines
23 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
import os
import sys
import json
import time
import socket
import requests
from dotenv import load_dotenv, set_key
sys.path.append(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "Banco"))
from db import get_connection
from requests.exceptions import RequestException
from datetime import datetime, timezone, timedelta
ENV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
load_dotenv(ENV_PATH)
# URL da nova API (build-draft-v2)
BASE_URL = "https://mar-api-gateway.prd.demanda-abastecimento.app.grupoboticario.com.br/orders-bff/api/draft/build-draft-v2"
PAGE_SIZE = 25
# API interna que guarda o token Bearer atualizado (renovado por outro processo)
TOKENS_API_URL = "https://api.grupoginseng.com.br/api/tokens"
# Token de autenticacao (Bearer) - expira em ~1h. Atualizado automaticamente via TOKENS_API_URL
# no inicio de cada execucao (funcao refresh_bearer_token). O valor no .env serve de fallback
# caso essa API esteja indisponivel.
BEARER_TOKEN = os.getenv("MAR_BEARER_TOKEN")
def build_authorization_header(token):
"""Garante o prefixo 'Bearer ' no header, aceitando token com ou sem prefixo ja incluso"""
if not token:
return None
token = token.strip()
return token if token.lower().startswith("bearer ") else f"Bearer {token}"
HEADERS = {
"accept": "*/*",
"accept-language": "pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7",
"authorization": build_authorization_header(BEARER_TOKEN),
"content-type": "application/json",
"origin": "https://abastecimentogb.grupoboticario.com.br",
"referer": "https://abastecimentogb.grupoboticario.com.br/",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36"
}
def refresh_bearer_token():
"""Busca o token Bearer mais recente em TOKENS_API_URL, salva no .env e atualiza HEADERS"""
global BEARER_TOKEN
response = requests.get(TOKENS_API_URL, timeout=15)
response.raise_for_status()
body = response.json()
registros = body.get("data", [])
if not registros:
raise ValueError("Resposta de TOKENS_API_URL nao contem registros em 'data'.")
registro_mais_recente = max(registros, key=lambda r: r.get("updated_at", ""))
token_value = registro_mais_recente.get("token")
if not token_value:
raise ValueError("Registro mais recente de TOKENS_API_URL nao contem o campo 'token'.")
set_key(ENV_PATH, "MAR_BEARER_TOKEN", token_value)
BEARER_TOKEN = token_value
HEADERS["authorization"] = build_authorization_header(BEARER_TOKEN)
return BEARER_TOKEN
# Lista de codigos das lojas
store_codes = [
"12522", "12817", "12818", "12820", "12823", "12824", "12826", "12828", "12829",
"12830", "12838", "13427", "14617", "19103", "20005", "20006", "20009", "20056",
"20057", "20441", "20858", "20968", "20969", "20970", "20986", "20988", "20989",
"20991", "20992", "20993", "20994", "20995", "20996", "20997", "20998", "20999",
"21000", "21001", "21068", "21277", "21278", "21296", "21375", "21381", "21383",
"21495", "21624", "21647", "22541", "3546", "4560", "5699", "910173", "910291",
"21007", "23703", "23475", "23156", "14668", "24253",
"24254", "24255", "24258", "24257", "24268", "24269", "24293", "23813", "24449", "24450", "24455", "24458", "24454", "24447", "24448", "24451", "24456", "24457", "24453", "24452"
]
# storeType da API v2 e definido pelo CANAL e pela MARCA ABR do PDV em base_pdvs:
# - PDV dedicado de marca propria (MARCA ABR = QDB - Quem Disse, Berenice?; EUD - Eudora; OUI - O.U.I,
# ex: PDVs 910173/910291) -> storeType = a propria sigla da marca; a API retorna o catalogo inteiro
# desse PDV sem precisar de filtro de marca (brandGroupCode None). Confirmado via DevTools so para QDB;
# EUD/OUI seguem por analogia (nenhum PDV assim existe em base_pdvs ainda).
# - Demais PDVs (marca BOT, vazio, etc.) -> storeType pelo CANAL: 'LJ' -> "BOT"; qualquer outro canal
# (VD, HIB, CD, ...) -> "VDF".
#
# Alem do PDV dedicado, BOT/EUD/OUI/QDB tambem podem ser vendidos DENTRO de um mesmo PDV comum, via o
# filtro filters.brandGroupCode (confirmado no DevTools, tela de filtro "Marcas" do MAR):
# - PDV tipo "BOT" (loja fisica, canal LJ): so vende marca BOT -> uma chamada com brandGroupCode ["BOT"].
# - PDV tipo "VDF" (VD/HIB/CD): vende as 4 marcas dentro do mesmo PDV -> uma chamada por marca
# (["BOT"], ["EUD"], ["OUI"], ["QDB"]), resultados somados na mesma loja/dia.
# - PDV dedicado de marca propria ("QDB"/"EUD"/"OUI"): uma unica chamada sem filtro (brandGroupCode None).
MARCAS_PROPRIAS = {"QDB", "EUD", "OUI"}
_store_type_cache = {}
def get_store_type(store_code, cursor):
"""Consulta CANAL/MARCA ABR do PDV em base_pdvs e retorna o storeType para a API v2."""
if store_code in _store_type_cache:
return _store_type_cache[store_code]
cursor.execute("SELECT CANAL, [MARCA ABR] FROM base_pdvs WHERE PDV = ?", (store_code,))
row = cursor.fetchone()
if not row or not row[0]:
raise ValueError(
f"Loja {store_code} nao encontrada (ou sem CANAL) na tabela base_pdvs."
)
canal = row[0].strip().upper()
marca_abr = (row[1] or "").strip().upper()
if marca_abr in MARCAS_PROPRIAS:
store_type = marca_abr
else:
store_type = "BOT" if canal == "LJ" else "VDF"
_store_type_cache[store_code] = store_type
return store_type
def get_brand_filters(store_type):
"""Lista de valores de filters.brandGroupCode a percorrer para um dado storeType."""
if store_type == "VDF":
return [["BOT"], ["EUD"], ["OUI"], ["QDB"]]
if store_type in MARCAS_PROPRIAS:
return [None]
return [["BOT"]]
def check_internet_connection():
"""Verifica se ha conexao com a internet"""
try:
socket.gethostbyname("google.com")
return True
except socket.error:
return False
def create_draft_v2_table(cursor, conn):
"""Cria a tabela Draft_mar_2026_v2 se ela nao existir"""
try:
cursor.execute("""
IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'Draft_mar_2026_v2' AND schema_id = SCHEMA_ID('dbo'))
BEGIN
CREATE TABLE dbo.Draft_mar_2026_v2 (
id INT IDENTITY(1,1) PRIMARY KEY,
date DATE DEFAULT CAST(GETDATE() AS DATE),
loja_id VARCHAR(50),
metadata_id VARCHAR(50),
metadata_savedAt VARCHAR(50),
metadata_isDraftReplace VARCHAR(10),
thirdToLastCycle VARCHAR(5),
thirdToLastCycleEud VARCHAR(5),
secondToLastCycle VARCHAR(5),
secondToLastCycleEud VARCHAR(5),
lastCycle VARCHAR(5),
lastCycleEud VARCHAR(5),
currentCycle VARCHAR(5),
currentCycleEud VARCHAR(5),
nextCycle VARCHAR(5),
nextCycleEud VARCHAR(5),
secondToNextCycle VARCHAR(5),
secondToNextCycleEud VARCHAR(5),
productId VARCHAR(50),
productIndex INT,
code VARCHAR(50),
description VARCHAR(255),
launch VARCHAR(50),
deactivation VARCHAR(50),
suggestionType VARCHAR(50),
discountPercent DECIMAL(18, 2),
criticalItem_isCritical VARCHAR(10),
criticalItem_blockedWallet VARCHAR(10),
criticalItem_dtProvidedRegularization VARCHAR(50),
sales_thirdToLastCycleSales INT,
sales_secondToLastCycleSales INT,
sales_lastCycleSales INT,
sales_currentCycleSales INT,
sales_nextCycleProjection INT,
sales_secondToNextCycleProjection INT,
stock_actual INT,
stock_inTransit INT,
stock_safetyStockQuantity INT,
stock_syncSafetyStock VARCHAR(10),
stock_codeMaterial VARCHAR(50),
cyclesPromotion NVARCHAR(MAX),
strategy_currentCycleStrategy_cycle VARCHAR(5),
strategy_currentCycleStrategy_isIAF VARCHAR(10),
strategy_currentCycleStrategy_isFoco VARCHAR(10),
strategy_nextCycleStrategy_cycle VARCHAR(5),
strategy_nextCycleStrategy_isIAF VARCHAR(10),
strategy_nextCycleStrategy_isFoco VARCHAR(10),
strategy_secondToNextCycleStrategy_cycle VARCHAR(5),
strategy_secondToNextCycleStrategy_isIAF VARCHAR(10),
strategy_secondToNextCycleStrategy_isFoco VARCHAR(10),
quantity INT,
analyzed VARCHAR(10),
quantityPerBox INT,
projectionSuggestionValue INT,
projectionValue INT,
launchType VARCHAR(50),
pendingOrder INT,
hasCoverage VARCHAR(10),
weekSuggestionValue INT,
salesCurve VARCHAR(50),
segmentedSalesCurve VARCHAR(50),
typeOfPromotion VARCHAR(255),
priceSellin DECIMAL(18, 2),
discountValue DECIMAL(18, 2),
promotionTypeValue VARCHAR(255),
targetOfPromotion VARCHAR(255),
restockSuggestion INT,
brandGroupCode VARCHAR(50),
isProductDeactivated VARCHAR(10),
status VARCHAR(50),
horizonInputLabel VARCHAR(50),
daysWithoutSales INT,
vlrStockPlanogram DECIMAL(18, 2),
historicSales NVARCHAR(MAX),
similar NVARCHAR(MAX),
comboItems NVARCHAR(MAX)
)
END
""")
conn.commit()
print("Tabela Draft_mar_2026_v2 verificada/criada com sucesso.")
except Exception as e:
print(f"Erro ao criar tabela Draft_mar_2026_v2: {e}")
conn.rollback()
raise
def to_json_text(value):
"""Serializa listas/objetos em texto JSON (ou None se vazio)"""
if value is None:
return None
return json.dumps(value, ensure_ascii=False)
def process_and_insert_data(store_code, response_data, cursor, conn):
"""Processa os produtos de uma pagina da API v2 e insere na tabela Draft_mar_2026_v2"""
metadata = response_data.get("metadata", {})
products = response_data.get("content", [])
if not products:
return 0
registros = []
for product in products:
critical_item = product.get("criticalItem", {}) or {}
sales = product.get("sales", {}) or {}
stock = product.get("stock", {}) or {}
strategy = product.get("strategy", {}) or {}
current_strategy = strategy.get("currentCycleStrategy", {}) or {}
next_strategy = strategy.get("nextCycleStrategy", {}) or {}
second_next_strategy = strategy.get("secondToNextCycleStrategy", {}) or {}
registro = (
store_code,
metadata.get("id"),
metadata.get("savedAt"),
metadata.get("isDraftReplace"),
metadata.get("thirdToLastCycle"),
metadata.get("thirdToLastCycleEud"),
metadata.get("secondToLastCycle"),
metadata.get("secondToLastCycleEud"),
metadata.get("lastCycle"),
metadata.get("lastCycleEud"),
metadata.get("currentCycle"),
metadata.get("currentCycleEud"),
metadata.get("nextCycle"),
metadata.get("nextCycleEud"),
metadata.get("secondToNextCycle"),
metadata.get("secondToNextCycleEud"),
product.get("id"),
product.get("index"),
product.get("code"),
product.get("description"),
product.get("launch"),
product.get("deactivation"),
product.get("suggestionType"),
product.get("discountPercent"),
critical_item.get("isCritical"),
critical_item.get("blockedWallet"),
critical_item.get("dtProvidedRegularization"),
sales.get("thirdToLastCycleSales"),
sales.get("secondToLastCycleSales"),
sales.get("lastCycleSales"),
sales.get("currentCycleSales"),
sales.get("nextCycleProjection"),
sales.get("secondToNextCycleProjection"),
stock.get("actual"),
stock.get("inTransit"),
stock.get("safetyStockQuantity"),
stock.get("syncSafetyStock"),
stock.get("codeMaterial"),
to_json_text(product.get("cyclesPromotion")),
current_strategy.get("cycle"),
current_strategy.get("IsIAF"),
current_strategy.get("IsFoco"),
next_strategy.get("cycle"),
next_strategy.get("IsIAF"),
next_strategy.get("IsFoco"),
second_next_strategy.get("cycle"),
second_next_strategy.get("IsIAF"),
second_next_strategy.get("IsFoco"),
product.get("quantity"),
product.get("analyzed"),
product.get("quantityPerBox"),
product.get("projectionSuggestionValue"),
product.get("projectionValue"),
product.get("launchType"),
product.get("pendingOrder"),
product.get("hasCoverage"),
product.get("weekSuggestionValue"),
product.get("salesCurve"),
product.get("segmentedSalesCurve"),
product.get("typeOfPromotion"),
product.get("priceSellin"),
product.get("discountValue"),
product.get("promotionTypeValue"),
product.get("targetOfPromotion"),
product.get("restockSuggestion"),
product.get("brandGroupCode"),
product.get("isProductDeactivated"),
product.get("status"),
product.get("horizonInputLabel"),
product.get("daysWithoutSales"),
product.get("vlrStockPlanogram"),
to_json_text(product.get("historicSales")),
to_json_text(product.get("similar")),
to_json_text(product.get("comboItems")),
)
registros.append(registro)
sql = """
INSERT INTO dbo.Draft_mar_2026_v2 (
loja_id,
metadata_id, metadata_savedAt, metadata_isDraftReplace,
thirdToLastCycle, thirdToLastCycleEud, secondToLastCycle, secondToLastCycleEud,
lastCycle, lastCycleEud, currentCycle, currentCycleEud,
nextCycle, nextCycleEud, secondToNextCycle, secondToNextCycleEud,
productId, productIndex, code, description, launch, deactivation, suggestionType, discountPercent,
criticalItem_isCritical, criticalItem_blockedWallet, criticalItem_dtProvidedRegularization,
sales_thirdToLastCycleSales, sales_secondToLastCycleSales, sales_lastCycleSales,
sales_currentCycleSales, sales_nextCycleProjection, sales_secondToNextCycleProjection,
stock_actual, stock_inTransit, stock_safetyStockQuantity, stock_syncSafetyStock, stock_codeMaterial,
cyclesPromotion,
strategy_currentCycleStrategy_cycle, strategy_currentCycleStrategy_isIAF, strategy_currentCycleStrategy_isFoco,
strategy_nextCycleStrategy_cycle, strategy_nextCycleStrategy_isIAF, strategy_nextCycleStrategy_isFoco,
strategy_secondToNextCycleStrategy_cycle, strategy_secondToNextCycleStrategy_isIAF, strategy_secondToNextCycleStrategy_isFoco,
quantity, analyzed, quantityPerBox, projectionSuggestionValue, projectionValue, launchType,
pendingOrder, hasCoverage, weekSuggestionValue, salesCurve, segmentedSalesCurve,
typeOfPromotion, priceSellin, discountValue, promotionTypeValue, targetOfPromotion,
restockSuggestion, brandGroupCode, isProductDeactivated, status, horizonInputLabel,
daysWithoutSales, vlrStockPlanogram,
historicSales, similar, comboItems
) VALUES (
?,
?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?,
?, ?, ?, ?, ?, ?, ?, ?,
?, ?, ?,
?, ?, ?,
?, ?, ?,
?, ?, ?, ?, ?,
?,
?, ?, ?,
?, ?, ?,
?, ?, ?,
?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?,
?, ?, ?, ?, ?,
?, ?, ?, ?, ?,
?, ?,
?, ?, ?
)
"""
cursor.executemany(sql, registros)
conn.commit()
return len(registros)
def fetch_draft_page(store_code, store_type, brand_group_code, page, cursor, conn):
"""Busca uma pagina do draft (loja + marca) e insere no banco. Retorna (inseridos, ultima_pagina)"""
params = {
"draftType": "SEM",
"page": page,
"sort": "description,asc",
}
payload = {
"storeCode": store_code,
"storeType": store_type,
"size": PAGE_SIZE,
"filters": {
"promotion": None,
"purchaseSuggestion": None,
"codeProductLine": None,
"brandGroupCode": brand_group_code,
"analyzed": None,
"isCritical": None,
"isFoco": None,
"isIaf": None,
"itemStatus": None,
"search": "",
"subCategoryCode": None,
},
"purchasingHorizon": "WEEK",
}
response = requests.post(BASE_URL, headers=HEADERS, params=params, json=payload)
if response.status_code in (401, 403):
print(f"Token expirado/invalido (status {response.status_code}) na loja {store_code}. Renovando token...")
refresh_bearer_token()
response = requests.post(BASE_URL, headers=HEADERS, params=params, json=payload)
response.raise_for_status()
response_data = response.json()
inseridos = process_and_insert_data(store_code, response_data, cursor, conn)
return inseridos, response_data.get("last", True)
def fetch_store_draft(store_code, cursor, conn):
"""Busca o draft de uma loja (todas as marcas aplicaveis e todas as paginas) e insere no banco"""
store_type = get_store_type(store_code, cursor)
brand_filters = get_brand_filters(store_type)
# Remove apenas os registros de HOJE dessa loja (permite reprocessar no mesmo dia sem duplicar),
# preservando o historico de dias anteriores.
delete_sql = "DELETE FROM dbo.Draft_mar_2026_v2 WHERE loja_id = ? AND date = CAST(GETDATE() AS DATE)"
cursor.execute(delete_sql, (store_code,))
conn.commit()
total_inseridos = 0
for brand_group_code in brand_filters:
page = 0
while True:
inseridos, ultima_pagina = fetch_draft_page(
store_code, store_type, brand_group_code, page, cursor, conn
)
total_inseridos += inseridos
if ultima_pagina:
break
page += 1
return total_inseridos
def fetch_and_insert_data(store_code, index, total, cursor, conn, retry_count=1):
"""Baixa os dados da API v2 e insere diretamente no banco, com retentativa"""
attempt = 0
while attempt < retry_count:
try:
num_produtos = fetch_store_draft(store_code, cursor, conn)
print(f"{index}/{total}: Loja {store_code} processada com sucesso ({num_produtos} produtos inseridos).")
return True
except RequestException as e:
print(f"{index}/{total}: Erro na requisicao para a loja {store_code}: {e}")
except ValueError as e:
print(f"{index}/{total}: {e}")
return False
except Exception as e:
print(f"{index}/{total}: Erro inesperado para a loja {store_code}: {e}")
attempt += 1
print(f"{index}/{total}: Tentativa {attempt} de {retry_count} para a loja {store_code}.")
time.sleep(5)
print(f"{index}/{total}: Falha ao processar a loja {store_code} apos {retry_count} tentativas.")
return False
def update_api_status(status):
"""Atualiza o status da API"""
try:
url_status = "https://api.grupoginseng.com.br/api/status/1"
sao_paulo_offset = timedelta(hours=-3)
current_datetime = datetime.now(timezone(sao_paulo_offset)).strftime("%Y-%m-%d %H:%M:%S")
payload = {
"STATUS": status,
"DATA": current_datetime
}
headers_status = {"Content-Type": "application/json"}
response = requests.put(url_status, json=payload, headers=headers_status)
print("Hora enviada:", current_datetime)
print("Status Code:", response.status_code)
print("Response Body:", response.text)
except Exception as e:
print(f"Erro ao atualizar status da API: {e}")
def main():
"""Funcao principal"""
if not check_internet_connection():
print("Sem conexao com a internet. Verifique sua conexao e tente novamente.")
update_api_status("FAIL")
return
try:
refresh_bearer_token()
print("Token Bearer atualizado automaticamente via TOKENS_API_URL.")
except Exception as e:
print(f"Nao foi possivel atualizar o token automaticamente ({e}). Usando o token atual do .env, se houver.")
if not BEARER_TOKEN:
print("MAR_BEARER_TOKEN nao definido no .env. Atualize o token antes de rodar o script.")
update_api_status("FAIL")
return
conn = get_connection()
cursor = conn.cursor()
try:
create_draft_v2_table(cursor, conn)
failed_stores = []
total = len(store_codes)
for index, store_code in enumerate(store_codes, start=1):
success = fetch_and_insert_data(store_code, index, total, cursor, conn)
if not success:
failed_stores.append(store_code)
if failed_stores:
print("\nTentando novamente as lojas que falharam.")
retry_count = 1
for attempt in range(1, retry_count + 1):
retry_failed_stores = []
for store_code in failed_stores:
index = store_codes.index(store_code) + 1
success = fetch_and_insert_data(store_code, index, total, cursor, conn)
if not success:
retry_failed_stores.append(store_code)
failed_stores = retry_failed_stores
if not failed_stores:
print("\nTodos os arquivos foram processados com sucesso.")
break
if failed_stores:
print(f"Tentativa {attempt} de {retry_count} para as lojas que falharam.")
time.sleep(10)
if failed_stores:
print(f"\nNao foi possivel processar algumas lojas: {failed_stores}")
else:
print("\nTodos os arquivos foram processados com sucesso.")
update_api_status("FAIL" if failed_stores else "OK")
except Exception as e:
print(f"Erro durante a execucao: {e}")
update_api_status("FAIL")
finally:
cursor.close()
conn.close()
print("\nProcesso finalizado.")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink