285 lines
12 KiB
Python
285 lines
12 KiB
Python
|
"""
|
||
|
|
Respuesta al proveedor: arma y envía el correo de aceptación o rechazo (plantillas
|
||
|
|
``plantilla/plantilla-aceptado.html`` y ``plantilla-rechazado.html``) a quien envió
|
||
|
|
el correo con el XML y el PDF.
|
||
|
|
|
||
|
|
Dos modos, según ``NOTIFICACION_MODO`` en el .env:
|
||
|
|
|
||
|
|
prueba (por defecto) la respuesta NO sale al proveedor: va al equipo
|
||
|
|
(``NOTIFICACION_EQUIPO``) con el asunto "[PRUEBA → remitente]".
|
||
|
|
real la respuesta va al remitente, con copia oculta al equipo.
|
||
|
|
|
||
|
|
Como el sistema contesta solo y a terceros, **no responde** a: remitentes
|
||
|
|
desconocidos o sin dominio válido, el propio buzón, direcciones ``no-reply`` y
|
||
|
|
afines, mensajes automáticos (respuestas automáticas, rebotes) y, en correos sin
|
||
|
|
comprobante, a respuestas (``Re:``) o mensajes que citan un código nuestro. Además
|
||
|
|
limita a ``MAX_RESPUESTAS_CORREO`` por remitente cada 24 h las respuestas a correos
|
||
|
|
sin comprobante.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import logging
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
from config_validator import MODOS_NOTIFICACION as MODOS
|
||
|
|
import observaciones as obs
|
||
|
|
import outlook_utilities
|
||
|
|
import plantillas
|
||
|
|
import seguimiento
|
||
|
|
from pipeline_utils import log
|
||
|
|
|
||
|
|
EQUIPO_POR_DEFECTO = ["jlopez@aberio.com.pe", "gcalle@pucp.pe"]
|
||
|
|
MAX_RESPUESTAS_CORREO = 3
|
||
|
|
|
||
|
|
_DIRECCION = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||
|
|
_AUTOMATICA = re.compile(r"no[-_.]?reply|do[-_.]?not[-_.]?reply|no[-_.]?responder|mailer-daemon|postmaster|bounce", re.I)
|
||
|
|
_CODIGO = re.compile(r"ABR-\d{6}-\d{5}")
|
||
|
|
_RESPUESTA = re.compile(r"^\s*(re|rv)\s*:", re.I)
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Configuración (.env)
|
||
|
|
# ============================================================================
|
||
|
|
def modo() -> str:
|
||
|
|
"""``prueba`` o ``real``. Cualquier otro valor (o ninguno) es ``prueba``."""
|
||
|
|
valor = os.getenv("NOTIFICACION_MODO", "prueba").strip().lower()
|
||
|
|
return valor if valor in MODOS else "prueba"
|
||
|
|
|
||
|
|
|
||
|
|
def equipo() -> list[str]:
|
||
|
|
"""Direcciones internas (``NOTIFICACION_EQUIPO``, separadas por coma o punto y coma)."""
|
||
|
|
lista = [d.strip() for d in re.split(r"[;,]", os.getenv("NOTIFICACION_EQUIPO", "")) if d.strip()]
|
||
|
|
return lista or list(EQUIPO_POR_DEFECTO)
|
||
|
|
|
||
|
|
|
||
|
|
def contacto_correo() -> str:
|
||
|
|
return os.getenv("NOTIFICACION_CONTACTO", "").strip() or os.getenv("EMAIL", "").strip()
|
||
|
|
|
||
|
|
|
||
|
|
def envio_correo() -> str:
|
||
|
|
"""Buzón al que el proveedor debe reenviar el comprobante corregido."""
|
||
|
|
return os.getenv("EMAIL", "").strip()
|
||
|
|
|
||
|
|
|
||
|
|
def logo_url() -> str:
|
||
|
|
return os.getenv("LOGO_URL", "").strip()
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Protecciones
|
||
|
|
# ============================================================================
|
||
|
|
def motivo_omision(email: dict, buzon: str, sin_comprobante: bool = False) -> str | None:
|
||
|
|
"""
|
||
|
|
Motivo por el que NO se debe responder a este correo, o None si se puede.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
email: Dict del correo (``sender``, ``subject``, ``es_automatico``).
|
||
|
|
buzon: Dirección del buzón del sistema (``EMAIL``).
|
||
|
|
sin_comprobante: True si la respuesta es un rechazo de un correo sin comprobante válido.
|
||
|
|
"""
|
||
|
|
remitente = (email.get("sender") or "").strip()
|
||
|
|
if not remitente or remitente == "Desconocido":
|
||
|
|
return "remitente desconocido"
|
||
|
|
if not _DIRECCION.match(remitente):
|
||
|
|
return f"dirección no válida ({remitente})"
|
||
|
|
if remitente.lower() == buzon.strip().lower():
|
||
|
|
return "el remitente es el propio buzón"
|
||
|
|
if _AUTOMATICA.search(remitente.split("@")[0]):
|
||
|
|
return "dirección automática (no-reply)"
|
||
|
|
if email.get("es_automatico"):
|
||
|
|
return "mensaje automático (respuesta automática o rebote)"
|
||
|
|
if sin_comprobante:
|
||
|
|
asunto = email.get("subject") or ""
|
||
|
|
if _RESPUESTA.match(asunto) or _CODIGO.search(asunto):
|
||
|
|
return "parece una respuesta a una notificación nuestra"
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _destinatarios(remitente: str) -> tuple[list[str], list[str], str]:
|
||
|
|
"""(para, copia oculta, prefijo del asunto) según el modo."""
|
||
|
|
if modo() == "real":
|
||
|
|
return [remitente], equipo(), ""
|
||
|
|
return equipo(), [], f"[PRUEBA → {remitente}] "
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Contexto de las plantillas
|
||
|
|
# ============================================================================
|
||
|
|
def _fecha_hora(email: dict) -> tuple[str, str]:
|
||
|
|
try:
|
||
|
|
dt = datetime.strptime(email["fecha"], "%Y-%m-%d %H:%M:%S")
|
||
|
|
except (KeyError, ValueError, TypeError):
|
||
|
|
dt = datetime.now()
|
||
|
|
return f"{dt:%d/%m/%Y}", f"{dt:%H:%M}"
|
||
|
|
|
||
|
|
|
||
|
|
def _comun(email: dict, registro: dict) -> dict:
|
||
|
|
fecha, hora = _fecha_hora(email)
|
||
|
|
return {
|
||
|
|
"logo_url": logo_url(),
|
||
|
|
"recepcion_fecha": fecha,
|
||
|
|
"recepcion_hora": hora,
|
||
|
|
"codigo_seguimiento": registro["codigo_seguimiento"],
|
||
|
|
"contacto_correo": contacto_correo(),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _comprobante(factura: dict) -> dict:
|
||
|
|
numero = f"{factura.get('serie', '')}-{factura.get('numero', '')}"
|
||
|
|
tipo = "Factura electrónica" if factura.get("tipo") == "01" else factura.get("tipo_descripcion", "Comprobante")
|
||
|
|
return {
|
||
|
|
"comprobante_tipo_corto": "factura",
|
||
|
|
"comprobante_tipo": tipo,
|
||
|
|
"comprobante_numero": numero,
|
||
|
|
"proveedor_razon_social": factura.get("razon_social_emisor") or "—",
|
||
|
|
"proveedor_ruc": factura.get("ruc_emisor") or None,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def guias_para_plantilla(factura: dict, resultados_val: list[dict]) -> list[dict]:
|
||
|
|
"""
|
||
|
|
Filas de la tabla "Guías revisadas". La validación es por línea, así que todas las
|
||
|
|
guías de una línea llevan el mismo resultado; el importe de la línea se muestra una
|
||
|
|
vez (en su primera guía) y las demás dicen que están incluidas en la línea.
|
||
|
|
"""
|
||
|
|
filas = []
|
||
|
|
for i, linea in enumerate(factura.get("lineas", [])):
|
||
|
|
resultado = resultados_val[i] if i < len(resultados_val) else {}
|
||
|
|
guias = [it["guia"] for it in resultado.get("guias_lt", [])] or linea.get("guias_remision") or []
|
||
|
|
for j, guia in enumerate(guias):
|
||
|
|
filas.append({
|
||
|
|
"guia": f"{guia.get('serie', '')}-{guia.get('numero', '')}",
|
||
|
|
"importe_factura": obs.moneda(linea.get("total_linea")) if j == 0 else "incluida en la línea",
|
||
|
|
"conforme": bool(resultado.get("ok")),
|
||
|
|
})
|
||
|
|
return filas
|
||
|
|
|
||
|
|
|
||
|
|
def _fecha_estimada_pago(draft: dict) -> str | None:
|
||
|
|
"""Fecha de vencimiento del draft, solo si es posterior a la de contabilización (hay condición de pago)."""
|
||
|
|
try:
|
||
|
|
vencimiento = datetime.strptime(draft["DocDueDate"], "%Y-%m-%d")
|
||
|
|
contabilizacion = datetime.strptime(draft["DocDate"], "%Y-%m-%d")
|
||
|
|
except (KeyError, ValueError, TypeError):
|
||
|
|
return None
|
||
|
|
return f"{vencimiento:%d/%m/%Y}" if vencimiento > contabilizacion else None
|
||
|
|
|
||
|
|
|
||
|
|
def contexto_aceptado(email: dict, registro: dict, factura: dict, resultados_val: list[dict], draft: dict) -> dict:
|
||
|
|
comprobante = _comprobante(factura)
|
||
|
|
numero = comprobante["comprobante_numero"]
|
||
|
|
return {
|
||
|
|
**_comun(email, registro), **comprobante,
|
||
|
|
"asunto": f"Aceptado · Factura {numero} · {registro['codigo_seguimiento']}",
|
||
|
|
"preheader": f"Su factura {numero} pasó todas las validaciones. No necesita hacer nada más.",
|
||
|
|
"fecha_estimada_pago": _fecha_estimada_pago(draft),
|
||
|
|
"guias": guias_para_plantilla(factura, resultados_val),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _bloque_observaciones(observaciones: list[obs.Observacion]) -> dict:
|
||
|
|
texto = obs.texto_observaciones(len(observaciones))
|
||
|
|
return {
|
||
|
|
"observaciones_texto": texto,
|
||
|
|
"preheader": f"Tiene {texto}. Corríjalas y envíenos el comprobante nuevamente.",
|
||
|
|
"observaciones": [
|
||
|
|
{"numero": n, "titulo": o.titulo, "como_corregir": o.como_corregir,
|
||
|
|
"filas": [{"etiqueta": e, "valor": v} for e, v in o.filas]}
|
||
|
|
for n, o in enumerate(observaciones, start=1)
|
||
|
|
],
|
||
|
|
"envio_correo": envio_correo(),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def contexto_rechazo_factura(email: dict, registro: dict, factura: dict, resultados_val: list[dict],
|
||
|
|
observaciones: list[obs.Observacion]) -> dict:
|
||
|
|
comprobante = _comprobante(factura)
|
||
|
|
guias = guias_para_plantilla(factura, resultados_val)
|
||
|
|
return {
|
||
|
|
**_comun(email, registro), **comprobante, **_bloque_observaciones(observaciones),
|
||
|
|
"asunto": f"Rechazado · Factura {comprobante['comprobante_numero']} · {registro['codigo_seguimiento']}",
|
||
|
|
"guias": guias, "tiene_guias": bool(guias),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def contexto_rechazo_correo(email: dict, registro: dict, observaciones: list[obs.Observacion]) -> dict:
|
||
|
|
"""Rechazo de un correo sin comprobante identificable: no hay proveedor, número ni guías que mostrar."""
|
||
|
|
return {
|
||
|
|
**_comun(email, registro), **_bloque_observaciones(observaciones),
|
||
|
|
"asunto": f"Rechazado · Comprobante no recibido · {registro['codigo_seguimiento']}",
|
||
|
|
"comprobante_tipo_corto": "comprobante", "comprobante_tipo": "", "comprobante_numero": None,
|
||
|
|
"proveedor_razon_social": email.get("sender") or "—", "proveedor_ruc": None,
|
||
|
|
"guias": [], "tiene_guias": False,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
# ============================================================================
|
||
|
|
# Envío
|
||
|
|
# ============================================================================
|
||
|
|
def _anotar(codigo: str, estado: str, error: str | None = None, destinatario: str | None = None) -> None:
|
||
|
|
try:
|
||
|
|
seguimiento.registrar_notificacion(codigo, estado, error=error, destinatario=destinatario)
|
||
|
|
except Exception as ex:
|
||
|
|
log(f"⚠️ No se pudo registrar la notificación en seguimiento: {ex}", severity=logging.WARNING)
|
||
|
|
|
||
|
|
|
||
|
|
def responder(token: str, buzon: str, email: dict, plantilla: str, contexto: dict, registro: dict,
|
||
|
|
sin_comprobante: bool = False) -> str:
|
||
|
|
"""
|
||
|
|
Responde al remitente con la plantilla ``plantilla`` ("aceptado" o "rechazado").
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
``"enviada"``, ``"fallida"`` u ``"omitida"``; también queda anotado en ``comprobantes``.
|
||
|
|
"""
|
||
|
|
codigo = registro["codigo_seguimiento"]
|
||
|
|
remitente = email.get("sender") or ""
|
||
|
|
|
||
|
|
motivo = motivo_omision(email, buzon, sin_comprobante)
|
||
|
|
if motivo is None and sin_comprobante and seguimiento.contar_respuestas_correo(remitente) >= MAX_RESPUESTAS_CORREO:
|
||
|
|
motivo = f"ya se respondió {MAX_RESPUESTAS_CORREO} veces a este remitente en 24 h"
|
||
|
|
if motivo:
|
||
|
|
log(f"⏭️ No se responde al proveedor: {motivo}", severity=logging.WARNING)
|
||
|
|
_anotar(codigo, "omitida", motivo)
|
||
|
|
return "omitida"
|
||
|
|
|
||
|
|
para, copia_oculta, prefijo = _destinatarios(remitente)
|
||
|
|
try:
|
||
|
|
html = plantillas.renderizar_plantilla(plantilla, contexto)
|
||
|
|
except plantillas.PlantillaError as ex:
|
||
|
|
log(f"❌ No se pudo armar la respuesta ({codigo}): {ex}", severity=logging.ERROR)
|
||
|
|
_anotar(codigo, "fallida", f"Plantilla: {ex}")
|
||
|
|
return "fallida"
|
||
|
|
|
||
|
|
enviado = outlook_utilities.send_email(
|
||
|
|
token, buzon, para, prefijo + contexto["asunto"], html, None, bcc_recipients=copia_oculta or None,
|
||
|
|
)
|
||
|
|
estado = "enviada" if enviado else "fallida"
|
||
|
|
log(f"{'✉️ ' if enviado else '❌'} Respuesta {plantilla} {estado} → {', '.join(para)} (modo {modo()})",
|
||
|
|
severity=logging.INFO if enviado else logging.ERROR)
|
||
|
|
_anotar(codigo, estado, None if enviado else "Microsoft Graph rechazó el envío (ver log)", ", ".join(para))
|
||
|
|
return estado
|
||
|
|
|
||
|
|
|
||
|
|
def omitir_respuesta(registro: dict, motivo: str) -> None:
|
||
|
|
"""Deja anotado que a este comprobante no se le respondió, y por qué."""
|
||
|
|
_anotar(registro["codigo_seguimiento"], "omitida", motivo)
|
||
|
|
|
||
|
|
|
||
|
|
def encabezado_tecnico(email: dict, registro: dict | None) -> str:
|
||
|
|
"""Datos de contexto que se anteponen al correo técnico que recibe el equipo."""
|
||
|
|
codigo = registro["codigo_seguimiento"] if registro else "(sin código: falló el seguimiento)"
|
||
|
|
return (
|
||
|
|
f"<p><b>Remitente:</b> {email.get('sender', '')}<br>"
|
||
|
|
f"<b>Asunto:</b> {email.get('subject', '')}<br>"
|
||
|
|
f"<b>Código de seguimiento:</b> {codigo}<br>"
|
||
|
|
f"<b>Modo de notificación:</b> {modo()}</p>"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def enviar_a_equipo(token: str, buzon: str, asunto: str, html: str, adjuntos: list[str] | None = None) -> bool:
|
||
|
|
"""Envía el correo técnico (tablas de SAP, JSON adjuntos…) solo al equipo interno."""
|
||
|
|
destinos = equipo()
|
||
|
|
if not destinos:
|
||
|
|
return False
|
||
|
|
return outlook_utilities.send_email(token, buzon, destinos, asunto, html, adjuntos)
|