Server : Apache System : Linux dedi-14684855.grupobig.com 5.14.0-611.49.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Apr 21 16:39:08 EDT 2026 x86_64 User : grupo692 ( 1004) PHP Version : 8.2.31 Disable Function : NONE Directory : /home/grupo692/appstigrupobig/ |
#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.")