126 lines
7.8 KiB
Python
126 lines
7.8 KiB
Python
import json as json_lib
|
|
from utils.api import Api
|
|
from utils.banco import Banco, NFeTable
|
|
from Models.nfe import NFe
|
|
from typing import List
|
|
from utils.Exchange2 import LoggerExchange2
|
|
from utils.exchangeModels import RealmType
|
|
from utils.notificador import NotificadorInterno
|
|
|
|
class RegistroNFeService:
|
|
# Classe responsável por fornecer as operações de comunicação com a PlugNotas para o documento NFe
|
|
def __init__(self, api_key: str, url_base: str, banco: Banco, logger: LoggerExchange2 = None):
|
|
self.api = Api(api_key)
|
|
self.url = f"{url_base}/nfe"
|
|
self.banco = banco
|
|
self.logger = logger
|
|
self.notificador = NotificadorInterno(logger=logger) if logger else None
|
|
|
|
def cadastrar_nfe(self, nfe_list: List[NFe], data: dict):
|
|
# Envia uma lista de NFes em lote ou unitária para emissão e registra no banco de dados
|
|
try:
|
|
id_empresa = data.get('id_empresa', "0")
|
|
payload = [nfe.model_dump() for nfe in nfe_list]
|
|
resposta = self.api.post(self.url, payload)
|
|
if 'error' in resposta:
|
|
id_usr = data.get('id_usuario', 1)
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('INSERT', 'nfe', id_usr, '0', 'ERRO', f"Erro na API: {resposta.get('error')}", json_lib.dumps(resposta)))
|
|
return resposta
|
|
if 'documents' in resposta:
|
|
for doc in resposta['documents']:
|
|
id_tecnospeed = doc.get('id')
|
|
if id_tecnospeed:
|
|
nfe_nova = NFeTable(
|
|
id_tecnospeed=id_tecnospeed,
|
|
id_empresa=id_empresa,
|
|
id_usuario=data.get('id_usuario', 1),
|
|
status=True
|
|
)
|
|
self.banco.add(nfe_nova)
|
|
id_usr = data.get('id_usuario', 1)
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('INSERT', 'nfe', id_usr, str(nfe_nova.id), 'SUCESSO', 'NFe cadastrada com sucesso', json_lib.dumps(doc)))
|
|
|
|
# Notifica o listener interno para sincronização imediata
|
|
if self.notificador:
|
|
self.notificador.notificar_emissao(id_tecnospeed=id_tecnospeed, documento="nfe", cnpj=id_empresa)
|
|
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('INSERT', 'nfe', data.get('id_usuario', 1), '0', 'ERRO', f"Erro ao cadastrar NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao cadastrar NFe: {e}")
|
|
|
|
def cancelar_nfe(self, id_nota: str, justificativa: str):
|
|
# Solicita o cancelamento de uma NFe existente mediante uma justificativa
|
|
try:
|
|
resposta = self.api.post(f"{self.url}/{id_nota}/cancelamento", {"justificativa": justificativa})
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('DELETE', 'nfe', 1, id_nota, 'SUCESSO', 'NFe cancelada com sucesso', json_lib.dumps(resposta)))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('DELETE', 'nfe', 1, id_nota, 'ERRO', f"Erro ao cancelar NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao cancelar NFe: {e}")
|
|
|
|
def enviar_cce(self, id_nota: str, correcao: str):
|
|
# Envia uma Carta de Correção Eletrônica (CCe) para retificar uma NFe já emitida
|
|
try:
|
|
resposta = self.api.post(f"{self.url}/{id_nota}/cce", {"correcao": correcao})
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('UPDATE', 'nfe', 1, id_nota, 'SUCESSO', 'CCe enviada com sucesso', json_lib.dumps(resposta)))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('UPDATE', 'nfe', 1, id_nota, 'ERRO', f"Erro ao enviar CCe NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao enviar CCe NFe: {e}")
|
|
|
|
def consultar_resumo(self, id_nota: str):
|
|
# Consulta o resumo e o status atual da NFe na PlugNotas pelo seu ID
|
|
try:
|
|
resposta = self.api.get(f"{self.url}/{id_nota}/resumo")
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', 1, id_nota, 'SUCESSO', 'Resumo de NFe consultado', json_lib.dumps(resposta)))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', 1, id_nota, 'ERRO', f"Erro ao consultar resumo NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao consultar resumo NFe: {e}")
|
|
|
|
def consultar_resumo_por_cnpj_integracao(self, cnpj: str, id_integracao: str):
|
|
try:
|
|
resposta = self.api.get(f"{self.url}/{cnpj}/{id_integracao}/resumo")
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', id_integracao, id_integracao, 'SUCESSO', 'Resumo por CNPJ/Integração consultado', json_lib.dumps(resposta)))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', id_integracao, id_integracao, 'ERRO', f"Erro ao consultar resumo NFe por CNPJ/Integração: {e}", "{}"))
|
|
raise Exception(f"Erro ao consultar resumo NFe por CNPJ/Integração: {e}")
|
|
|
|
def consultar_status_cancelamento(self, id_nota: str):
|
|
try:
|
|
resposta = self.api.get(f"{self.url}/{id_nota}/cancelamento/status")
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', id_nota, id_nota, 'SUCESSO', 'Status de cancelamento consultado', json_lib.dumps(resposta)))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', id_nota, id_nota, 'ERRO', f"Erro ao consultar status de cancelamento NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao consultar status de cancelamento NFe: {e}")
|
|
|
|
def consultar_status_cce(self, id_nota: str):
|
|
try:
|
|
resposta = self.api.get(f"{self.url}/{id_nota}/cce/status")
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', id_nota, id_nota, 'SUCESSO', 'Status CCe consultado', json_lib.dumps(resposta)))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('SELECT', 'nfe', id_nota, id_nota, 'ERRO', f"Erro ao consultar status CCe NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao consultar status CCe NFe: {e}")
|
|
|
|
def baixar_pdf(self, id_nota: str):
|
|
# Obtém o link para download ou o stream do PDF da Nota Fiscal processada
|
|
try:
|
|
resposta = self.api.get(f"{self.url}/{id_nota}/pdf")
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('DOWNLOAD (PDF)', 'nfe', id_nota, id_nota, 'SUCESSO', 'PDF baixado com sucesso', json_lib.dumps(resposta) if isinstance(resposta, dict) else '{}'))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('DOWNLOAD (PDF)', 'nfe', id_nota, id_nota, 'ERRO', f"Erro ao baixar PDF NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao baixar PDF NFe: {e}")
|
|
|
|
def baixar_xml(self, id_nota: str):
|
|
try:
|
|
resposta = self.api.get(f"{self.url}/{id_nota}/xml")
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('DOWNLOAD (XML)', 'nfe', id_nota, id_nota, 'SUCESSO', 'XML baixado com sucesso', json_lib.dumps(resposta) if isinstance(resposta, dict) else '{}'))
|
|
return resposta
|
|
except Exception as e:
|
|
if self.logger: self.logger.logIC2(RealmType.OPERATIONAL, ('DOWNLOAD (XML)', 'nfe', id_nota, id_nota, 'ERRO', f"Erro ao baixar XML NFe: {e}", "{}"))
|
|
raise Exception(f"Erro ao baixar XML NFe: {e}")
|