import requests import pyodbc import json from datetime import datetime, timedelta import logging from concurrent.futures import ThreadPoolExecutor, as_completed import time # Configuração de logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # ===== CONFIGURAÇÕES DO SCRIPT ===== # Usar data do dia anterior automaticamente USE_YESTERDAY = True # True: busca apenas o dia anterior | False: usa intervalo de datas abaixo # Intervalo de datas para buscar invoices (formato: YYYY-MM-DD) # Usado apenas quando USE_YESTERDAY = False START_DATE = "2025-06-01" # Data inicial END_DATE = "2025-12-17" # Data final (inclusiva) # =================================== class RGBFiscalInvoicesExtractor: def __init__(self): # Configurações da API self.api_url = "https://api.grupoboticario.com.br/global/v1/franchising/gb-stores-data/fiscal/invoices" # Configurações do banco de dados self.driver = "ODBC Driver 17 for SQL Server" self.connection_string = ( f'DRIVER={{{self.driver}}};' 'SERVER=10.77.77.10;' 'DATABASE=GINSENG;' 'UID=supginseng;' 'PWD=Ginseng@;' 'PORT=1433;' 'TrustServerCertificate=yes;' 'Encrypt=yes' ) def get_bearer_token_from_db(self): """ Busca o token de autenticação na tabela rgb_token """ connection = None cursor = None try: logger.info("Conectando ao banco de dados para buscar token...") connection = pyodbc.connect(self.connection_string) cursor = connection.cursor() # Query para buscar o token mais recente query = "SELECT TOP 1 token FROM [GINSENG].[dbo].[rgb_token] ORDER BY updatedAt DESC" cursor.execute(query) result = cursor.fetchone() if result: token = result[0] logger.info("Token recuperado com sucesso do banco de dados") return token else: logger.error("Nenhum token encontrado na tabela rgb_token") raise Exception("Token não encontrado no banco de dados") except pyodbc.Error as e: logger.error(f"Erro ao conectar ao banco de dados: {e}") raise finally: if cursor: cursor.close() if connection: connection.close() def fetch_page(self, headers, updated_at_date, start, count, page_num, max_retries=3): """ Faz uma única requisição para buscar uma página específica Implementa retry automático com backoff exponencial para erros 429 """ params = { 'invoice.updatedAt': updated_at_date, 'start': start, 'count': count } for attempt in range(max_retries): try: if attempt > 0: logger.info(f"Tentativa {attempt + 1}/{max_retries} - Página {page_num} (start: {start})") else: logger.info(f"Fazendo requisição - Página {page_num} (start: {start}, count: {count})") response = requests.get(self.api_url, headers=headers, params=params, timeout=60) response.raise_for_status() data = response.json() items = data.get('items', []) logger.info(f"Página {page_num}: recebidos {len(items)} registros") return { 'page': page_num, 'start': start, 'items': items, 'total': data.get('total', 0) } except requests.exceptions.HTTPError as e: if e.response.status_code == 429: # Too Many Requests if attempt < max_retries - 1: # Backoff exponencial: 2s, 4s, 8s wait_time = 2 ** (attempt + 1) logger.warning(f"Erro 429 na página {page_num}. Aguardando {wait_time}s antes de tentar novamente...") time.sleep(wait_time) continue else: logger.error(f"Erro 429 na página {page_num} após {max_retries} tentativas") raise else: # Outros erros HTTP, não tenta novamente raise except Exception as e: # Outros erros, não tenta novamente raise def get_fiscal_invoices_data(self, updated_at_date="2025-09-27"): """ Extrai dados da API de invoices fiscais do Grupo Boticário Implementa paginação PARALELA para buscar todos os registros rapidamente Faz até 5 requisições simultâneas """ try: # Buscar token do banco de dados bearer_token = self.get_bearer_token_from_db() headers = { 'Authorization': f'Bearer {bearer_token}', 'Content-Type': 'application/json' } count = 25 # Tamanho da página (padrão da API) max_workers = 3 # Número de requisições paralelas (reduzido para evitar rate limit) logger.info(f"Iniciando busca PARALELA na API de invoices fiscais - Data: {updated_at_date}") logger.info(f"Configuração: {max_workers} requisições simultâneas, {count} registros por página") logger.info("Sistema de retry automático ativado para erros 429 (Too Many Requests)") # Primeira requisição para descobrir o total de registros logger.info("Fazendo requisição inicial para descobrir total de registros...") first_response = self.fetch_page(headers, updated_at_date, 0, count, 1) total_records = first_response['total'] all_invoices = first_response['items'] logger.info(f"Total de registros disponíveis: {total_records}") if total_records <= count: # Se tem apenas uma página, retorna direto logger.info("Apenas uma página encontrada, busca concluída") return { 'start': 0, 'count': len(all_invoices), 'total': total_records, 'items': all_invoices } # Calcular quantas páginas faltam buscar total_pages = (total_records + count - 1) // count # Arredonda para cima remaining_pages = [] for page_num in range(2, total_pages + 1): start = (page_num - 1) * count remaining_pages.append((start, page_num)) logger.info(f"Total de páginas: {total_pages}. Buscando {len(remaining_pages)} páginas restantes em paralelo...") failed_pages = [] # Buscar páginas restantes em paralelo with ThreadPoolExecutor(max_workers=max_workers) as executor: # Criar futures para todas as páginas futures = { executor.submit(self.fetch_page, headers, updated_at_date, start, count, page_num): (start, page_num) for start, page_num in remaining_pages } # Coletar resultados conforme completam for future in as_completed(futures): start, page_num = futures[future] try: result = future.result() all_invoices.extend(result['items']) except Exception as e: logger.error(f"Erro ao buscar página {page_num}: {e}") failed_pages.append((start, page_num)) # Tentar páginas que falharam novamente (sequencialmente) if failed_pages: logger.warning(f"Tentando novamente {len(failed_pages)} página(s) que falharam...") for start, page_num in failed_pages: try: logger.info(f"Retry sequencial - Página {page_num}") time.sleep(2) # Espera 2s entre tentativas result = self.fetch_page(headers, updated_at_date, start, count, page_num) all_invoices.extend(result['items']) logger.info(f"Página {page_num} recuperada com sucesso!") except Exception as e: logger.error(f"Falha definitiva na página {page_num}: {e}") logger.info(f"Busca paginada paralela concluída. Total de registros coletados: {len(all_invoices)}") # Retornar dados no mesmo formato da resposta original final_data = { 'start': 0, 'count': len(all_invoices), 'total': total_records, 'items': all_invoices } return final_data except requests.exceptions.RequestException as e: logger.error(f"Erro na requisição da API: {e}") if hasattr(e, 'response') and e.response is not None: logger.error(f"Status Code: {e.response.status_code}") logger.error(f"Response: {e.response.text}") raise except json.JSONDecodeError as e: logger.error(f"Erro ao decodificar JSON: {e}") raise def connect_database(self): """ Conecta ao banco de dados SQL Server """ try: logger.info("Conectando ao banco de dados...") connection = pyodbc.connect(self.connection_string) logger.info("Conexão estabelecida com sucesso") return connection except pyodbc.Error as e: logger.error(f"Erro ao conectar ao banco de dados: {e}") raise def parse_datetime(self, date_string): """ Converte string de data para formato datetime """ if not date_string: return None try: # Remove o timezone offset e converte if date_string.endswith('-03:00') or date_string.endswith('+00:00'): date_string = date_string[:-6] elif 'T' in date_string and date_string.endswith('.000'): date_string = date_string[:-4] # Tenta diferentes formatos de data formats = [ '%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d' ] for fmt in formats: try: return datetime.strptime(date_string, fmt) except ValueError: continue logger.warning(f"Não foi possível converter a data: {date_string}") return None except Exception as e: logger.error(f"Erro ao converter data {date_string}: {e}") return None def insert_invoice_item(self, cursor, invoice_data, item_data=None): """ Insere um item de invoice na tabela rgb_fiscal_invoices Se item_data for None, insere apenas os dados da invoice (sem itens) """ insert_query = """ INSERT INTO [GINSENG].[dbo].[rgb_fiscal_invoices] ([invoiceId], [storeId], [stockLocationId], [fiscalOperationId], [fiscalOperationDescription], [supplierId], [supplierName], [clientId], [clientName], [emissionDate], [operationDate], [dateOfPassageOnFiscalPost], [deletedAt], [updatedAt], [invoiceNumber], [serie], [key], [buyerEmployeeId], [emitterEmployeeId], [cfop], [operationType], [shippingType], [paymentCondition], [fiscalDocumentType], [modality], [updateCost], [updateStock], [composeABC], [situation], [observation], [importType], [classification], [financeGenerationType], [shippingValue], [otherExpensesValue], [insuranceValue], [discountValue], [totalItemsValue], [documentValue], [valorDoDAE], [baseDeCalculoDoICMS], [valorDoICMS], [baseDeCalculoDoICMSSubstituicaoTributaria], [valorDoICMSSubstituicaoTributaria], [valorDoIPI], [valorDoPIS], [valorDoCOFINS], [valorDoICMSDesonerado], [baseDeCalculoFecop], [valorFecop], [baseDeCalculoFecopSubstituicaoTributaria], [valorFecopSubstituicaoTributaria], [itemId], [itemProductId], [itemSequential], [itemOrderNumber], [itemOrderItemSequencial], [itemComposeTotal], [itemCfop], [itemUnitType], [itemQuantityOfItensOnUnit], [itemQuantity], [itemCompleteQuantity], [itemUnitValue], [itemDiscountInputType], [itemUntaxedDiscountValue], [itemTaxedDiscountValue], [itemDiscountPercentage], [itemShippingInputType], [itemShippingValue], [itemShippingPercentage], [itemInsuranceInputType], [itemInsuranceValue], [itemInsurancePercentage], [itemOtherExpensesInputValue], [itemOtherExpensesValue], [itemOtherExpensesPercentage], [itemTotalValue], [itemTaxedPercentage], [itemProductCost], [itemFiscalCost], [itemAverageCost], [itemTipoDeEntradaDAE], [itemValorDoDAE], [itemPercentualDoDAE], [itemFiscalSituationId], [itemCsosn], [itemOutrasDespesasCompoeBaseDeCalculoIcms], [itemTributacao], [itemNcm], [itemCest], [itemAliquotaNacional], [itemAliquotaImportado], [itemAliquotaEstadual], [itemAliquotaMunicipal], [itemModalidadeDaBaseDeCalculo], [itemPercentualICMSDeCompra], [itemValorDoICMS], [itemValorDoICMSNoSimples], [itemBaseDeCalculoDoICMS], [itemBaseDeCalculoDoICMSComSubstituicaoTributaria], [itemAliquotaDoICMSComSubstituicaoTributaria], [itemValorDoICMSComSubstituicaoTributaria], [itemPercentualDeAgregacao], [itemPercentualDeReducaoDASubstituicaoTributaria], [itemAliquotaDoICMS], [itemAliquotaDoICMSDeVenda], [itemAliquotaDoICMSAntecipado], [itemValorDoICMSAntecipado], [itemAliquotaNoSimples], [itemCstDoIPI], [itemBaseDeCalculoDoIPI], [itemAliquotaDoIPI], [itemTipoDeEntradaIPI], [itemValorDoIPI], [itemPercentualDoIPI], [itemBaseDeCalculoDoPIS], [itemAliquotaDoPIS], [itemValorDoPIS], [itemBaseDeCalculoDoCOFINS], [itemAliquotaDoCOFINS], [itemValorDoCOFINS], [itemCodigoNaturezaDeImpostoFederal], [itemBaseDeCalculoDoFecop], [itemAliquotaDoFecop], [itemValorDoFecop], [itemBaseDeCalculoDoFecopSubstituto], [itemAliquotaDoFecopSubstituto], [itemValorDoFecopSubstituto], [itemValorDoICMSDesonerado], [itemMotivoDesoneracao], [itemCodigoBeneficioFiscal], [itemPercentualDiferimento], [itemValorICMSDiferimento]) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ # Preparar os dados da invoice (campos principais) invoice_values = [ invoice_data.get('id'), # invoiceId invoice_data.get('storeId'), invoice_data.get('stockLocationId'), invoice_data.get('fiscalOperationId'), invoice_data.get('fiscalOperationDescription'), invoice_data.get('supplierId'), invoice_data.get('supplierName'), invoice_data.get('clientId'), invoice_data.get('clientName'), self.parse_datetime(invoice_data.get('emissionDate')), self.parse_datetime(invoice_data.get('operationDate')), self.parse_datetime(invoice_data.get('dateOfPassageOnFiscalPost')), self.parse_datetime(invoice_data.get('deletedAt')), self.parse_datetime(invoice_data.get('updatedAt')), invoice_data.get('invoiceNumber'), invoice_data.get('serie'), invoice_data.get('key'), invoice_data.get('buyerEmployeeId'), invoice_data.get('emitterEmployeeId'), invoice_data.get('cfop'), invoice_data.get('operationType'), invoice_data.get('shippingType'), invoice_data.get('paymentCondition'), invoice_data.get('fiscalDocumentType'), invoice_data.get('modality'), invoice_data.get('updateCost'), invoice_data.get('updateStock'), invoice_data.get('composeABC'), invoice_data.get('situation'), invoice_data.get('observation'), invoice_data.get('importType'), invoice_data.get('classification'), invoice_data.get('financeGenerationType'), invoice_data.get('shippingValue'), invoice_data.get('otherExpensesValue'), invoice_data.get('insuranceValue'), invoice_data.get('discountValue'), invoice_data.get('totalItemsValue'), invoice_data.get('documentValue'), invoice_data.get('valorDoDAE'), invoice_data.get('baseDeCalculoDoICMS'), invoice_data.get('valorDoICMS'), invoice_data.get('baseDeCalculoDoICMSSubstituicaoTributaria'), invoice_data.get('valorDoICMSSubstituicaoTributaria'), invoice_data.get('valorDoIPI'), invoice_data.get('valorDoPIS'), invoice_data.get('valorDoCOFINS'), invoice_data.get('valorDoICMSDesonerado'), invoice_data.get('baseDeCalculoFecop'), invoice_data.get('valorFecop'), invoice_data.get('baseDeCalculoFecopSubstituicaoTributaria'), invoice_data.get('valorFecopSubstituicaoTributaria') ] # Preparar os dados do item (se existir) if item_data: item_values = [ item_data.get('id'), # itemId item_data.get('productId'), # itemProductId item_data.get('sequential'), # itemSequential item_data.get('orderNumber'), # itemOrderNumber item_data.get('orderItemSequencial'), # itemOrderItemSequencial item_data.get('composeTotal'), # itemComposeTotal item_data.get('cfop'), # itemCfop item_data.get('unitType'), # itemUnitType item_data.get('quantityOfItensOnUnit'), # itemQuantityOfItensOnUnit item_data.get('quantity'), # itemQuantity item_data.get('completeQuantity'), # itemCompleteQuantity item_data.get('unitValue'), # itemUnitValue item_data.get('discountInputType'), # itemDiscountInputType item_data.get('untaxedDiscountValue'), # itemUntaxedDiscountValue item_data.get('taxedDiscountValue'), # itemTaxedDiscountValue item_data.get('discountPercentage'), # itemDiscountPercentage item_data.get('shippingInputType'), # itemShippingInputType item_data.get('shippingValue'), # itemShippingValue item_data.get('shippingPercentage'), # itemShippingPercentage item_data.get('insuranceInputType'), # itemInsuranceInputType item_data.get('insuranceValue'), # itemInsuranceValue item_data.get('insurancePercentage'), # itemInsurancePercentage item_data.get('otherExpensesInputValue'), # itemOtherExpensesInputValue item_data.get('otherExpensesValue'), # itemOtherExpensesValue item_data.get('otherExpensesPercentage'), # itemOtherExpensesPercentage item_data.get('totalValue'), # itemTotalValue item_data.get('taxedPercentage'), # itemTaxedPercentage item_data.get('productCost'), # itemProductCost item_data.get('fiscalCost'), # itemFiscalCost item_data.get('averageCost'), # itemAverageCost item_data.get('tipoDeEntradaDAE'), # itemTipoDeEntradaDAE item_data.get('valorDoDAE'), # itemValorDoDAE item_data.get('percentualDoDAE'), # itemPercentualDoDAE item_data.get('fiscalSituationId'), # itemFiscalSituationId item_data.get('csosn'), # itemCsosn item_data.get('outrasDespesasCompoeBaseDeCalculoIcms'), # itemOutrasDespesasCompoeBaseDeCalculoIcms item_data.get('tributacao'), # itemTributacao item_data.get('ncm'), # itemNcm item_data.get('cest'), # itemCest item_data.get('aliquotaNacional'), # itemAliquotaNacional item_data.get('aliquotaImportado'), # itemAliquotaImportado item_data.get('aliquotaEstadual'), # itemAliquotaEstadual item_data.get('aliquotaMunicipal'), # itemAliquotaMunicipal item_data.get('modalidadeDaBaseDeCalculo'), # itemModalidadeDaBaseDeCalculo item_data.get('percentualICMSDeCompra'), # itemPercentualICMSDeCompra item_data.get('valorDoICMS'), # itemValorDoICMS item_data.get('valorDoICMSNoSimples'), # itemValorDoICMSNoSimples item_data.get('baseDeCalculoDoICMS'), # itemBaseDeCalculoDoICMS item_data.get('baseDeCalculoDoICMSComSubstituicaoTributaria'), # itemBaseDeCalculoDoICMSComSubstituicaoTributaria item_data.get('aliquotaDoICMSComSubstituicaoTributaria'), # itemAliquotaDoICMSComSubstituicaoTributaria item_data.get('valorDoICMSComSubstituicaoTributaria'), # itemValorDoICMSComSubstituicaoTributaria item_data.get('percentualDeAgregacao'), # itemPercentualDeAgregacao item_data.get('percentualDeReducaoDASubstituicaoTributaria'), # itemPercentualDeReducaoDASubstituicaoTributaria item_data.get('aliquotaDoICMS'), # itemAliquotaDoICMS item_data.get('aliquotaDoICMSDeVenda'), # itemAliquotaDoICMSDeVenda item_data.get('aliquotaDoICMSAntecipado'), # itemAliquotaDoICMSAntecipado item_data.get('valorDoICMSAntecipado'), # itemValorDoICMSAntecipado item_data.get('aliquotaNoSimples'), # itemAliquotaNoSimples item_data.get('cstDoIPI'), # itemCstDoIPI item_data.get('baseDeCalculoDoIPI'), # itemBaseDeCalculoDoIPI item_data.get('aliquotaDoIPI'), # itemAliquotaDoIPI item_data.get('tipoDeEntradaIPI'), # itemTipoDeEntradaIPI item_data.get('valorDoIPI'), # itemValorDoIPI item_data.get('percentualDoIPI'), # itemPercentualDoIPI item_data.get('baseDeCalculoDoPIS'), # itemBaseDeCalculoDoPIS item_data.get('aliquotaDoPIS'), # itemAliquotaDoPIS item_data.get('valorDoPIS'), # itemValorDoPIS item_data.get('baseDeCalculoDoCOFINS'), # itemBaseDeCalculoDoCOFINS item_data.get('aliquotaDoCOFINS'), # itemAliquotaDoCOFINS item_data.get('valorDoCOFINS'), # itemValorDoCOFINS item_data.get('codigoNaturezaDeImpostoFederal'), # itemCodigoNaturezaDeImpostoFederal item_data.get('baseDeCalculoDoFecop'), # itemBaseDeCalculoDoFecop item_data.get('aliquotaDoFecop'), # itemAliquotaDoFecop item_data.get('valorDoFecop'), # itemValorDoFecop item_data.get('baseDeCalculoDoFecopSubstituto'), # itemBaseDeCalculoDoFecopSubstituto item_data.get('aliquotaDoFecopSubstituto'), # itemAliquotaDoFecopSubstituto item_data.get('valorDoFecopSubstituto'), # itemValorDoFecopSubstituto item_data.get('valorDoICMSDesonerado'), # itemValorDoICMSDesonerado item_data.get('motivoDesoneracao'), # itemMotivoDesoneracao item_data.get('codigoBeneficioFiscal'), # itemCodigoBeneficioFiscal item_data.get('percentualDiferimento'), # itemPercentualDiferimento item_data.get('valorICMSDiferimento') # itemValorICMSDiferimento ] else: # Se não há item, preencher com None item_values = [None] * 82 # 82 campos de item # Combinar valores da invoice e do item all_values = invoice_values + item_values try: cursor.execute(insert_query, all_values) logger.info(f"Invoice/Item inserido: Invoice ID {invoice_data.get('id')}") return True except pyodbc.Error as e: logger.error(f"Erro ao inserir invoice ID {invoice_data.get('id')}: {e}") return False def prepare_invoice_values(self, invoice_data, item_data=None): """ Prepara os valores para inserção sem executar o INSERT Retorna a tupla de valores pronta para executemany """ # Preparar os dados da invoice (campos principais) invoice_values = [ invoice_data.get('id'), # invoiceId invoice_data.get('storeId'), invoice_data.get('stockLocationId'), invoice_data.get('fiscalOperationId'), invoice_data.get('fiscalOperationDescription'), invoice_data.get('supplierId'), invoice_data.get('supplierName'), invoice_data.get('clientId'), invoice_data.get('clientName'), self.parse_datetime(invoice_data.get('emissionDate')), self.parse_datetime(invoice_data.get('operationDate')), self.parse_datetime(invoice_data.get('dateOfPassageOnFiscalPost')), self.parse_datetime(invoice_data.get('deletedAt')), self.parse_datetime(invoice_data.get('updatedAt')), invoice_data.get('invoiceNumber'), invoice_data.get('serie'), invoice_data.get('key'), invoice_data.get('buyerEmployeeId'), invoice_data.get('emitterEmployeeId'), invoice_data.get('cfop'), invoice_data.get('operationType'), invoice_data.get('shippingType'), invoice_data.get('paymentCondition'), invoice_data.get('fiscalDocumentType'), invoice_data.get('modality'), invoice_data.get('updateCost'), invoice_data.get('updateStock'), invoice_data.get('composeABC'), invoice_data.get('situation'), invoice_data.get('observation'), invoice_data.get('importType'), invoice_data.get('classification'), invoice_data.get('financeGenerationType'), invoice_data.get('shippingValue'), invoice_data.get('otherExpensesValue'), invoice_data.get('insuranceValue'), invoice_data.get('discountValue'), invoice_data.get('totalItemsValue'), invoice_data.get('documentValue'), invoice_data.get('valorDoDAE'), invoice_data.get('baseDeCalculoDoICMS'), invoice_data.get('valorDoICMS'), invoice_data.get('baseDeCalculoDoICMSSubstituicaoTributaria'), invoice_data.get('valorDoICMSSubstituicaoTributaria'), invoice_data.get('valorDoIPI'), invoice_data.get('valorDoPIS'), invoice_data.get('valorDoCOFINS'), invoice_data.get('valorDoICMSDesonerado'), invoice_data.get('baseDeCalculoFecop'), invoice_data.get('valorFecop'), invoice_data.get('baseDeCalculoFecopSubstituicaoTributaria'), invoice_data.get('valorFecopSubstituicaoTributaria') ] # Preparar os dados do item (se existir) if item_data: item_values = [ item_data.get('id'), item_data.get('productId'), item_data.get('sequential'), item_data.get('orderNumber'), item_data.get('orderItemSequencial'), item_data.get('composeTotal'), item_data.get('cfop'), item_data.get('unitType'), item_data.get('quantityOfItensOnUnit'), item_data.get('quantity'), item_data.get('completeQuantity'), item_data.get('unitValue'), item_data.get('discountInputType'), item_data.get('untaxedDiscountValue'), item_data.get('taxedDiscountValue'), item_data.get('discountPercentage'), item_data.get('shippingInputType'), item_data.get('shippingValue'), item_data.get('shippingPercentage'), item_data.get('insuranceInputType'), item_data.get('insuranceValue'), item_data.get('insurancePercentage'), item_data.get('otherExpensesInputValue'), item_data.get('otherExpensesValue'), item_data.get('otherExpensesPercentage'), item_data.get('totalValue'), item_data.get('taxedPercentage'), item_data.get('productCost'), item_data.get('fiscalCost'), item_data.get('averageCost'), item_data.get('tipoDeEntradaDAE'), item_data.get('valorDoDAE'), item_data.get('percentualDoDAE'), item_data.get('fiscalSituationId'), item_data.get('csosn'), item_data.get('outrasDespesasCompoeBaseDeCalculoIcms'), item_data.get('tributacao'), item_data.get('ncm'), item_data.get('cest'), item_data.get('aliquotaNacional'), item_data.get('aliquotaImportado'), item_data.get('aliquotaEstadual'), item_data.get('aliquotaMunicipal'), item_data.get('modalidadeDaBaseDeCalculo'), item_data.get('percentualICMSDeCompra'), item_data.get('valorDoICMS'), item_data.get('valorDoICMSNoSimples'), item_data.get('baseDeCalculoDoICMS'), item_data.get('baseDeCalculoDoICMSComSubstituicaoTributaria'), item_data.get('aliquotaDoICMSComSubstituicaoTributaria'), item_data.get('valorDoICMSComSubstituicaoTributaria'), item_data.get('percentualDeAgregacao'), item_data.get('percentualDeReducaoDASubstituicaoTributaria'), item_data.get('aliquotaDoICMS'), item_data.get('aliquotaDoICMSDeVenda'), item_data.get('aliquotaDoICMSAntecipado'), item_data.get('valorDoICMSAntecipado'), item_data.get('aliquotaNoSimples'), item_data.get('cstDoIPI'), item_data.get('baseDeCalculoDoIPI'), item_data.get('aliquotaDoIPI'), item_data.get('tipoDeEntradaIPI'), item_data.get('valorDoIPI'), item_data.get('percentualDoIPI'), item_data.get('baseDeCalculoDoPIS'), item_data.get('aliquotaDoPIS'), item_data.get('valorDoPIS'), item_data.get('baseDeCalculoDoCOFINS'), item_data.get('aliquotaDoCOFINS'), item_data.get('valorDoCOFINS'), item_data.get('codigoNaturezaDeImpostoFederal'), item_data.get('baseDeCalculoDoFecop'), item_data.get('aliquotaDoFecop'), item_data.get('valorDoFecop'), item_data.get('baseDeCalculoDoFecopSubstituto'), item_data.get('aliquotaDoFecopSubstituto'), item_data.get('valorDoFecopSubstituto'), item_data.get('valorDoICMSDesonerado'), item_data.get('motivoDesoneracao'), item_data.get('codigoBeneficioFiscal'), item_data.get('percentualDiferimento'), item_data.get('valorICMSDiferimento') ] else: item_values = [None] * 82 return tuple(invoice_values + item_values) def delete_existing_invoices(self, cursor, invoice_keys): """ Deleta invoices existentes no banco de dados baseado nas keys Processa em lotes de 1000 keys para evitar limite de 2100 parâmetros do SQL Server """ if not invoice_keys: return 0 try: total_deleted = 0 batch_size = 1000 # Limite seguro abaixo dos 2100 parâmetros # Processar em lotes for i in range(0, len(invoice_keys), batch_size): batch = invoice_keys[i:i + batch_size] # Criar lista de placeholders para a query placeholders = ','.join(['?' for _ in batch]) delete_query = f"DELETE FROM [GINSENG].[dbo].[rgb_fiscal_invoices] WHERE [key] IN ({placeholders})" cursor.execute(delete_query, batch) deleted_count = cursor.rowcount total_deleted += deleted_count logger.info(f"Lote {i//batch_size + 1}: Deletados {deleted_count} registros de {len(batch)} keys") logger.info(f"Total deletado: {total_deleted} registros antigos de {len(invoice_keys)} keys") return total_deleted except pyodbc.Error as e: logger.error(f"Erro ao deletar invoices existentes: {e}") raise def process_invoices_to_database(self, invoices_data, batch_size=500): """ Processa as invoices e insere no banco de dados EM LOTES Usa executemany para inserção rápida em lotes de 500 registros ANTES de inserir, deleta registros antigos com mesmo invoiceId """ connection = None cursor = None try: # Conectar ao banco de dados connection = self.connect_database() cursor = connection.cursor() # Extrair lista de invoices if isinstance(invoices_data, dict): if 'items' in invoices_data: invoices = invoices_data['items'] else: invoices = [invoices_data] elif isinstance(invoices_data, list): invoices = invoices_data else: invoices = [invoices_data] total_invoices = len(invoices) logger.info(f"Processando {total_invoices} invoices para inserção no banco de dados...") logger.info(f"Usando inserção em lotes de {batch_size} registros") # PASSO 1: Coletar todas as keys para verificar se já existem invoice_keys = [invoice.get('key') for invoice in invoices if invoice.get('key')] logger.info(f"Verificando {len(invoice_keys)} keys no banco de dados...") # PASSO 2: Deletar invoices existentes deleted_count = self.delete_existing_invoices(cursor, invoice_keys) connection.commit() if deleted_count > 0: logger.info(f"Invoices antigas deletadas com sucesso. Prosseguindo com inserção dos novos dados...") # Query de inserção (será usada com executemany) insert_query = """ INSERT INTO [GINSENG].[dbo].[rgb_fiscal_invoices] ([invoiceId], [storeId], [stockLocationId], [fiscalOperationId], [fiscalOperationDescription], [supplierId], [supplierName], [clientId], [clientName], [emissionDate], [operationDate], [dateOfPassageOnFiscalPost], [deletedAt], [updatedAt], [invoiceNumber], [serie], [key], [buyerEmployeeId], [emitterEmployeeId], [cfop], [operationType], [shippingType], [paymentCondition], [fiscalDocumentType], [modality], [updateCost], [updateStock], [composeABC], [situation], [observation], [importType], [classification], [financeGenerationType], [shippingValue], [otherExpensesValue], [insuranceValue], [discountValue], [totalItemsValue], [documentValue], [valorDoDAE], [baseDeCalculoDoICMS], [valorDoICMS], [baseDeCalculoDoICMSSubstituicaoTributaria], [valorDoICMSSubstituicaoTributaria], [valorDoIPI], [valorDoPIS], [valorDoCOFINS], [valorDoICMSDesonerado], [baseDeCalculoFecop], [valorFecop], [baseDeCalculoFecopSubstituicaoTributaria], [valorFecopSubstituicaoTributaria], [itemId], [itemProductId], [itemSequential], [itemOrderNumber], [itemOrderItemSequencial], [itemComposeTotal], [itemCfop], [itemUnitType], [itemQuantityOfItensOnUnit], [itemQuantity], [itemCompleteQuantity], [itemUnitValue], [itemDiscountInputType], [itemUntaxedDiscountValue], [itemTaxedDiscountValue], [itemDiscountPercentage], [itemShippingInputType], [itemShippingValue], [itemShippingPercentage], [itemInsuranceInputType], [itemInsuranceValue], [itemInsurancePercentage], [itemOtherExpensesInputValue], [itemOtherExpensesValue], [itemOtherExpensesPercentage], [itemTotalValue], [itemTaxedPercentage], [itemProductCost], [itemFiscalCost], [itemAverageCost], [itemTipoDeEntradaDAE], [itemValorDoDAE], [itemPercentualDoDAE], [itemFiscalSituationId], [itemCsosn], [itemOutrasDespesasCompoeBaseDeCalculoIcms], [itemTributacao], [itemNcm], [itemCest], [itemAliquotaNacional], [itemAliquotaImportado], [itemAliquotaEstadual], [itemAliquotaMunicipal], [itemModalidadeDaBaseDeCalculo], [itemPercentualICMSDeCompra], [itemValorDoICMS], [itemValorDoICMSNoSimples], [itemBaseDeCalculoDoICMS], [itemBaseDeCalculoDoICMSComSubstituicaoTributaria], [itemAliquotaDoICMSComSubstituicaoTributaria], [itemValorDoICMSComSubstituicaoTributaria], [itemPercentualDeAgregacao], [itemPercentualDeReducaoDASubstituicaoTributaria], [itemAliquotaDoICMS], [itemAliquotaDoICMSDeVenda], [itemAliquotaDoICMSAntecipado], [itemValorDoICMSAntecipado], [itemAliquotaNoSimples], [itemCstDoIPI], [itemBaseDeCalculoDoIPI], [itemAliquotaDoIPI], [itemTipoDeEntradaIPI], [itemValorDoIPI], [itemPercentualDoIPI], [itemBaseDeCalculoDoPIS], [itemAliquotaDoPIS], [itemValorDoPIS], [itemBaseDeCalculoDoCOFINS], [itemAliquotaDoCOFINS], [itemValorDoCOFINS], [itemCodigoNaturezaDeImpostoFederal], [itemBaseDeCalculoDoFecop], [itemAliquotaDoFecop], [itemValorDoFecop], [itemBaseDeCalculoDoFecopSubstituto], [itemAliquotaDoFecopSubstituto], [itemValorDoFecopSubstituto], [itemValorDoICMSDesonerado], [itemMotivoDesoneracao], [itemCodigoBeneficioFiscal], [itemPercentualDiferimento], [itemValorICMSDiferimento]) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ # Coletar todos os registros para inserir batch_values = [] total_items_count = 0 for i, invoice in enumerate(invoices, 1): invoice_id = invoice.get('id') store_id = invoice.get('storeId') items = invoice.get('items', []) if items: for item in items: batch_values.append(self.prepare_invoice_values(invoice, item)) total_items_count += 1 else: batch_values.append(self.prepare_invoice_values(invoice, None)) total_items_count += 1 # Inserir em lotes if len(batch_values) >= batch_size: logger.info(f"Inserindo lote de {len(batch_values)} registros... (Invoice {i}/{total_invoices})") cursor.executemany(insert_query, batch_values) connection.commit() logger.info(f"Lote inserido com sucesso!") batch_values = [] # Inserir registros restantes if batch_values: logger.info(f"Inserindo lote final de {len(batch_values)} registros...") cursor.executemany(insert_query, batch_values) connection.commit() logger.info(f"Lote final inserido com sucesso!") # Relatório final logger.info("=" * 60) logger.info("RELATÓRIO DE INSERÇÃO NO BANCO DE DADOS") logger.info("=" * 60) logger.info(f"Total de invoices processadas: {total_invoices}") logger.info(f"Registros antigos deletados: {deleted_count}") logger.info(f"Total de registros inseridos: {total_items_count}") logger.info("=" * 60) return total_items_count, 0, deleted_count except Exception as e: logger.error(f"Erro geral no processamento: {e}") if connection: connection.rollback() raise finally: if cursor: cursor.close() if connection: connection.close() logger.info("Conexão com banco de dados fechada") def main(): """ Função principal para executar o extrator de invoices fiscais """ extractor = RGBFiscalInvoicesExtractor() print("=== EXTRATOR DE INVOICES FISCAIS RGB - GRUPO BOTICÁRIO ===") print("Iniciando busca de invoices fiscais...") # Determinar datas a processar if USE_YESTERDAY: # Usar data do dia anterior yesterday = datetime.now() - timedelta(days=1) search_dates = [yesterday.strftime("%Y-%m-%d")] print(f"Modo: Busca automática do dia anterior") print(f"Data a processar: {search_dates[0]}\n") else: # Gerar lista de datas entre START_DATE e END_DATE start = datetime.strptime(START_DATE, "%Y-%m-%d") end = datetime.strptime(END_DATE, "%Y-%m-%d") search_dates = [] current = start while current <= end: search_dates.append(current.strftime("%Y-%m-%d")) current += timedelta(days=1) print(f"Modo: Intervalo de datas configurado") print(f"Intervalo de datas: {START_DATE} até {END_DATE}") print(f"Total de datas a processar: {len(search_dates)}\n") # Contadores totais total_invoices_all_dates = 0 total_inserted_all_dates = 0 total_errors_all_dates = 0 total_deleted_all_dates = 0 try: # Processar cada data for idx, search_date in enumerate(search_dates, 1): print("\n" + "="*80) print(f"PROCESSANDO DATA {idx}/{len(search_dates)}: {search_date}") print("="*80) # Buscar dados da API para esta data data = extractor.get_fiscal_invoices_data(updated_at_date=search_date) # Mostrar resumo dos dados recebidos if isinstance(data, dict) and 'items' in data: invoices = data['items'] elif isinstance(data, list): invoices = data else: invoices = [data] if data else [] print(f"Dados recebidos da API: {len(invoices)} invoices encontradas para {search_date}") total_invoices_all_dates += len(invoices) # Contar total de itens total_items = 0 for invoice in invoices: items = invoice.get('items', []) total_items += len(items) if items else 1 # Se não tem itens, conta como 1 registro print(f"Total de registros que serão processados: {total_items}") # Salvar no banco de dados if len(invoices) > 0: print("\nIniciando inserção no banco de dados...") inserted, errors, deleted = extractor.process_invoices_to_database(data) total_inserted_all_dates += inserted total_errors_all_dates += errors total_deleted_all_dates += deleted print(f"\nResumo da inserção para {search_date}:") print(f"- Registros antigos deletados: {deleted}") print(f"- Registros inseridos: {inserted}") print(f"- Erros: {errors}") else: print(f"Nenhuma invoice encontrada para {search_date} - pulando inserção") # Relatório final consolidado print("\n" + "="*80) print("RELATÓRIO FINAL CONSOLIDADO") print("="*80) print(f"Total de datas processadas: {len(search_dates)}") print(f"Total de invoices encontradas: {total_invoices_all_dates}") print(f"Total de registros antigos deletados: {total_deleted_all_dates}") print(f"Total de registros inseridos: {total_inserted_all_dates}") print(f"Total de erros: {total_errors_all_dates}") print("="*80) print("\nProcessamento concluído com sucesso!") except Exception as e: print(f"\nErro durante o processamento: {e}") logger.error(f"Erro fatal: {e}") return 1 return 0 if __name__ == "__main__": main()