126 lines
3.5 KiB
Python
126 lines
3.5 KiB
Python
# /SGMP_PROD/solicitacoes/intf_winthor.py
|
|
import os
|
|
import oracledb
|
|
|
|
# Tenta inicializar o Oracle Client usando a variável de ambiente ou caminho padrão
|
|
# No Docker, usa /opt/oracle/instantclient_21_13 (definido no Dockerfile)
|
|
# Em desenvolvimento local, pode usar /usr/lib/oracle/23/client64/lib
|
|
#
|
|
# ORACLE_USE_THIN=1 — não chama init_oracle_client; usa modo thin (sem Instant Client).
|
|
# Útil quando o servidor Oracle é acessível da máquina de dev mas não há libclntsh instalada.
|
|
_oracle_client_initialized = False
|
|
|
|
|
|
def _init_oracle_client():
|
|
"""Inicializa o Oracle Client (thick) de forma lazy, ou deixa o modo thin padrão."""
|
|
global _oracle_client_initialized
|
|
|
|
if _oracle_client_initialized:
|
|
return
|
|
|
|
if os.getenv("ORACLE_USE_THIN", "").strip().lower() in ("1", "true", "yes", "on"):
|
|
_oracle_client_initialized = True
|
|
return
|
|
|
|
oracle_lib_dir = os.getenv("ORACLE_LIB_DIR", "/opt/oracle/instantclient_21_13")
|
|
# API pública: oracledb.DatabaseError (não usar oracledb.exceptions.* — quebra em algumas instalações)
|
|
_DatabaseError = getattr(oracledb, "DatabaseError", Exception)
|
|
|
|
# Tenta inicializar com o caminho do Docker / env primeiro
|
|
try:
|
|
oracledb.init_oracle_client(lib_dir=oracle_lib_dir)
|
|
_oracle_client_initialized = True
|
|
return
|
|
except (_DatabaseError, OSError, Exception):
|
|
pass
|
|
|
|
# Se falhar, tenta o caminho alternativo (desenvolvimento local com RPM Oracle)
|
|
if oracle_lib_dir != "/usr/lib/oracle/23/client64/lib":
|
|
try:
|
|
oracledb.init_oracle_client(lib_dir="/usr/lib/oracle/23/client64/lib")
|
|
_oracle_client_initialized = True
|
|
return
|
|
except (_DatabaseError, OSError, Exception):
|
|
pass
|
|
|
|
# Sem thick: segue em modo thin (sem segunda tentativa de init neste processo)
|
|
_oracle_client_initialized = True
|
|
|
|
def get_oracle_connection():
|
|
_init_oracle_client()
|
|
|
|
dsn = oracledb.makedsn(
|
|
host="10.1.1.241",
|
|
port=1521,
|
|
service_name="WINT"
|
|
)
|
|
|
|
return oracledb.connect(
|
|
user="SEVEN",
|
|
password="usr54sev",
|
|
dsn=dsn
|
|
)
|
|
|
|
def buscar_colaborador_oracle(cpf: str):
|
|
conn = get_oracle_connection()
|
|
cursor = conn.cursor()
|
|
|
|
query = """
|
|
SELECT
|
|
MATRICULA,
|
|
NOME,
|
|
CPF,
|
|
SITUACAO,
|
|
ADMISSAO,
|
|
DT_EXCLUSAO,
|
|
ENDERECO,
|
|
BAIRRO,
|
|
CIDADE,
|
|
ESTADO
|
|
FROM
|
|
PCEMPR
|
|
WHERE
|
|
CPF = :cpf
|
|
"""
|
|
cursor.execute(query, {"cpf": cpf})
|
|
row = cursor.fetchone()
|
|
colunas = [c[0].lower() for c in cursor.description] if row else []
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return dict(zip(colunas, row))
|
|
|
|
def autenticar_usuario(login: str, senha: str):
|
|
conn = get_oracle_connection()
|
|
cursor = conn.cursor()
|
|
|
|
query = """
|
|
SELECT
|
|
MATRICULA,
|
|
USUARIOBD,
|
|
NOME,
|
|
EMAIL,
|
|
CPF,
|
|
CODFILIAL,
|
|
SITUACAO
|
|
FROM PCEMPR
|
|
WHERE USUARIOBD = UPPER(:login)
|
|
AND SENHABD = CRYPT(UPPER(:senha), USUARIOBD)
|
|
AND SITUACAO = 'A'
|
|
"""
|
|
|
|
cursor.execute(query, {"login": login, "senha": senha})
|
|
row = cursor.fetchone()
|
|
colunas = [c[0].lower() for c in cursor.description] if row else []
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
if not row:
|
|
return None # login ou senha inválidos
|
|
|
|
return dict(zip(colunas, row))
|
|
|