from django.contrib.admin.views.decorators import staff_member_required from django.db.models import Count, F, Sum, Avg from django.db.models.functions import ExtractYear, ExtractMonth from django.http import JsonResponse, HttpResponse from django.shortcuts import render from django.contrib.admin.models import LogEntry, ADDITION from django.contrib.contenttypes.models import ContentType from datetime import datetime, timedelta, date from notas.models import Order, Vendor, Product from utils.charts import months, colorPrimary, colorSuccess, colorDanger, generate_color_palette, get_year_dict from django.db import IntegrityError from seleniumbase import SB from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from seleniumbase.common.exceptions import NoSuchElementException, TimeoutException from filelock import FileLock import undetected_chromedriver as uc import subprocess import time import os import re from bs4 import BeautifulSoup from notas.sefaz import SefazError, _setup_ppp0_interface def index(request): return render(request, 'index.html') def terms(request): return render(request, 'termos.html') def privacy(request): return render(request, 'privacidade.html') def data_deletion(request): return render(request, 'exclusao-dados.html') def support(request): return render(request, 'suporte.html') # Função auxiliar para obter IDs de pedidos criados pelo usuário def get_user_order_ids(request): if request.user.is_superuser: return Order.objects.all().values_list('id', flat=True) # Superusuários veem todos content_type = ContentType.objects.get(app_label='notas', model='order') user_log_entries = LogEntry.objects.filter( user_id=request.user.id, action_flag=ADDITION, content_type_id=content_type.id ).values_list('object_id', flat=True) return [int(obj_id) for obj_id in user_log_entries] # Converte object_id para inteiros @staff_member_required def statistics_view(request): return render(request, "statistics.html", {}) @staff_member_required def get_filter_options(request): # Filtra os pedidos do usuário user_order_ids = get_user_order_ids(request) # Extrai os anos dos pedidos filtrados grouped_purchases = Order.objects.filter(id__in=user_order_ids).annotate(year=ExtractYear("sale_date")).values("year").order_by("-year").distinct() options = [purchase["year"] for purchase in grouped_purchases] return JsonResponse({ "options": options, }) @staff_member_required def get_sales_chart(request, year): # Filtra os pedidos do usuário user_order_ids = get_user_order_ids(request) purchases = Order.objects.filter(sale_date__year=year, id__in=user_order_ids) grouped_purchases = purchases.annotate(price=F("orderproduct__product_total_price")).annotate(month=ExtractMonth("sale_date"))\ .values("month").annotate(average=Sum("orderproduct__product_total_price")).values("month", "average").order_by("month") sales_dict = get_year_dict() for group in grouped_purchases: sales_dict[months[group["month"]-1]] = round(group["average"], 2) return JsonResponse({ "title": f"Sales in {year}", "data": { "labels": list(sales_dict.keys()), "datasets": [{ "label": "Amount ($)", "backgroundColor": colorPrimary, "borderColor": colorPrimary, "data": list(sales_dict.values()), }] }, }) @staff_member_required def spend_per_customer_chart(request, year): # Filtra os pedidos do usuário user_order_ids = get_user_order_ids(request) purchases = Order.objects.filter(sale_date__year=year, id__in=user_order_ids) grouped_purchases = purchases.annotate(price=F("orderproduct__product_total_price")).annotate(month=ExtractMonth("sale_date"))\ .values("month").annotate(average=Sum("orderproduct__product_total_price")).values("month", "average").order_by("month") spend_per_customer_dict = get_year_dict() for group in grouped_purchases: spend_per_customer_dict[months[group["month"]-1]] = round(group["average"], 2) return JsonResponse({ "title": f"Spend per customer in {year}", "data": { "labels": list(spend_per_customer_dict.keys()), "datasets": [{ "label": "Amount ($)", "backgroundColor": colorPrimary, "borderColor": colorPrimary, "data": list(spend_per_customer_dict.values()), }] }, }) @staff_member_required def payment_success_chart(request, year): # Filtra os pedidos do usuário user_order_ids = get_user_order_ids(request) purchases = Order.objects.filter(sale_date__year=year, id__in=user_order_ids) return JsonResponse({ "title": f"Payment success rate in {year}", "data": { "labels": ["Com Desconto", "Sem Desconto"], "datasets": [{ "label": "Amount ($)", "backgroundColor": [colorSuccess, colorDanger], "borderColor": [colorSuccess, colorDanger], "data": [ purchases.filter(total_discount=0).count(), purchases.filter(total_discount=1).count(), # Nota: possível erro aqui, explicado abaixo ], }] }, }) @staff_member_required def payment_method_chart(request, year): # Filtra os pedidos do usuário user_order_ids = get_user_order_ids(request) purchases = Order.objects.filter(sale_date__year=year, id__in=user_order_ids) grouped_purchases = purchases.values("payment_form").annotate(count=Count("id"))\ .values("payment_form", "count").order_by("payment_form") payment_method_dict = dict() for payment_method in Order.PAYMENT_FORM: payment_method_dict[payment_method[1]] = 0 for group in grouped_purchases: payment_method_dict[dict(Order.PAYMENT_FORM)[group["payment_form"]]] = group["count"] return JsonResponse({ "title": f"Payment method rate in {year}", "data": { "labels": list(payment_method_dict.keys()), "datasets": [{ "label": "Amount ($)", "backgroundColor": generate_color_palette(len(payment_method_dict)), "borderColor": generate_color_palette(len(payment_method_dict)), "data": list(payment_method_dict.values()), }] }, }) @staff_member_required def import_by_key(request): usuario_id = request.user.id try: return _import_by_key_inner(request, usuario_id) except Exception as e: if usuario_id == 1: return JsonResponse({ "dados": [], "erro_msg": f'"Erro ao processar nota: {str(e)}"' }) return JsonResponse({ "dados": [], "erro_msg": '"Ocorreu um erro ao importar a nota. Tente novamente mais tarde."' }) def _import_by_key_inner(request, usuario_id): url_sefaz = request.POST['url_sefaz'] lock_path = "/tmp/import_by_key.lock" lock = FileLock(lock_path, timeout=60) # Timeout de 60 segundos chave_sefaz = re.search("[0-9]{44}", url_sefaz) if chave_sefaz: chave_sefaz = chave_sefaz.group(0) #Ex: 53241135377959000125650070005328261462151580 else: return JsonResponse({ "dados": [], "erro_msg": f'"Chave inválida. Não foi possivel extrair os 44 digitos."' }) try: with lock: # Garante que apenas uma requisição execute o bloco crítico print('tudo ok com a VPN!') print('iniciando o chrome..') try: _setup_ppp0_interface() except SefazError as e: if usuario_id == 1: return JsonResponse({"error": 1, "msg": str(e)}) return JsonResponse({ "dados": [], "erro_msg": '"Ocorreu um erro ao importar a nota. Tente novamente mais tarde."' }) os.environ["DISPLAY"] = ":99" os.environ["HOME"] = "/var/www/html/melhorprecodf.com.br/chrome_temp" # Configura opções do Chrome chrome_options = uc.ChromeOptions() chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--disable-dev-shm-usage') chrome_options.add_argument('--remote-debugging-port=0') #chrome_options.add_argument('--log-level=0') #chrome_options.add_argument('--enable-logging') chrome_options.add_argument('--v=1') #chrome_options.add_argument('--user-data-dir=/home/ubuntu/.config/google-chrome/Default') chrome_options.add_argument('--user-data-dir=/var/www/html/melhorprecodf.com.br/chrome_temp/user-data') chrome_options.add_argument('--proxy-server=http://127.0.0.1:3128') # Proxy do Squid chrome_options.add_argument('--disable-crash-reporter') # Desativar Crashpad #chrome_options.add_argument('--crash-dumps-dir=/var/www/html/melhorprecodf.com.br/chrome_temp/crashpad') #chrome_options.add_argument('--log-file=/var/www/html/melhorprecodf.com.br/chrome_temp/chrome_debug.log') chrome_options.add_argument('--disable-extensions') # Evitar extensões que podem interferir chrome_options.add_argument('--disable-popup-blocking') # Permitir pop-ups, se necessário chrome_options.add_argument('--no-first-run') # Evitar tela inicial do Chrome chrome_options.add_argument('--disable-default-apps') # Desativar aplicativos padrão chrome_options.add_argument('--user-agent=Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36') html = 'vazio' ### Obtem o Link com Token ### with SB(uc=True) as sb: # Configurar o driver com opções personalizadas sb.driver = uc.Chrome(options=chrome_options, version_main=151) url = "https://ww1.receita.fazenda.df.gov.br/DecVisualizador/Nfce/Captcha?Chave="+chave_sefaz+"&Codigo=100&Mensagem=QR^%^20Code^%^20Inv^%^c3^%^a1lido&Descricao=Hash^%^20QR^%^20Code^%^20inv^%^c3^%^a1lido." #url = "https://ifconfig.me/" #sb.uc_open_with_reconnect(url, 4) sb.driver.get(url) #sb.uc_gui_click_captcha() time.sleep(20) #importante para dar tempo do captcha carregar sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/screenshot1.png") #os._exit(0) sb.set_default_timeout(120) # Define o tempo limite padrão como 60 segundos html = sb.get_page_source() try: sb.wait_for_element_clickable('input.btn.btn-primary[type="submit"]', by=By.CSS_SELECTOR) sb.click('input.btn.btn-primary[type="submit"]') #time.sleep(30) #importante para dar tempo do captcha carregar sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/screenshot2.png") except (NoSuchElementException, TimeoutException) as e: html = sb.get_page_source() soup = BeautifulSoup(html, "html.parser") h1_element = soup.find('h1') erro_chave_nao_encontrada = h1_element.get_text(strip=True) if h1_element else f"Erro: {str(e)}" sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/screenshot2-erro.png") #os._exit(0) sb.driver.quit() if usuario_id == 1: return JsonResponse({ "dados": [], "erro_msg": f'"Sistema sobrecarregado: {erro_chave_nao_encontrada}"' }) return JsonResponse({ "dados": [], "erro_msg": '"Ocorreu um erro ao importar a nota. Tente novamente mais tarde."' }) sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/screenshot3.png") try: sb.wait_for_element_clickable('/html/body/div[1]/div/a', by=By.XPATH) sb.click('/html/body/div[1]/div/a') #time.sleep(30) sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/screenshot4.png") #os._exit(0) except (TimeoutException, NoSuchElementException) as e: html = sb.get_page_source() soup = BeautifulSoup(html, "html.parser") h1_element = soup.find('h1') erro_chave_nao_encontrada = h1_element.get_text(strip=True) if h1_element else f"Erro: {str(e)}" sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/screenshot3-erro.png") #os._exit(0) sb.driver.quit() return JsonResponse({ "dados": [], "erro_msg": '"Chave de acesso não encontrada"' }) html = sb.get_page_source() sb.driver.quit() except TimeoutError: if usuario_id == 1: return JsonResponse({ "dados": [], "erro_msg": '"Sistema ocupado. Tente novamente mais tarde."' }) return JsonResponse({ "dados": [], "erro_msg": '"Ocorreu um erro ao importar a nota. Tente novamente mais tarde."' }) #print(html) soup = BeautifulSoup(html, "html.parser") ## According 1 - HOME - Dados NFC-e nfce_home_numero = soup.find(id="accordion1").find_all("div","campo-xml")[0].getText() nfce_home_data_hora = soup.find(id="accordion1").find_all("div","campo-xml")[1].getText() #print("numero nfce: ", nfce_home_numero) #print("data hora nfce: ", nfce_home_data_hora) #print('--------------------------------') ## According 2 - HOME - Dados Básicos nfce_home_modelo = soup.find(id="accordion2").find_all("div","campo-xml")[0].getText() nfce_home_serie = soup.find(id="accordion2").find_all("div","campo-xml")[1].getText() nfce_home_numero = soup.find(id="accordion2").find_all("div","campo-xml")[2].getText() nfce_home_data_emissao = soup.find(id="accordion2").find_all("div","campo-xml")[3].getText() nfce_home_data_saida = soup.find(id="accordion2").find_all("div","campo-xml")[4].getText() nfce_home_valor_total = soup.find(id="accordion2").find_all("div","campo-xml")[5].getText() #print("modelo nfce: ", re.sub(' +|\n+', ' ', nfce_home_modelo)) #print("serie nfce: ", re.sub(' +|\n+', ' ', nfce_home_serie)) #print("numero nfce: ", re.sub(' +|\n+', ' ', nfce_home_numero)) #print("data emissao nfce: ", re.sub(' +|\n+', ' ', nfce_home_data_emissao)) #print("data saida nfce: ", re.sub(' +|\n+', ' ', nfce_home_data_saida)) #print("valor total nfce: ", re.sub(' +|\n+', ' ', nfce_home_valor_total)) #print('--------------------------------') ## According 3 - HOME - Emitente nfce_home_emit_cnpj = soup.find(id="accordion3").find_all("div","campo-xml")[0].getText() nfce_home_emit_razao_social = soup.find(id="accordion3").find_all("div","campo-xml")[1].getText() nfce_home_emit_ie = soup.find(id="accordion3").find_all("div","campo-xml")[2].getText() nfce_home_emit_uf = soup.find(id="accordion3").find_all("div","campo-xml")[3].getText() #print("emitente cnpj: ", re.sub(' +|\n+', ' ', nfce_home_emit_cnpj)) #print("emitente razao: ", re.sub(' +|\n+', ' ', nfce_home_emit_razao_social)) #print("emitente ie: ", re.sub(' +|\n+', ' ', nfce_home_emit_ie)) #print("emitente uf: ", re.sub(' +|\n+', ' ', nfce_home_emit_uf)) #print('--------------------------------') ## According 4 - HOME - Destinatário nfce_home_dest_cpf = soup.find(id="accordion4").find_all("div","campo-xml")[0].getText() #print("destinatario cpf: ", re.sub(' +|\n+', ' ', nfce_home_dest_cpf)) #print('--------------------------------') ## According 5 - HOME - Emissão nfce_home_natureza = soup.find(id="accordion5").find_all("div","campo-xml")[5].getText() #print("natureza operacao: ", re.sub(' +|\n+', ' ', nfce_home_natureza)) #print('--------------------------------') ## According 6 - HOME - informações Suplementares nfce_home_qrcode = soup.find(id="accordion6").find_all("div","campo-xml")[0].getText() #print("qrcode: ", re.sub(' +|\n+', ' ', nfce_home_qrcode)) ## According 7 - HOME - Situação nfce_home_situacao = soup.find(id="accordion7").find_all("div","campo-xml")[0].getText() nfce_home_protocolo = soup.find(id="accordion7").find_all("div","campo-xml")[1].getText() #print("situacao: ", re.sub(' +|\n+', ' ', nfce_home_situacao)) #print("protocolo: ", re.sub(' +|\n+', ' ', nfce_home_protocolo)) #print('--------------------------------') ## According 8 - Emitente nfce_emit_razao_social = soup.find(id="accordion8").find_all("div","campo-xml")[0].getText() nfce_emit_fantasia = soup.find(id="accordion8").find_all("div","campo-xml")[1].getText() nfce_emit_cnpj = soup.find(id="accordion8").find_all("div","campo-xml")[2].getText() nfce_emit_endereco = soup.find(id="accordion8").find_all("div","campo-xml")[3].getText() nfce_emit_bairro = soup.find(id="accordion8").find_all("div","campo-xml")[4].getText() nfce_emit_cep = soup.find(id="accordion8").find_all("div","campo-xml")[5].getText() nfce_emit_municipio = soup.find(id="accordion8").find_all("div","campo-xml")[6].getText() nfce_emit_telefone = soup.find(id="accordion8").find_all("div","campo-xml")[7].getText() nfce_emit_uf = soup.find(id="accordion8").find_all("div","campo-xml")[8].getText() nfce_emit_pais = soup.find(id="accordion8").find_all("div","campo-xml")[9].getText() nfce_emit_ie = soup.find(id="accordion8").find_all("div","campo-xml")[10].getText() #print("razao social: ", re.sub(' +|\n+', ' ', nfce_emit_razao_social)) #print("fantasia: ", re.sub(' +|\n+', ' ', nfce_emit_fantasia)) #print("cnpj: ", re.sub(' +|\n+', ' ', nfce_emit_cnpj)) #print("endereco: ", re.sub(' +|\n+', ' ', nfce_emit_endereco)) #print("bairro: ", re.sub(' +|\n+', ' ', nfce_emit_bairro)) #print("cep: ", re.sub(' +|\n+', ' ', nfce_emit_cep)) #print("municipio: ", re.sub(' +|\n+', ' ', nfce_emit_municipio)) #print("telefone: ", re.sub(' +|\n+', ' ', nfce_emit_telefone)) #print("uf: ", re.sub(' +|\n+', ' ', nfce_emit_uf)) #print("pais: ", re.sub(' +|\n+', ' ', nfce_emit_pais)) #print("ie: ", re.sub(' +|\n+', ' ', nfce_emit_ie)) #print('--------------------------------') ## According 9 - Destinatario nfce_dest_cpf = soup.find(id="accordion9").find_all("div","campo-xml")[1].getText() #print("cpf: ", re.sub(' +|\n+', ' ', nfce_dest_cpf)) #print('--------------------------------') ## According 10 - Produtos e Servicos nfce_itens = soup.find(id="accordion10").find_all("div","item-accordion") itens_salvos = [] for item in nfce_itens: item_descricao = item.find_all("div","campo-xml")[0].getText() item_quantidade = item.find_all("div", "campo-xml")[1].getText() item_un = item.find_all("div", "campo-xml")[2].getText() item_valor_total = item.find_all("div", "campo-xml")[3].getText() item_codigo = item.find_all("div", "campo-xml")[4].getText() item_ncm = item.find_all("div", "campo-xml")[5].getText() item_cfop = item.find_all("div", "campo-xml")[8].getText() item_ean = item.find_all("div", "campo-xml")[14].getText()[:13] item_valorcomercializao = item.find_all("div", "campo-xml")[20].getText() item_tot_tributos = item.find_all("div", "campo-xml")[22].getText() #print('Item:') #print("descricao: ", re.sub(' +|\n+', ' ', item_descricao)) #print("quantidade: ", re.sub(' +|\n+', ' ', item_quantidade)) #print("unidade: ", re.sub(' +|\n+', ' ', item_un)) #print("valor: ", re.sub(' +|\n+', ' ', item_valor)) #print("codigo: ", re.sub(' +|\n+', ' ', item_codigo)) #print("ncm: ", re.sub(' +|\n+', ' ', item_ncm)) #print("cfop: ", re.sub(' +|\n+', ' ', item_cfop)) #print("ean: ", re.sub(' +|\n+', ' ', item_ean)) #print("tributos: ", re.sub(' +|\n+', ' ', item_tot_tributos)) #print('--------------------------------') item_ncm = item_ncm.strip() item_ean = item_ean.strip() item_cfop = item_cfop.strip() product = Product( description_text=item_descricao, ncm=item_ncm, cfop=item_cfop, ean=item_ean, unidademedida=item_un, ) try: product, criado = Product.objects.get_or_create( ean=item_ean, description_text=item_descricao, defaults={ "description_text": item_descricao, "ncm": item_ncm, "cfop": item_cfop, "unidademedida": item_un, } ) if criado: LogEntry.objects.log_action( user_id=usuario_id, content_type_id=ContentType.objects.get_for_model(product).pk, object_id=product.pk, object_repr=str(product), action_flag=ADDITION, change_message='[{"added": {}}]' ) except IntegrityError as e: pass itens_salvos.append({ "id": product.pk, "description_text": item_descricao, "item_valor": item_valorcomercializao, "item_quantidade": item_quantidade, "item_tot_tributos": item_tot_tributos, "valor_total": item_valor_total }) ## According 11 - Totais nfce_totais_base_icms = soup.find(id="accordion11").find_all("div","campo-xml")[0].getText() nfce_totais_valor_icms = soup.find(id="accordion11").find_all("div","campo-xml")[1].getText() nfce_totais_valor_produtos = soup.find(id="accordion11").find_all("div","campo-xml")[7].getText() nfce_totais_valor_ipi = soup.find(id="accordion11").find_all("div","campo-xml")[11].getText() nfce_totais_valor_nfce = soup.find(id="accordion11").find_all("div","campo-xml")[12].getText() nfce_totais_valor_descontos = soup.find(id="accordion11").find_all("div","campo-xml")[13].getText() nfce_totais_valor_pis = soup.find(id="accordion11").find_all("div","campo-xml")[15].getText() nfce_totais_valor_cofins = soup.find(id="accordion11").find_all("div","campo-xml")[16].getText() #print("base icms: ", re.sub(' +|\n+', ' ', nfce_totais_base_icms)) #print("valor icms: ", re.sub(' +|\n+', ' ', nfce_totais_valor_icms)) #print("valor produtos: ", re.sub(' +|\n+', ' ', nfce_totais_valor_produtos)) #print("valor ipi: ", re.sub(' +|\n+', ' ', nfce_totais_valor_ipi)) #print("valor nfce: ", re.sub(' +|\n+', ' ', nfce_totais_valor_nfce)) #print("valor descontos: ", re.sub(' +|\n+', ' ', nfce_totais_valor_descontos)) #print("valor pis: ", re.sub(' +|\n+', ' ', nfce_totais_valor_pis)) #print("valor cofins: ", re.sub(' +|\n+', ' ', nfce_totais_valor_cofins)) #print('--------------------------------') ## According 17 - Cobrança #nfce_cobranca_forma = soup.find(id="accordion17").find_all("div","campo-xml")[0].getText() #nfce_cobranca_valor = soup.find(id="accordion17").find_all("div","campo-xml")[1].getText() #nfce_cobranca_tipo = soup.find(id="accordion17").find_all("div","campo-xml")[2].getText() #nfce_cobranca_cnpj_credenciadora = soup.find(id="accordion17").find_all("div","campo-xml")[3].getText() #nfce_cobranca_bandeira = soup.find(id="accordion17").find_all("div","campo-xml")[4].getText() #nfce_cobranca_num_autorizacao = soup.find(id="accordion17").find_all("div","campo-xml")[5].getText() #print("forma: ", re.sub(' +|\n+', ' ', nfce_cobranca_forma)) #print("valor: ", re.sub(' +|\n+', ' ', nfce_cobranca_valor)) #print("tipo: ", re.sub(' +|\n+', ' ', nfce_cobranca_tipo)) #print("cnpj: ", re.sub(' +|\n+', ' ', nfce_cobranca_cnpj_credenciadora)) #print("bandeira: ", re.sub(' +|\n+', ' ', nfce_cobranca_bandeira)) #print("numero autorizacao: ", re.sub(' +|\n+', ' ', nfce_cobranca_num_autorizacao)) #print('--------------------------------') # Encontra todas as seções de "row espacamento-campos" dentro do accordion17 formas_pagamento = [] valores_pagamento = [] #tipos = [] #cnpjs_credenciadora = [] #bandeiras = [] #numeros_autorizacao = [] formas_valores_pagamento = [] # Encontra todas as seções de "row espacamento-campos" dentro do accordion17 cobranca_rows = soup.find(id="accordion17").find_all("div", class_="row espacamento-campos") # Itera sobre cada seção de pagamento for row in cobranca_rows: campos = row.find_all("div", class_="campo-xml") # Verifica se há pelo menos 6 campos; se não, preenche com string vazia forma_pagamento = campos[0].getText().strip() if len(campos) > 0 else "" formas_pagamento.append(forma_pagamento) valor_pagamento = campos[1].getText().strip() if len(campos) > 1 else "" valores_pagamento.append(valor_pagamento) #tipos.append(campos[2].getText().strip() if len(campos) > 2 else "") #cnpjs_credenciadora.append(campos[3].getText().strip() if len(campos) > 3 else "") #bandeiras.append(campos[4].getText().strip() if len(campos) > 4 else "") #numeros_autorizacao.append(campos[5].getText().strip() if len(campos) > 5 else "") formas_valores_pagamento.append(forma_pagamento+' ('+valor_pagamento+')') # Concatena os valores com ";" #nfce_cobranca_forma = ";".join(formas_pagamento) #nfce_cobranca_valor = ";".join(valores_pagamento) #nfce_cobranca_tipo = ";".join(tipos) #nfce_cobranca_cnpj_credenciadora = ";".join(cnpjs_credenciadora) #nfce_cobranca_bandeira = ";".join(bandeiras) #nfce_cobranca_num_autorizacao = ";".join(numeros_autorizacao) nfce_cobranca_forma_valor = ";".join(formas_valores_pagamento) #print('--------------------------------') ## According 18 - Informacoes Adicionais nfce_inf_adic_formato_impressao = soup.find(id="accordion18").find_all("div","campo-xml")[0].getText() nfce_inf_adic_versao_xslt = soup.find(id="accordion18").find_all("div","campo-xml")[1].getText() #print("formato: ", re.sub(' +|\n+', ' ', nfce_inf_adic_formato_impressao)) #print("versao: ", re.sub(' +|\n+', ' ', nfce_inf_adic_versao_xslt)) #print('----------- FIM -------------') # Verifica se o vendor já existe com o mesmo CNPJ vendor, created = Vendor.objects.get_or_create( cpfcnpj_text=nfce_emit_cnpj.strip(), defaults={ 'nomerazaosocial_text': nfce_emit_razao_social, 'nomefantasia_text': nfce_emit_fantasia, 'endereco_text': nfce_emit_endereco, 'bairro_text': nfce_emit_bairro, 'cep_text': nfce_emit_cep, 'municipio_text': nfce_emit_municipio, 'uf_text': nfce_emit_uf, 'telefone_text': nfce_emit_telefone, 'inscricaoestadual_text': nfce_emit_ie, } ) # Registra a ação no LogEntry (apenas se criado) if created: try: LogEntry.objects.log_action( user_id=usuario_id, content_type_id=ContentType.objects.get_for_model(vendor).pk, object_id=vendor.pk, object_repr=str(vendor), action_flag=ADDITION, change_message='[{"added": {}}]' ) except Exception as e: print(f"Erro ao registrar LogEntry: {e}") if usuario_id == 1: return JsonResponse({"error": f"Erro ao registrar log: {str(e)}"}, status=500) return JsonResponse({ "dados": [], "erro_msg": '"Ocorreu um erro ao importar a nota. Tente novamente mais tarde."' }, status=500) return JsonResponse({ "dados": { "data_hora": nfce_home_data_emissao, "valores": { "total_descontos": nfce_totais_valor_descontos, "total_produtos": nfce_totais_valor_produtos, "total_pagamentos": nfce_home_valor_total, }, "emitente": { "id": vendor.id, # Agora deve ter um valor válido "cpfcnpj": nfce_emit_cnpj, "nomefantasia": nfce_emit_fantasia, }, "cobranca": { "forma": nfce_cobranca_forma_valor[4:] }, "produtos": itens_salvos } }) #@staff_member_required def jogar_megasena(): lock_path = "/tmp/jogar_megasena.lock" lock = FileLock(lock_path, timeout=60) # Timeout de 60 segundos try: with lock: # Garante que apenas uma requisição execute o bloco crítico print('tudo ok com a VPN!') print('iniciando o chrome..') try: _setup_ppp0_interface() except SefazError as e: return JsonResponse({"error": 1, "msg": str(e)}) os.environ["DISPLAY"] = ":99" os.environ["HOME"] = "/var/www/html/melhorprecodf.com.br/chrome_temp" # Configura opções do Chrome chrome_options = uc.ChromeOptions() chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--disable-dev-shm-usage') chrome_options.add_argument('--remote-debugging-port=0') #chrome_options.add_argument('--log-level=0') #chrome_options.add_argument('--enable-logging') chrome_options.add_argument('--v=1') #chrome_options.add_argument('--user-data-dir=/home/ubuntu/.config/google-chrome/Default') chrome_options.add_argument('--user-data-dir=/var/www/html/melhorprecodf.com.br/chrome_temp/user-data') chrome_options.add_argument('--proxy-server=http://127.0.0.1:3128') # Proxy do Squid chrome_options.add_argument('--disable-crash-reporter') # Desativar Crashpad #chrome_options.add_argument('--crash-dumps-dir=/var/www/html/melhorprecodf.com.br/chrome_temp/crashpad') #chrome_options.add_argument('--log-file=/var/www/html/melhorprecodf.com.br/chrome_temp/chrome_debug.log') chrome_options.add_argument('--disable-extensions') # Evitar extensões que podem interferir chrome_options.add_argument('--disable-popup-blocking') # Permitir pop-ups, se necessário chrome_options.add_argument('--no-first-run') # Evitar tela inicial do Chrome chrome_options.add_argument('--disable-default-apps') # Desativar aplicativos padrão chrome_options.add_argument('--user-agent=Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36') html = 'vazio' with SB(uc=True) as sb: # Configurar o driver com opções personalizadas sb.driver = uc.Chrome(options=chrome_options, version_main=151) url = "https://internetbanking.caixa.gov.br/sinbc/#!nb/login" #url = "https://ifconfig.me/" #sb.uc_open_with_reconnect(url, 4) sb.driver.get(url) #sb.uc_gui_click_captcha() time.sleep(15) #importante para dar tempo do captcha carregar sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/caixa1.png") #os._exit(0) sb.set_default_timeout(180) # Define o tempo limite padrão como 60 segundos # Esperar até 15 segundos pelo campo de texto com id="nome_usuario" sb.wait_for_element("#nomeUsuario", timeout=15) sb.type("#nomeUsuario", "d4ni3lc4t40") sb.click("#btnLogin") time.sleep(120) #importante para dar tempo do captcha carregar sb.driver.get_screenshot_as_file("/var/www/html/melhorprecodf.com.br/downloaded_files/caixa2.png") html = sb.get_page_source() print(html) sb.driver.quit() return JsonResponse({ "dados": [], "erro_msg": '"Ok. Deu certo."' }) except TimeoutError: return JsonResponse({ "dados": [], "erro_msg": '"Sistema ocupado. Tente novamente mais tarde."' })