Viewing: consultaVendaCPF.py
#Consulta de Vendas por CPF e Mercadoria #http://162.214.65.39:8503/ #nohup python3 -m streamlit run consultaVendaCPF.py --server.port 8503 --server.address 0.0.0.0 > consultaVendaCPF.log 2>&1 & import streamlit as st import cx_Oracle import base64 from Crypto.Cipher import AES from Crypto.Protocol.KDF import PBKDF2 # Defina a variável de ambiente ORACLE_HOME #oracledb.init_oracle_client(lib_dir=r"C:\Oracle\instantclient_21_13") # Função para validar o CPF def validar_cpf(cpf): # Remove caracteres não numéricos cpf = ''.join([char for char in cpf if char.isdigit()]) # Verifica se o CPF tem 11 dígitos if len(cpf) != 11: return False # Evita CPFs inválidos como "111.111.111.11", "222.222.222.22", etc. if cpf == cpf[0] * 11: return False # Calcula o primeiro dígito verificador soma = sum(int(cpf[i]) * (10 - i) for i in range(9)) resto = soma % 11 digito1 = 0 if resto < 2 else 11 - resto if int(cpf[9]) != digito1: return False # Calcula o segundo dígito verificador soma = sum(int(cpf[i]) * (11 - i) for i in range(10)) resto = soma % 11 digito2 = 0 if resto < 2 else 11 - resto if int(cpf[10]) != digito2: return False return True def generate_key_iv(): key_iv = PBKDF2( MASTER_PASSWORD.encode("utf-8"), SALT, dkLen=48, # 32 bytes key + 16 bytes IV count=ITERATIONS ) return key_iv[:32], key_iv[32:] def decrypt(cipher_text: str) -> str: key, iv = generate_key_iv() cipher = AES.new(key, AES.MODE_CBC, iv) decrypted = cipher.decrypt(base64.b64decode(cipher_text)) # Remove PKCS7 padding pad_len = decrypted[-1] text = decrypted[:-pad_len].decode("utf-8", errors="ignore") # Remove BOM (se existir) return text.lstrip("\ufeff") # Configuração inicial da conexão com o banco de dados Oracle _pass = decrypt("+ivGpR7n9D4zKHONh/3W2A==") dsn = "187.109.221.38:1521/PROTON" user = "EDIUSER" password = _pass # Função para executar a consulta def consultar_venda_por_cpf(cpf, mercadoria): query = """ SELECT pv.TPED_NUMERO_DOCUMENTO FROM TPED_PEDIDO_VENDA pv LEFT JOIN TPED_PEDIDO_VENDA_ITEM pvi ON pvi.TPED_NUMERO_PEDIDO_FK_PK = pv.TPED_NUMERO_PEDIDO_PK INNER JOIN TPED_CONSUMIDOR_FINAL cf ON cf.TPED_PEDIDO_VENDA_UK = pv.TPED_NUMERO_PEDIDO_PK WHERE pv.TPED_UNIDADE_FK_PK IN (9, 13, 14, 18, 19, 20, 21, 23) AND cf.TPED_NUM_DOCUMENTO = :cpf AND pvi.TPED_CODIGO_BARRAS IN (:mercadoria) AND cf.TPED_UNIDADE_FK IN (9, 13, 14, 18, 19, 20, 21, 23) AND pv.TPED_STATUS_PEDIDO NOT IN ('CA') AND pv.TPED_NUMERO_PEDIDO_PK IS NOT NULL UNION ALL SELECT pv.TPED_NUMERO_DOCUMENTO FROM TPED_HISTORICO_VENDA pv LEFT JOIN TPED_HISTORICO_VENDA_ITEM pvi ON pvi.TPED_NUMERO_PEDIDO_FK_PK = pv.TPED_NUMERO_PEDIDO_PK INNER JOIN TPED_CONSUMIDOR_FINAL cf ON cf.TPED_PEDIDO_VENDA_UK = pv.TPED_NUMERO_PEDIDO_PK WHERE pv.TPED_UNIDADE_FK_PK IN (9, 13, 14, 18, 19, 20, 21, 23) AND cf.TPED_NUM_DOCUMENTO = :cpf AND pvi.TPED_CODIGO_BARRAS IN (:mercadoria) AND cf.TPED_UNIDADE_FK IN (9, 13, 14, 18, 19, 20, 21, 23) AND pv.TPED_STATUS_PEDIDO NOT IN ('CA') AND pv.TPED_NUMERO_PEDIDO_PK IS NOT NULL """ try: # Estabelecendo a conexão com o banco with cx_Oracle.connect(user=user, password=password, dsn=dsn) as connection: with connection.cursor() as cursor: cursor.execute(query, cpf=cpf, mercadoria=mercadoria) # Recupera os resultados resultados = cursor.fetchall() return resultados except cx_Oracle.DatabaseError as e: st.error(f"Erro ao consultar o banco de dados: {e}") return None # Configuração da interface do Streamlit st.title("Consulta de Vendas por CPF e Mercadoria") st.write("Insira um CPF e o código de barras para verificar se há vendas associadas.") # Entrada do usuário cpf_input = st.text_input("Digite o CPF (somente números):", max_chars=11) mercadoria_input = st.text_input("Digite o código de barras da mercadoria (somente números):", max_chars=13, value="7898147771029") # Botão para executar a consulta if st.button("Consultar"): if not cpf_input.isdigit() or len(cpf_input) != 11: st.error("Por favor, insira um CPF válido (somente números, 11 dígitos).") elif not validar_cpf(cpf_input): st.error("O CPF inserido é inválido. Por favor, verifique.") elif not mercadoria_input.isdigit() or len(mercadoria_input) not in [8, 12, 13]: st.error("Por favor, insira um código de barras válido (8, 12 ou 13 dígitos).") else: with st.spinner("Consultando..."): resultados = consultar_venda_por_cpf(cpf_input, mercadoria_input) if resultados: st.error("Vendas encontradas para o CPF e mercadoria informados!") st.write("Detalhes das vendas:") for row in resultados: st.write(f"NFe: {row[0]}") else: st.success("Nenhuma venda encontrada para o CPF e mercadoria informados.")
Return