1041 lines
44 KiB
Python
1041 lines
44 KiB
Python
import json, time, sys, inspect, os
|
|
import hmac, hashlib, uuid, re
|
|
|
|
import pika, pprint
|
|
from datetime import datetime
|
|
from pytz import timezone
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.primitives import padding
|
|
from cryptography.hazmat.backends import default_backend
|
|
|
|
from queue import Queue, Full, Empty
|
|
from threading import Thread, Event
|
|
from abc import ABC, abstractmethod
|
|
from typing import Sequence, Any
|
|
import logging
|
|
|
|
from utils.exchangeModels import *
|
|
|
|
versao = "v1.0.0 14/12/2023 - Baltazar - Prefetch count"
|
|
versao = "v1.1.0 06/02/2024 - MJ. - Detecta Serviço rabbitmq ausente no inicio"
|
|
versao = "v1.2.0 20/02/2024 - MJ. - Possibilita uma taksqueue por serviço"
|
|
versao = "v1.3.0 06/03/2024 - MJ. - Verboso apenas nos erros"
|
|
versao = "v1.5.0 07/03/2024 - MJ. - External Rabbitmq Connection"
|
|
versao = "v1.6.0 26/06/2024 - MJ. - Conversao dos parametros host, service e client para lowercase"
|
|
versao = "v1.7.0 26/07/2024 - MJ. - Prepacao para utilizar Cloudamqp ou Rabbitmq local amqps/system"
|
|
versao = "v2.0.0 16/08/2024 - MJ. - Nova classe RPC e restruturação de Exchange2 e Tasks"
|
|
versao = "v2.1.0 21/11/2024 - MJ. - Clareza no log e tratamento de erros"
|
|
versao = "v2.2.0 31/01/2025 - MJ. - Remoção de alguns logs e melhorias de performance"
|
|
versao = "v2.3.0 12/02/2025 - MJ. - Melhorias na conexão com o Rabbitmq"
|
|
versao = "v2.4.0 14/02/2025 - MJ. - Check amqp url parameters"
|
|
versao = "v2.4.1 19/02/2025 - MJ. - Correcao prefetch_count=1"
|
|
versao = "v2.5.1 13/05/2025 - MJ. - Payload até 128kb - ajuste documentação e visual"
|
|
versao = "v2.5.2 01/10/2025 - MJ. - Evita especificar um client vazio"
|
|
versao = "v3"
|
|
versaostr = f"{versao}0.0 16/01/2026 S.D.G. - SG_IC2 LoggerExchange"
|
|
|
|
def is_amqp_url(url):
|
|
pattern = r"^amqps?:\/\/([a-zA-Z0-9_\-\.]+):([^@]+)@([a-zA-Z0-9_\-\.]+)(:[0-9]+)?(\/[a-zA-Z0-9_\-\.]*)?$"
|
|
|
|
return bool(re.match(pattern, url))
|
|
|
|
# QUEUE EXCHANGE - Apoia no comunicação inter-serviços de um sistema
|
|
# EXCHANGE TIPO DIRECT
|
|
# NAO DURAVEL - EXCLUSIVO - AUTODELETE
|
|
# SE NAO HOUVER QUEM RECEBA A MSG A MSG SERÁ PERDIDA -> O ENVIO FUNCIONA MESMO QUE NINGUEM EXISTA PARA RECEBER
|
|
class QueueExchange2:
|
|
def __init__( self,
|
|
amqps:str,
|
|
origin:str,
|
|
system:str,
|
|
service:str,
|
|
version:str,
|
|
client:str = '#',
|
|
callback: object = None,
|
|
delivery_mode = 1):
|
|
|
|
paramOK = len(service) and len(system) and len(amqps) and len(origin) and len(version) and len(client)
|
|
if not (paramOK) or not is_amqp_url(amqps):
|
|
if not is_amqp_url(amqps):
|
|
msg = f"NOT A VALID AMQP URL: {amqps}"
|
|
else:
|
|
msg = f"VERIFY PARAMETERS: is_amqp_url: {is_amqp_url(amqps)}\n amqps[{amqps}]\n origin[{origin}]\n system[{system}]\n service[{service}]\n version[{version}\n client[{client}]"
|
|
raise Exception(msg)
|
|
|
|
self.delivery = delivery_mode
|
|
self.payload = ""
|
|
self.hmac = ""
|
|
self.version = version
|
|
self.serviceid = ""
|
|
self.type = "msg"
|
|
self.origin = origin.lower()
|
|
self.client = client.lower()
|
|
self.service = service.lower()
|
|
self.system = system.lower()
|
|
self.client = client
|
|
self.params = pika.URLParameters(amqps)
|
|
self.callback = callback
|
|
self.queue = f"{self.system}_{service}_{self.client}" if client != '#' else f"{self.system}_{service}"
|
|
if callback is not None:
|
|
try:
|
|
self.connection = pika.BlockingConnection(self.params)
|
|
self.channelIn = self.connection.channel() # start main channel
|
|
self.channelIn.exchange_declare (exchange = f"{self.system}.direct", exchange_type = 'direct')
|
|
self.channelIn.queue_declare (queue = self.queue, durable = True)
|
|
self.channelIn.queue_bind (queue = self.queue, exchange = f"{self.system}.direct", routing_key= self.queue)
|
|
self.channelIn.basic_qos (prefetch_count=1)
|
|
self.consumerTag = self.channelIn.basic_consume (queue = self.queue,
|
|
auto_ack = False,
|
|
on_message_callback = self.callback)
|
|
except Exception as error:
|
|
raise(error)
|
|
else:
|
|
conexao = pika.BlockingConnection(self.params) #testa conexao
|
|
conexao.close()
|
|
|
|
def cancel(self):
|
|
self.channelIn.basic_cancel(self.consumerTag)
|
|
|
|
# METODOS INTERNOS _
|
|
# MODELO MSG QUE TRAFEGA ENTRE SERVIÇOS E CONTROLLER
|
|
"""
|
|
MSG JSON CONTROLADOR
|
|
data:
|
|
{
|
|
"salesContractId": <int>,
|
|
"client": "bambui",
|
|
"service": "SALES_CONTRACT_CREATED"
|
|
}
|
|
|
|
message:
|
|
{ ...,
|
|
"serviceid": <registred uuid within controller>,
|
|
"payload": aes-256 <data>,
|
|
"hmac": hmac <payload>,
|
|
"type": "work",
|
|
...
|
|
}
|
|
"""
|
|
def _setMsg(self, msg:str, caller:str):
|
|
now = datetime.now(timezone('America/Sao_Paulo'))
|
|
message =msg.strip('"')
|
|
|
|
self.Msg = {
|
|
"system": self.system,
|
|
"target": self.target,
|
|
"service": self.service,
|
|
"serviceid": self.serviceid,
|
|
"payload": self.payload,
|
|
"hmac": self.hmac,
|
|
"type": self.type,
|
|
"version": self.version,
|
|
"origin": self.origin,
|
|
"client": self.client,
|
|
"caller": caller,
|
|
"logDate": f"{now}",
|
|
"msg": f"{message}"
|
|
}
|
|
|
|
# Envia uma mensagem para um serviço associado ao sistema
|
|
# controller é o nome padrão do serviço, caso nao seja informado
|
|
def sendMsg(self, message: str, service:str = 'controller'):
|
|
trials = 5
|
|
waitTime = 12
|
|
messageSent = False
|
|
caller = inspect.stack()[1].function
|
|
if caller == '<module>': caller = 'main'
|
|
|
|
self.target = service
|
|
self._setMsg(message, caller)
|
|
targetQueue = f"{self.system}_{self.target}_{self.client}" if self.client != '#' else f"{self.system}_{self.target}"
|
|
while not messageSent and trials:
|
|
try:
|
|
connection = pika.BlockingConnection(self.params)
|
|
channelOut = connection.channel() # start a channel
|
|
#channelOut.exchange_declare(exchange = f"{self.system}.direct", exchange_type = 'direct')
|
|
#channelOut.queue_declare (queue= targetQueue, durable= True)
|
|
#channelOut.queue_bind (queue= targetQueue, exchange= f"{self.system}.direct", routing_key= targetQueue)
|
|
channelOut.basic_publish(
|
|
exchange = f"{self.system}.direct",
|
|
routing_key = targetQueue,
|
|
body = json.dumps(self.Msg),
|
|
properties = pika.BasicProperties(content_type = 'text/plain', delivery_mode= self.delivery))
|
|
messageSent = True
|
|
connection.close()
|
|
except Exception as ex:
|
|
print(f"EXCH2: TROUBLE CNX RABBITMQ REMAIN ATTEMPT: {trials}")
|
|
print(ex, flush=True)
|
|
sys.stdout.flush()
|
|
time.sleep(waitTime)
|
|
trials -= 1
|
|
if messageSent:
|
|
self.payload = ""
|
|
self.hmac = ""
|
|
else:
|
|
print(f"EXCH2: ERROR MESSAGE NOT SENT TO [{targetQueue}]", flush=True)
|
|
|
|
def start_consuming(self):
|
|
if self.callback is not None:
|
|
self.consumerTag = self.channelIn.start_consuming()
|
|
|
|
def setCrypto(self, serviceid:str, servicekey:str):
|
|
self.serviceid = serviceid
|
|
self.servicekey = servicekey
|
|
self.Crypto = Payload(serviceid, servicekey)
|
|
|
|
def setPayload(self, data:str):
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.setPayload(data)
|
|
except Exception as ex:
|
|
print(f"EXCH2: CRYPTO ERROR: {ex}", flush=True)
|
|
|
|
def getPayload(self, hmac:str, safeData:str):
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.getPayload(safeData)
|
|
except Exception as ex:
|
|
print(ex)
|
|
return False
|
|
|
|
return hmac == self.hmac
|
|
|
|
# CRIPTOGRAFIA DO PAYLOAD
|
|
class Payload:
|
|
def __init__(self, serviceid:str, servicekey:str):
|
|
self.serviceid = serviceid
|
|
self.servicekey = servicekey
|
|
|
|
|
|
def setPayload(self, data:str):
|
|
data = data.encode('utf-8')
|
|
tamanho = len(data)
|
|
if tamanho > 1024 * 1024 * 10 or tamanho <= 0:
|
|
raise Exception("EXCHANGE2 Payload setPayload: Invalid Payload size")
|
|
|
|
hmacData= hmac.new(self.serviceid.encode('utf-8'), data, 'sha256').hexdigest()
|
|
iv = os.urandom(16)
|
|
padder = padding.PKCS7(128).padder()
|
|
datarw = padder.update(data) + padder.finalize()
|
|
cipher = Cipher(algorithms.AES(self.servicekey.encode('utf-8')), modes.CBC(iv))
|
|
encrypt = cipher.encryptor()
|
|
safeData = encrypt.update(datarw) + encrypt.finalize()
|
|
payload = (iv + safeData)
|
|
|
|
return (hmacData, payload.hex())
|
|
|
|
def getPayload(self, safeData:str):
|
|
safeData = bytes.fromhex(safeData)
|
|
iv = safeData[:16]
|
|
cipher = Cipher(algorithms.AES(self.servicekey.encode('utf-8')), modes.CBC(iv))
|
|
decrypt = cipher.decryptor()
|
|
rawdata = decrypt.update(safeData[16:]) + decrypt.finalize()
|
|
if len(rawdata)%16 != 0:
|
|
raise Exception("EXCHANGE2 Payload getPayload: Invalid Payload size block {0}".format(len(rawdata)))
|
|
|
|
unpadder = padding.PKCS7(128).unpadder()
|
|
data = unpadder.update(rawdata) + unpadder.finalize()
|
|
hmacData = hmac.new(self.serviceid.encode('utf-8'), data, 'sha256').hexdigest()
|
|
|
|
return (hmacData, data.decode('utf-8'))
|
|
|
|
# TASK EXCHANGE
|
|
# PERMITE CRIAR UM DISTRIBUIDOR DE TAREFAS
|
|
# O DISTRIBUIDOR DEVE RECEBER UMA FUNCAO DE CALLBACK
|
|
class TaskExchange2:
|
|
def __init__(self,
|
|
amqps:str,
|
|
origin:str,
|
|
system:str,
|
|
service:str,
|
|
version:str,
|
|
client:str = '#', callback: object = None, delivery_mode:int = 2):
|
|
paramOK = len(service) and len(system) and len(amqps) and len(origin) and len(version)
|
|
if not (paramOK) or not is_amqp_url(amqps):
|
|
if not is_amqp_url(amqps):
|
|
msg = f"NOT A VALID AMQP URL {amqps}"
|
|
else:
|
|
msg = f"VERIFY PARAMETERS: is_amqp_url: {is_amqp_url(amqps)}\n amqps[{amqps}]\n origin[{origin}]\n system[{system}]\n service[{service}]\n version[{version}]"
|
|
raise Exception(msg)
|
|
|
|
self.delivery = delivery_mode
|
|
self.payload = ""
|
|
self.hmac = ""
|
|
self.version = version
|
|
self.serviceid = ""
|
|
self.type = "task"
|
|
self.origin = origin.lower()
|
|
self.service = service.lower()
|
|
self.system = system.lower()
|
|
self.client = client.lower()
|
|
self.params = pika.URLParameters(amqps)
|
|
self.callback = callback
|
|
self.queue = f'{self.system}_{self.service}_tasks_{client}' if client != '#' else f'{self.system}_{self.service}_tasks'
|
|
print(f"EXCH2: TASK EXCHANGE: [{self.queue}] [{amqps}]", flush=True)
|
|
if callback is not None:
|
|
self.connection = pika.BlockingConnection(self.params)
|
|
self.channelIn = self.connection.channel() # start main channel
|
|
self.channelIn.exchange_declare (exchange = f"{self.system}.direct", exchange_type = 'direct')
|
|
self.channelIn.queue_declare (queue = self.queue, durable = True)
|
|
self.channelIn.queue_bind (queue = self.queue, exchange = f"{self.system}.direct", routing_key= self.queue)
|
|
self.channelIn.basic_qos (prefetch_count=1)
|
|
self.consumerTag = self.channelIn.basic_consume (queue = self.queue,
|
|
auto_ack = False,
|
|
on_message_callback = self.callback)
|
|
else:
|
|
conexao = pika.BlockingConnection(self.params) #testa conexao
|
|
conexao.close()
|
|
def _setTask(self, msg:str, caller:str):
|
|
now = datetime.now(timezone('America/Sao_Paulo'))
|
|
message =msg.strip('"')
|
|
|
|
self.Msg = {
|
|
"system": self.system,
|
|
"target": self.target,
|
|
"service": self.service,
|
|
"serviceid": self.serviceid,
|
|
"payload": self.payload,
|
|
"hmac": self.hmac,
|
|
"type": self.type,
|
|
"version": self.version,
|
|
"origin": self.origin,
|
|
"client": self.client,
|
|
"caller": caller,
|
|
"logDate": f"{now}",
|
|
"msg": f"{message}"
|
|
}
|
|
|
|
def cancel(self):
|
|
self.channelIn.basic_cancel(self.consumerTag)
|
|
|
|
def sendTask(self, message:str):
|
|
trials = 5
|
|
waitTime = 12
|
|
messageSent = False
|
|
caller = 'TaskExchange2'
|
|
self.target = 'workers'
|
|
self._setTask(message, caller)
|
|
|
|
while not messageSent and trials:
|
|
try:
|
|
connection = pika.BlockingConnection(self.params)
|
|
channelOut = connection.channel() # start a channel
|
|
channelOut.exchange_declare(exchange = f"{self.system}.direct", exchange_type = 'direct')
|
|
channelOut.queue_declare (queue = self.queue, durable = True)
|
|
channelOut.queue_bind (queue = self.queue, exchange = f"{self.system}.direct", routing_key= self.queue)
|
|
|
|
channelOut.basic_publish (exchange = f"{self.system}.direct",
|
|
routing_key = self.queue,
|
|
body = json.dumps(self.Msg),
|
|
properties = pika.BasicProperties(content_type = 'text/plain',
|
|
delivery_mode= self.delivery))
|
|
messageSent = True
|
|
connection.close()
|
|
except Exception as ex:
|
|
print(f"EXCH2: TROUBLE CNX RABBITMQ REMAIN ATTEMPT: {trials}")
|
|
print(ex, flush=True)
|
|
time.sleep(waitTime)
|
|
trials -= 1
|
|
if not messageSent:
|
|
print(f"EXCH2: ERROR TASK NOT SENT TO: {self.queue}", flush=True)
|
|
|
|
def start_consuming(self):
|
|
if self.callback is not None:
|
|
print(f"EXCH2: SEND TASKS TO: [{self.queue}]", flush=True)
|
|
self.consumerTag = self.channelIn.start_consuming()
|
|
|
|
def setCrypto(self, serviceid:str, servicekey:str):
|
|
self.serviceid = serviceid
|
|
self.servicekey = servicekey
|
|
self.Crypto = Payload(serviceid, servicekey)
|
|
|
|
def setPayload(self, data:str):
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.setPayload(data)
|
|
except Exception as ex:
|
|
print(f"EXCH2: CRYPTO ERROR: {ex}", flush=True)
|
|
|
|
def getPayload(self, hmac:str, safeData:str):
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.getPayload(safeData)
|
|
except Exception as ex:
|
|
print(f"EXCH2: CRYPTO ERROR: {ex}", flush=True)
|
|
return False
|
|
|
|
return hmac == self.hmac
|
|
|
|
# RPC - QUANDO FOR UTILIZAR COMO ENDPOINT
|
|
# FORNECER FUNCAO DE TRABALHO NA INICIALIZACAO
|
|
# COMO RPC Server: FILA ONDE RECEBERA AS TAREFAS => SISTEM_SERVICE_RPC ou SISTEM_SERVICE_RPC_CLIENT
|
|
# COMO RPC Client: UTILIZA callRPC
|
|
class RPCExchange2:
|
|
def __init__(self,
|
|
amqps:str, # ENDERECO URL RABBITMQ - CLOUDAMQP
|
|
origin:str, # NOME DO HOST ONDE O SCRIPT ESTÁ RODANDO
|
|
system:str, # NOME DO SISTEMA QUE O SERVIÇO FAZ PARTE
|
|
service:str, # NOME DO SERVIÇO DENTRO DO SISTEMA
|
|
version:str, # VERSÃO DO SERVIÇO
|
|
client:str = '#', # CASO ESPECIFICO A UM CLIENTE, COMO NO CASO DO SG_ETL
|
|
workfunction=None, # FUNCAO DE TRABALHO PARA O USO COMO RCP Server, i.e. para substituir um endpoint fastapi
|
|
delivery_mode=2 # VAI PERSISTIR AS MENSAGENS NO MODO 2
|
|
):
|
|
paramOK = len(service) and len(system) and len(amqps) and len(origin) and len(version)
|
|
|
|
if not (paramOK) or not is_amqp_url(amqps):
|
|
if not is_amqp_url(amqps):
|
|
msg = f"RPCExchange2: NOT A VALID AMQP URL: {amqps}"
|
|
else:
|
|
msg = f"RPCExchange2: VERIFY PARAMETERS: is_amqp_url: {is_amqp_url(amqps)}\n amqps[{amqps}]\n origin[{origin}]\n system[{system}]\n service[{service}]\n version[{version}]"
|
|
raise Exception(msg)
|
|
|
|
self.callback = workfunction
|
|
self.delivery = delivery_mode
|
|
self.payload = ""
|
|
self.hmac = ""
|
|
self.version = version
|
|
self.serviceid = ""
|
|
self.servicekey = ""
|
|
self.correlation_id = ""
|
|
self.consumerTag = None
|
|
self.type = "rpc"
|
|
self.origin = origin.lower()
|
|
self.service = service.lower()
|
|
self.system = system.lower()
|
|
self.client = client.lower()
|
|
self.params = pika.URLParameters(amqps)
|
|
# Server ouve nessa fila - Cliente publica nesta fila
|
|
self.queue = f'{self.system}_{self.service}_rpc_{client}' if client != '#' else f'{self.system}_{self.service}_rpc'
|
|
self.connection = pika.BlockingConnection(self.params)
|
|
self.channel = self.connection.channel() # start main channel
|
|
self.channel.exchange_declare (exchange = f"{self.system}.direct", exchange_type = 'direct')
|
|
|
|
if (workfunction is not None): # FUNCIONAMENTO COMO RPC SERVER
|
|
self.target = 'rpcclt' # SINALIZA O DESTINO DA MSG SER UM RPC CLIENT
|
|
self.channel.queue_declare (queue = self.queue, exclusive = False, durable=True, auto_delete = True)
|
|
self.channel.queue_bind (exchange = f"{self.system}.direct", queue = self.queue, routing_key = self.queue)
|
|
self.channel.basic_qos (prefetch_count = 1)
|
|
self.consumerTag = self.channel.basic_consume(queue = self.queue,
|
|
auto_ack = True,
|
|
on_message_callback = self._RPCResp) # de lá chamamos self.callback
|
|
else: # FUNCIONAMENTO COMO RPC CLIENT
|
|
self.target = 'rpcsrv' # SINALIZA O DESTINO DA MSG SER UM RPC SERVER
|
|
|
|
# PREPARA QUEUE DE RETORNO
|
|
result = self.channel.queue_declare(queue='', exclusive=True, auto_delete=True)
|
|
self.callback_queue = result.method.queue
|
|
self.channel.queue_bind(exchange = f"{self.system}.direct",
|
|
queue = self.callback_queue,
|
|
routing_key = self.callback_queue)
|
|
|
|
self.channel.basic_consume(
|
|
queue = self.callback_queue,
|
|
on_message_callback = self._on_response,
|
|
auto_ack = True
|
|
)
|
|
#print("EXCH2: RPC CLT - QUEUE:", self.callback_queue, flush=True)
|
|
|
|
# RPC - QUANDO FOR UTILIZAR COMO ENDPOINT
|
|
def _RPCResp(self, ch, method, props, body):
|
|
#ch.basic_ack(delivery_tag = method.delivery_tag)
|
|
message = body.decode("utf-8")
|
|
reply_to = getattr(props, 'reply_to', None)
|
|
correlation_id = getattr(props, 'correlation_id', None)
|
|
resposta = "{}" # SEM WORKFUNCTION
|
|
# ENVIA A MENSAGEM AO CALLBACK
|
|
self.payload = ""
|
|
self.hmac = ""
|
|
try:
|
|
if self.callback is not None:
|
|
resposta = self.callback(message)
|
|
rspObj = json.loads(resposta)
|
|
# para enviar um payload criptografado na resposta ao caller
|
|
# retorne um objeto contendo serviceid, servicekey e payload
|
|
# essas chaves serao retiradas do objeto e o restante da informacao
|
|
# irá no chave msg
|
|
if 'payload' in rspObj and 'serviceid' in rspObj and 'servicekey' in rspObj:
|
|
self.setCrypto(rspObj['serviceid'], rspObj['servicekey'])
|
|
self.setPayload(json.dumps(rspObj['payload']))
|
|
rspObj.pop('payload', None)
|
|
rspObj.pop('serviceid', None)
|
|
rspObj.pop('servicekey', None)
|
|
resposta = json.dumps(rspObj)
|
|
|
|
|
|
except Exception as ex:
|
|
erro = f"ERROR CALLBACK: {str(ex)}"
|
|
print(erro, flush=True)
|
|
resposta = json.dumps({"status": "error", "message": erro})
|
|
|
|
# ENVIA RESPOSTA AO CALLER
|
|
msg = json.loads(message)
|
|
msg = msg['msg']
|
|
#print(f"RPC <== {msg}", flush=True)
|
|
#print(f"RPC ==> {resposta}", flush=True)
|
|
try:
|
|
if reply_to:
|
|
self.target = "rpcclt"
|
|
self._setJSON(resposta, 'RPCResp')
|
|
|
|
ch.basic_publish(exchange = f"{self.system}.direct",
|
|
routing_key = reply_to,
|
|
body = json.dumps(self.Msg),
|
|
properties = pika.BasicProperties(content_type = 'application/json',
|
|
delivery_mode = self.delivery,
|
|
correlation_id = correlation_id))
|
|
else:
|
|
print(f"RPCExchange2: NO TARGET FOUND: {resposta}", props, flush=True)
|
|
|
|
except Exception as ex:
|
|
print(ex)
|
|
resposta = f"RPCExchange2: ERRO MSG CALLER: {str(ex)}"
|
|
raise Exception(resposta)
|
|
|
|
|
|
def _setJSON(self, msg:str, caller:str):
|
|
now = datetime.now(timezone('America/Sao_Paulo'))
|
|
message =msg.strip('"')
|
|
|
|
self.Msg = {
|
|
"system": self.system,
|
|
"target": self.target,
|
|
"service": self.service,
|
|
"serviceid": self.serviceid,
|
|
"payload": self.payload,
|
|
"hmac": self.hmac,
|
|
"type": self.type,
|
|
"correlation_id": self.correlation_id,
|
|
"version": self.version,
|
|
"origin": self.origin,
|
|
"client": self.client,
|
|
"caller": caller,
|
|
"logDate": f"{now}",
|
|
"msg": f"{message}"
|
|
}
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
|
|
|
|
def _on_response(self, ch, method, properties, body):
|
|
try:
|
|
if self.correlation_id == properties.correlation_id:
|
|
self.response = body.decode("utf-8")
|
|
except Exception as ex:
|
|
print(f"RPCExchange2: RPC ERROR: {ex}", flush=True)
|
|
sys.exit(0)
|
|
|
|
return self.response
|
|
|
|
# ENVIA A MENSAGEM AO CALLBACK
|
|
# RPC CLIENT CHAMA O RPC SERVER
|
|
def callRPC(self, message:str, timeout:int=10):
|
|
# ENVIA A MENSAGEM
|
|
self.response = None
|
|
self.correlation_id = str(uuid.uuid4())
|
|
self._setJSON(message, 'callRPC') # COLOCA A message DENTRO DA CHAVE msg em self.Msg
|
|
|
|
self.channel.exchange_declare(exchange = f"{self.system}.direct", exchange_type = 'direct')
|
|
self.channel.queue_declare (queue= self.queue, durable=True, exclusive=False, auto_delete=True)
|
|
self.channel.queue_bind (queue= self.queue, exchange= f"{self.system}.direct", routing_key= self.queue)
|
|
|
|
self.channel.basic_publish (exchange = f"{self.system}.direct",
|
|
routing_key = self.queue,
|
|
body = json.dumps(self.Msg),
|
|
properties = pika.BasicProperties(content_type = 'application/json',
|
|
delivery_mode = self.delivery,
|
|
reply_to = self.callback_queue,
|
|
correlation_id= self.correlation_id))
|
|
# AGUARDA O RETORNO
|
|
while self.response is None:
|
|
self.connection.process_data_events(time_limit=timeout)
|
|
return self.response
|
|
|
|
def cancel(self):
|
|
print(f"RPCExchange2: CONSUMER TAG CANCELED: {self.consumerTag}", flush=True)
|
|
self.channel.basic_cancel(self.consumerTag)
|
|
|
|
def start_consuming(self):
|
|
if self.callback is not None:
|
|
print(f"RPCExchange2: {self.service.upper()} RPC SRV QUEUE: [{self.queue}]", flush=True)
|
|
self.channel.start_consuming()
|
|
|
|
def setCrypto(self, serviceid:str, servicekey:str):
|
|
self.serviceid = serviceid
|
|
self.servicekey = servicekey
|
|
self.Crypto = Payload(serviceid, servicekey)
|
|
|
|
def setPayload(self, data:str):
|
|
print(f"RPCExchange2: SET PAYLOAD", flush=True)
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.setPayload(data)
|
|
except Exception as ex:
|
|
print(ex)
|
|
|
|
def getPayload(self, hmac:str, safeData:str):
|
|
print(f"RPCExchange2: GET PAYLOAD", flush=True)
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.getPayload(safeData)
|
|
except Exception as ex:
|
|
print(ex)
|
|
return False
|
|
|
|
return hmac == self.hmac
|
|
|
|
# CADA APLICAÇÃO CRIA UMA CLASSE QUE ESTENDE IC2Builder IMPLEMENTANDO
|
|
# OS METODOS ABSTRATOS DE CADA REALMTYPE
|
|
class IC2Builder(ABC):
|
|
@abstractmethod
|
|
def _setLogs(self, message: tuple) -> InfoConnector: ...
|
|
@abstractmethod
|
|
def _setErrors(self, message: tuple) -> InfoConnector: ...
|
|
@abstractmethod
|
|
def _setOperational(self, message: tuple) -> InfoConnector: ...
|
|
@abstractmethod
|
|
def _setStatus(self, message: tuple) -> InfoConnector: ...
|
|
@abstractmethod
|
|
def _setMetrics(self, message: tuple) -> InfoConnector: ...
|
|
|
|
def build(self, realm: RealmType, message: Sequence[Any]) -> InfoConnector:
|
|
handlers = {
|
|
RealmType.LOGS: self._setLogs,
|
|
RealmType.ERRORS: self._setErrors,
|
|
RealmType.OPERATIONAL: self._setOperational,
|
|
RealmType.STATUS: self._setStatus,
|
|
RealmType.METRICS: self._setMetrics,
|
|
}
|
|
|
|
try:
|
|
return handlers[realm](message)
|
|
except KeyError:
|
|
raise ValueError(f"RealmType não suportado: {realm}")
|
|
|
|
class LoggerExchange2:
|
|
def __enter__(self): return self
|
|
def __exit__(self, exc_type, exc, tb): self.close()
|
|
|
|
def __init__(self,
|
|
ic2_builder: IC2Builder,
|
|
amqps : str,
|
|
system : str,
|
|
service : str,
|
|
version : str,
|
|
origin : str = "",
|
|
client : str = "#",
|
|
logType : str = 'logs',
|
|
logLocal : bool = False,
|
|
callback : object = None):
|
|
self.callback = callback
|
|
self.ic2_builder = ic2_builder
|
|
self.system = system.lower()
|
|
self.target = ""
|
|
self.service = service.lower()
|
|
self.serviceid = "AF1699F719F1B6C81DE2BD38AB811994E3089B35C8E9314813A770AC7AE25D8D"
|
|
self.servicekey = "KoG704lAafBA7Yafj8jyg5lAHfarrQ12"
|
|
self.payload = ""
|
|
self.hmac = ""
|
|
self.type = "logger"
|
|
self.version = version
|
|
self.origin = origin
|
|
self.client = client.lower()
|
|
self.params = pika.URLParameters(amqps)
|
|
self.params.socket_timeout = 15
|
|
self.params.heartbeat = 60
|
|
self.logType = logType.lower()
|
|
self.logTypeSvd = logType.lower()
|
|
self.logLocal = logLocal
|
|
self.pp = pprint.PrettyPrinter()
|
|
self.diaMes = datetime.now().strftime("%d") # PARA SABER SE ATUALIZA OU NÃO O ARQUIVO DE LOG LOCAL
|
|
self.flush = True
|
|
self.base_routeKey = f"{self.service}.{self.system}"
|
|
self.connection = None
|
|
self.channel = None
|
|
self.logging = logging.getLogger(self.base_routeKey)
|
|
self.Crypto = Payload(self.serviceid, self.servicekey)
|
|
self.exchange_name = 'info_connector.topic'
|
|
|
|
try:
|
|
self._connect()
|
|
except Exception as err:
|
|
self.logging.error(f"ERROR CONNECTION {err}")
|
|
raise
|
|
|
|
if self.logLocal: self.setLogLocalName()
|
|
|
|
# --- async logging ---
|
|
self.queue_size = 1000
|
|
self.queue = Queue(maxsize=self.queue_size)
|
|
self.stop_event = Event()
|
|
self.worker_thread = Thread(
|
|
target = self._worker,
|
|
name = "logger-worker",
|
|
daemon = True
|
|
)
|
|
self.worker_thread.start()
|
|
|
|
self.logging.info(f"params: {self.params} amqps: {amqps}")
|
|
if self.callback is not None:
|
|
try:
|
|
self.queueIC2 = "info_connector_collector"
|
|
self.channelIn = self.connection.channel()
|
|
self.channelIn.queue_declare (queue = self.queueIC2, exclusive=True, durable=True)
|
|
self.channelIn.queue_bind (queue = self.queueIC2, exchange = self.exchange_name, routing_key = "#")
|
|
self.channelIn.basic_qos (prefetch_count = 1)
|
|
self.consumerTag = self.channelIn.basic_consume (queue = self.queueIC2,
|
|
auto_ack = False,
|
|
on_message_callback = self.callback)
|
|
self.logging.info("COLLECTING TOPIC MESSAGES VHOST info_connector_host EXCHANGE info_connector.topic KEY #")
|
|
except Exception as error:
|
|
raise(error)
|
|
else:
|
|
conexao = pika.BlockingConnection(self.params) #testa conexao
|
|
conexao.close()
|
|
self.logging.info(f"Logger Messages topics: {self.logType}.{self.base_routeKey}.{self.client}")
|
|
|
|
|
|
# PERSISTENCIA LOGS MESSGERIA
|
|
def _worker(self):
|
|
while not self.stop_event.is_set() or not self.queue.empty():
|
|
try:
|
|
payload, client = self.queue.get(timeout=0.5)
|
|
except Empty:
|
|
continue
|
|
|
|
try:
|
|
self.sendMsgWork(payload, client)
|
|
except Exception as e:
|
|
# última linha de defesa
|
|
self.logging.error(f"ASYNC LOGGER FAILURE: {e}")
|
|
if self.logLocal:
|
|
self.logging.info(payload)
|
|
finally:
|
|
self.queue.task_done()
|
|
|
|
# PERSISTENCIA CONEXAO AMQPS
|
|
def _connect(self):
|
|
"""Conecta ou reconecta ao RabbitMQ."""
|
|
if self.connection and self.connection.is_open:
|
|
return
|
|
try:
|
|
if self.connection:
|
|
self.connection.close()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
self.connection = pika.BlockingConnection(self.params)
|
|
self.channel = self.connection.channel()
|
|
self.channel.exchange_declare(exchange = self.exchange_name,
|
|
exchange_type = 'topic',
|
|
auto_delete = False)
|
|
|
|
self.logging.info("Conexão ao RabbitMQ estabelecida.")
|
|
except pika.exceptions.AMQPConnectionError as e:
|
|
self.logging.error(f"Falha na conexão: {e}")
|
|
raise
|
|
|
|
def _setMsg(self, msg:str, caller:str):
|
|
now = datetime.now(timezone('America/Sao_Paulo')) # melhoria: acrescentar timezone info no system_config
|
|
return {
|
|
"system" : self.system,
|
|
"target" : self.target,
|
|
"service" : self.service,
|
|
"serviceid": self.serviceid,
|
|
"payload" : self.payload,
|
|
"hmac" : self.hmac,
|
|
"type" : self.type,
|
|
"version" : self.version,
|
|
"origin" : self.origin,
|
|
"client" : self.client,
|
|
"caller" : caller,
|
|
"logDate" : f"{now}",
|
|
"msg" : msg
|
|
}
|
|
|
|
def _sendMsg(self, msg: str, caller: str, *args):
|
|
# Avalia arquivo de logs
|
|
if self.logLocal:
|
|
current_dia = datetime.now().strftime("%d")
|
|
if current_dia != self.diaMes:
|
|
self.diaMes = current_dia
|
|
self.setLogLocalName()
|
|
|
|
message = msg.strip()
|
|
|
|
try:
|
|
if args and isinstance(args[0], (list, tuple)):
|
|
for item in args[0]:
|
|
msgArg = pprint.pformat(item).replace("'", "").strip()
|
|
message = f"{message}, {msgArg}"
|
|
|
|
Msg = self._setMsg(message, caller)
|
|
payload = json.dumps(Msg)
|
|
|
|
# console / docker logs continuam síncronos
|
|
self.pp.pprint(f"{Msg['logDate'][0:21]} {Msg['msg']}")
|
|
|
|
# log local imediato (garantia mínima)
|
|
if self.logLocal:
|
|
Msg['payload'] = Msg['payload'][:20]
|
|
self.logging.info(self.pp.pformat(Msg))
|
|
|
|
# envio assíncrono
|
|
try:
|
|
self.queue.put_nowait((payload, self.client))
|
|
except Full:
|
|
# fila cheia → degrada com dignidade
|
|
self.logging.warning("LOGGER QUEUE FULL — DROPPING MESSAGE")
|
|
|
|
if self.flush:
|
|
sys.stdout.flush()
|
|
|
|
except Exception as e:
|
|
erro = f"LOGGER ERROR: {e}"
|
|
self.logging.error(erro)
|
|
print(erro, flush=True)
|
|
|
|
|
|
def _sendMsgLocal(self, msg: str, caller: str, *args):
|
|
# Avalia arquivo de logs
|
|
if self.logLocal:
|
|
current_dia = datetime.now().strftime("%d")
|
|
if current_dia != self.diaMes:
|
|
self.diaMes = current_dia
|
|
self.setLogLocalName()
|
|
|
|
message = msg.strip()
|
|
|
|
try:
|
|
if args and isinstance(args[0], (list, tuple)):
|
|
for item in args[0]:
|
|
msgArg = pprint.pformat(item).replace("'", "").strip()
|
|
message = f"{message}, {msgArg}"
|
|
|
|
Msg = self._setMsg(message, caller)
|
|
payload = json.dumps(Msg)
|
|
|
|
# console / docker logs continuam síncronos
|
|
self.pp.pprint(f"{Msg['logDate'][0:21]} {Msg['msg']}")
|
|
|
|
# log local imediato (garantia mínima)
|
|
if self.logLocal:
|
|
Msg['payload'] = Msg['payload'][:20]
|
|
self.logging.info(self.pp.pformat(Msg))
|
|
|
|
except Exception as e:
|
|
erro = f"LOGGER ERROR: {e}"
|
|
self.logging.error(erro)
|
|
print(erro, flush=True)
|
|
|
|
def start_consuming(self):
|
|
if self.callback is not None:
|
|
self.consumerTag = self.channelIn.start_consuming()
|
|
|
|
def close(self):
|
|
# sinaliza parada
|
|
self.stop_event.set()
|
|
|
|
# aguarda flush da fila (timeout defensivo)
|
|
try:
|
|
self.queue.join()
|
|
except Exception:
|
|
pass
|
|
|
|
if self.worker_thread.is_alive():
|
|
self.worker_thread.join(timeout=2)
|
|
|
|
if self.channel:
|
|
self.channel.close()
|
|
if self.connection:
|
|
self.connection.close()
|
|
|
|
def setLogLocalName(self):
|
|
self.logLocalName = 'logs/'+ self.system.split()[0] + '_' + self.diaMes
|
|
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
|
fhandler = logging.FileHandler(self.logLocalName)
|
|
fhandler.setLevel(logging.INFO)
|
|
fhandler.setFormatter(formatter)
|
|
self.logging.propagate = False
|
|
self.logging.setLevel(logging.INFO)
|
|
if not any(isinstance(h, logging.FileHandler) for h in self.logging.handlers):
|
|
self.logging.addHandler(fhandler)
|
|
|
|
# Funcionamento: caso nao seja possivel enviar a msg, tentar nova conexao
|
|
# e envio da msg original mais uma vez
|
|
def sendMsgWork(self, message:str, client:str = "#"):
|
|
messageSent = False
|
|
trials = 5
|
|
waitTime = 5
|
|
while not messageSent and trials > 0:
|
|
self._connect()
|
|
# Cliente informado - MAIOR PRECEDENCIA
|
|
if client != "#":
|
|
routeKey = self.base_routeKey + f".{client.lower()}"
|
|
elif self.client != "#":
|
|
routeKey = self.base_routeKey + f".{self.client.lower()}"
|
|
else:
|
|
routeKey = self.base_routeKey
|
|
|
|
try:
|
|
routing_key = f"{self.logType}.{routeKey}"
|
|
self.channel.basic_publish(
|
|
exchange = "info_connector.topic",
|
|
routing_key = routing_key,
|
|
body = message,
|
|
properties = pika.BasicProperties(
|
|
content_type = 'text/plain',
|
|
delivery_mode= pika.DeliveryMode.Transient))
|
|
messageSent = True
|
|
except (pika.exceptions.ConnectionClosed, pika.exceptions.ChannelClosed, pika.exceptions.AMQPConnectionError) as ex:
|
|
self.logging.warning(f"TROUBLE CNX LOGGER REMAIN ATTEMPT: {trials} {ex}")
|
|
self._connect()
|
|
except Exception as ex:
|
|
self.logging.error(f"LOGGER ERROR: {ex}")
|
|
self.logging.warning(str(self.params))
|
|
finally:
|
|
if not messageSent:
|
|
trials -= 1
|
|
time.sleep(waitTime)
|
|
waitTime *= 2
|
|
|
|
if not messageSent:
|
|
self.logging.error(f"LOGGER ERROR: {message}")
|
|
|
|
# Para quando desejar trocar o header do topico
|
|
# Salva o tópico padrão especificado no momento da inicialização
|
|
def setLogType(self, logType:str):
|
|
if not logType or len(logType) > 25:
|
|
return
|
|
self.logTypeSvd = self.logType
|
|
self.logType = logType.lower()
|
|
|
|
# Volta ao header topico padrao
|
|
def clrLogType(self):
|
|
self.logType = self.logTypeSvd
|
|
|
|
def logIC2(self, realm: RealmType, msg: tuple = ()):
|
|
ic_obj = self.ic2_builder.build(realm, msg)
|
|
payload = ic_obj.model_dump_json()
|
|
self.setPayload(data=payload)
|
|
self.setLogType(logType=realm.value)
|
|
self._sendMsg(f"LOGIC2 {realm.value.upper()} [{str(msg)[:20]} ...]", caller="IC2")
|
|
self.clrLogType()
|
|
self.clrPayload()
|
|
|
|
def logMsgError(self, msg:str, *args):
|
|
self.setLogType('error.logs')
|
|
caller = inspect.stack()[1].function
|
|
if caller == '<module>': caller = 'main'
|
|
|
|
self._sendMsgLocal(msg, caller, args)
|
|
self.clrLogType()
|
|
|
|
# LOGS GERAIS
|
|
# INFORM logs.<service>.<system>.#
|
|
# Log geral de acordo com o valor logType
|
|
def logMsg (self, msg: str, *args):
|
|
caller = inspect.stack()[1].function
|
|
if caller == '<module>': caller = 'main'
|
|
Msg = f"{msg}"
|
|
self._sendMsgLocal(Msg, caller, args)
|
|
|
|
# FUNÇÕES PARA CRYPTOGRAFIA
|
|
def clrPayload(self):
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
|
|
def setPayload(self, data:str):
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.setPayload(data)
|
|
except Exception as ex:
|
|
print(f"EXCH2: CRYPTO ERROR: {ex}", flush=True)
|
|
|
|
def getPayload(self, hmac:str, safeData:str):
|
|
try:
|
|
self.hmac = ""
|
|
self.payload = ""
|
|
self.hmac, self.payload = self.Crypto.getPayload(safeData)
|
|
except Exception as ex:
|
|
print(f"EXCH2: CRYPTO ERROR: {ex}", flush=True)
|
|
return False
|
|
|
|
return hmac == self.hmac
|
|
|
|
class AppIC2Builder(IC2Builder):
|
|
def __init__(self, service_name: str, versionstr: str, origin: str):
|
|
self.service_name = service_name
|
|
self.versionstr = versionstr
|
|
self.origin = origin
|
|
|
|
def _get_common_fields(self):
|
|
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
return {
|
|
"log_date": now,
|
|
"service_date": now,
|
|
"service": self.service_name,
|
|
"versionstr": self.versionstr,
|
|
"origin": self.origin
|
|
}
|
|
|
|
def _setLogs(self, message: tuple) -> InfoConnector:
|
|
msg_str = str(message[0]) if len(message) > 0 else ""
|
|
ent_id = str(message[1]) if len(message) > 1 else "0"
|
|
ent_name = str(message[2]) if len(message) > 2 else "Unknown"
|
|
tenant = str(message[3]) if len(message) > 3 else "#"
|
|
|
|
ic_msg = ICLogs(**self._get_common_fields(), enterprise_id=ent_id, enterprise_name=ent_name, message=msg_str, tenant=tenant)
|
|
return InfoConnector(realm=RealmType.LOGS, messageIC=ic_msg)
|
|
|
|
def _setErrors(self, message: tuple) -> InfoConnector:
|
|
msg_str = str(message[0]) if len(message) > 0 else ""
|
|
ent_id = str(message[1]) if len(message) > 1 else "0"
|
|
ent_name = str(message[2]) if len(message) > 2 else "Unknown"
|
|
tenant = str(message[3]) if len(message) > 3 else "#"
|
|
|
|
ic_msg = ICErrors(**self._get_common_fields(), enterprise_id=ent_id, enterprise_name=ent_name, message=msg_str, tenant=tenant)
|
|
return InfoConnector(realm=RealmType.ERRORS, messageIC=ic_msg)
|
|
|
|
def _setOperational(self, message: tuple) -> InfoConnector:
|
|
acao = str(message[0]) if len(message) > 0 else ""
|
|
tabela = str(message[1]) if len(message) > 1 else ""
|
|
id_usuario = str(message[2]) if len(message) > 2 else "0"
|
|
id_registro = str(message[3]) if len(message) > 3 else "0"
|
|
status_str = str(message[4]).upper() if len(message) > 4 else "INICIO"
|
|
mensagem = str(message[5]) if len(message) > 5 else ""
|
|
json_data = str(message[6]) if len(message) > 6 else "{}"
|
|
|
|
if status_str == 'SUCESSO': op_status = OpStatus.OK
|
|
elif status_str == 'ERRO': op_status = OpStatus.ERROR
|
|
else: op_status = OpStatus.ACTIVE
|
|
|
|
common = self._get_common_fields()
|
|
ic_msg = ICOperational(
|
|
log_date=common["log_date"], service_date=common["service_date"], service_name=self.service_name,
|
|
versionstr=self.versionstr, origin=self.origin, enterprise_id=id_usuario, enterprise_name=f"User {id_usuario}",
|
|
status=op_status, step_status=acao, message=f"{mensagem} | Tabela: {tabela} | ID_Reg: {id_registro} | Data: {json_data}"
|
|
)
|
|
return InfoConnector(realm=RealmType.OPERATIONAL, messageIC=ic_msg)
|
|
|
|
def _setStatus(self, message: tuple) -> InfoConnector:
|
|
common = self._get_common_fields()
|
|
ic_msg = ICStatus( service=self.service_name, versionstr=self.versionstr, system_name="integracao",
|
|
startup_date=common["log_date"], access_date=common["log_date"], origin=self.origin )
|
|
return InfoConnector(realm=RealmType.STATUS, messageIC=ic_msg)
|
|
|
|
def _setMetrics(self, message: tuple) -> InfoConnector:
|
|
metric_name = str(message[0]) if len(message) > 0 else ""
|
|
value = str(message[1]) if len(message) > 1 else "0"
|
|
unit = str(message[2]) if len(message) > 2 else ""
|
|
|
|
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
ic_msg = ICMetrics( metric_date=now, metric=metric_name, metric_name=metric_name, value=value, unit=unit, tenant="#" )
|
|
return InfoConnector(realm=RealmType.METRICS, messageIC=ic_msg)
|