0s - Explorer

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