import logging
import hashlib
import hmac
import httpx
import ipaddress
import socket
from datetime import datetime, timezone
import redis
from urllib.parse import urlparse
from html import escape
from sqlalchemy import text
from app.core.db import db_session
from app.core.config import get_settings
from app.core.timezone import santiago_text
from app.core.security import decrypt_text, decrypt_integration_secret, keyed_hash
from app.services.mailer import send_email
from app.services.pdf_branding import build_payment_receipt
from .celery_app import celery_app
log = logging.getLogger(__name__)
@celery_app.task(name="app.tasks.jobs.heartbeat_operacion")
def heartbeat_operacion():
"""Prueba ejecutable del ciclo scheduler -> broker -> worker."""
now = datetime.now(timezone.utc).isoformat()
client = redis.Redis.from_url(get_settings().redis_url, socket_connect_timeout=2, socket_timeout=2)
client.set("documenta:h12:worker-scheduler", now, ex=240)
return {"ok": True, "at": now}
def _public_webhook_target(url: str) -> bool:
parsed = urlparse(url)
if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password:
return False
try:
addresses = {item[4][0] for item in socket.getaddrinfo(parsed.hostname, parsed.port or 443)}
return bool(addresses) and all(ipaddress.ip_address(address).is_global for address in addresses)
except (OSError, ValueError):
return False
@celery_app.task(name="app.tasks.jobs.procesar_webhooks_pro")
def procesar_webhooks_pro():
"""Entrega webhooks H9 firmados; reintentos acotados y sin bloquear operaciones."""
with db_session() as db:
rows = db.execute(text("""
SELECT e.id,e.evento,e.payload_json,e.intentos,w.url,w.secreto_cifrado
FROM organizacion_webhook_entregas e
JOIN organizacion_webhooks w ON w.id=e.webhook_id AND w.activo=TRUE
WHERE e.estado IN ('PENDIENTE','ERROR') AND e.disponible_en<=UTC_TIMESTAMP(6)
AND e.intentos<8 ORDER BY e.id LIMIT 25 FOR UPDATE SKIP LOCKED
""")).mappings().all()
delivered = 0
for row in rows:
body = row["payload_json"].encode()
secret = decrypt_integration_secret(row["secreto_cifrado"])
signature = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
try:
if not _public_webhook_target(row["url"]):
raise ValueError("WEBHOOK_TARGET_NOT_PUBLIC")
response = httpx.post(row["url"], content=body, timeout=15.0,
headers={"Content-Type": "application/json",
"User-Agent": "Documenta-Pro-Webhooks/1.0",
"X-Documenta-Event": row["evento"],
"X-Documenta-Signature": f"sha256={signature}"})
if 200 <= response.status_code < 300:
db.execute(text("""UPDATE organizacion_webhook_entregas
SET estado='ENVIADO',intentos=intentos+1,http_status=:status,
ultimo_error=NULL,entregado_en=UTC_TIMESTAMP(6) WHERE id=:id"""),
{"status": response.status_code, "id": row["id"]})
delivered += 1
continue
error = f"HTTP_{response.status_code}"
status = response.status_code
except (httpx.RequestError, ValueError) as exc:
error = type(exc).__name__
status = None
attempts = int(row["intentos"]) + 1
db.execute(text("""UPDATE organizacion_webhook_entregas
SET estado=IF(:attempts>=8,'AGOTADO','ERROR'),intentos=:attempts,
http_status=:status,ultimo_error=:error,
disponible_en=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL LEAST(3600,30*POW(2,LEAST(:attempts,7))) SECOND)
WHERE id=:id"""), {"attempts": attempts, "status": status,
"error": error[:500], "id": row["id"]})
return {"encontrados": len(rows), "entregados": delivered}
def _money(value) -> str:
return f"${float(value):,.0f}".replace(",", ".")
def _email_html(*, title: str, lead: str, rows: list[tuple[str, object]], action_label: str,
action_url: str, secondary_label: str, secondary_url: str, debug: bool) -> str:
details = "".join(f'| {escape(label)} | '
f'{escape(str(value or "—"))} |
'
for label, value in rows)
debug_text = "Activo: copia oculta al administrador" if debug else "Inactivo"
return f'''
'''
@celery_app.task(name="app.tasks.jobs.reintentar_gesnot")
def reintentar_gesnot():
"""Crea OTs nuevas o recupera éxitos locales; nunca repite un POST incierto."""
from app.api.routes import _create_gesnot_after_payment
with db_session() as db:
pending = db.execute(text("""
SELECT v.id venta_id, MAX(p.id) pago_id
FROM ventas v JOIN estados_venta ev ON ev.id=v.estado_venta_id
JOIN pagos p ON p.venta_id=v.id JOIN estados_pago ep ON ep.id=p.estado_pago_id
LEFT JOIN ordenes_gesnot og ON og.venta_id=v.id
WHERE ep.codigo='AUTORIZADO'
AND EXISTS (
SELECT 1 FROM oferta_conectores oc
JOIN proveedores_integracion pi ON pi.id=oc.proveedor_integracion_id
WHERE oc.oferta_id=v.directorio_id AND oc.evento='POST_PAGO'
AND oc.activo=TRUE AND pi.codigo='GESNOT'
)
AND ((ev.codigo='PENDIENTE_GESNOT' AND og.id IS NULL)
OR (ev.codigo='OT_CREADA' AND og.numero_ot LIKE 'OT-SBX-%'))
AND NOT EXISTS (
SELECT 1 FROM intentos_integracion reciente
WHERE reciente.venta_id=v.id AND reciente.operacion='CREAR_OT'
AND reciente.iniciado_en>DATE_SUB(UTC_TIMESTAMP(6),INTERVAL 2 MINUTE)
)
AND NOT EXISTS (
SELECT 1 FROM intentos_integracion enviado
JOIN estados_integracion eie ON eie.id=enviado.estado_integracion_id
WHERE enviado.venta_id=v.id AND enviado.operacion='CREAR_OT'
AND eie.codigo IN ('REINTENTABLE','FALLIDA')
AND NOT EXISTS (
SELECT 1 FROM intentos_integracion exitoso
JOIN estados_integracion eix ON eix.id=exitoso.estado_integracion_id
WHERE exitoso.venta_id=v.id AND exitoso.operacion='CREAR_OT'
AND eix.codigo='EXITOSA'
)
)
GROUP BY v.id ORDER BY v.id DESC LIMIT 5
""")).mappings().all()
completed = 0
for item in pending:
lock_name = f"documenta:gesnot:{item['venta_id']}"
with db_session() as lock_db:
acquired = lock_db.execute(text("SELECT GET_LOCK(:name,0)"), {"name": lock_name}).scalar()
if not acquired:
continue
try:
_create_gesnot_after_payment(item["venta_id"], item["pago_id"])
completed += 1
except Exception:
log.exception("Reintento GesNot fallido venta_id=%s", item["venta_id"])
finally:
lock_db.execute(text("SELECT RELEASE_LOCK(:name)"), {"name": lock_name})
return {"pendientes": len(pending), "completadas": completed}
@celery_app.task(name="app.tasks.jobs.vencer_documentos")
def vencer_documentos():
with db_session() as db:
db.execute(text("SET @cantidad=0"))
db.execute(text("CALL sp_vencer_documentos(200,@cantidad)"))
return {"vencidos": db.execute(text("SELECT @cantidad")).scalar_one()}
@celery_app.task(name="app.tasks.jobs.procesar_outbox")
def procesar_outbox():
with db_session() as db:
events = db.execute(text("""
SELECT id,agregado_id,evento_tipo FROM eventos_outbox
WHERE procesado_en IS NULL AND disponible_en<=UTC_TIMESTAMP(6)
ORDER BY id LIMIT 50 FOR UPDATE SKIP LOCKED
""")).mappings().all()
processed = 0
for event in events:
log.info("outbox_event id=%s type=%s", event["id"], event["evento_tipo"])
if event["evento_tipo"] in ("VENTA_PAGO_APROBADO", "VENTA_REALIZADO", "DOCUMENTO_REALIZADO", "DOCUMENTO_VENCIDO", "RESERVA_CONFIRMADA"):
contact = db.execute(text("""
SELECT c.email_cifrado,c.nombres_cifrado,c.apellidos_cifrado,n.id notaria_id,
n.email_notificaciones,n.email_operaciones,
n.nombre_fantasia,n.direccion,v.numero_publico,v.total,v.moneda,m.nombre materia,
p.buy_order,p.codigo_autorizacion,p.monto,p.confirmado_en,
rt.estado reserva_estado,rb.inicio_utc reserva_inicio_utc,
rs.nombre reserva_servicio,rc.codigo reserva_codigo,
ra.email_notificaciones reserva_avisos
FROM ventas v JOIN clientes c ON c.id=v.cliente_id
JOIN notarias n ON n.id=v.notaria_id JOIN materias m ON m.id=v.materia_id
LEFT JOIN pagos p ON p.id=(SELECT MAX(p2.id) FROM pagos p2 JOIN estados_pago ep2 ON ep2.id=p2.estado_pago_id WHERE p2.venta_id=v.id AND ep2.codigo='AUTORIZADO')
LEFT JOIN reservas_tramites rt ON rt.venta_id=v.id
LEFT JOIN reservas_bloques rb ON rb.id=rt.bloque_id
LEFT JOIN reservas_servicios rs ON rs.id=rb.servicio_id
LEFT JOIN reservas_citas rc ON rc.bloque_id=rb.id AND rc.estado='CONFIRMADA'
LEFT JOIN reservas_ajustes ra ON ra.notaria_id=n.id
WHERE v.id=:id
"""), {"id": event["agregado_id"]}).mappings().first()
if event["evento_tipo"] == "RESERVA_CONFIRMADA" and (
not contact or contact["reserva_estado"] != "CONFIRMADA" or not contact["reserva_codigo"]):
db.execute(text("UPDATE eventos_outbox SET procesado_en=UTC_TIMESTAMP(6) WHERE id=:id"),{"id":event["id"]})
processed += 1
continue
if contact:
labels = {
"VENTA_PAGO_APROBADO": "Pago aprobado",
"DOCUMENTO_REALIZADO": "Documento realizado",
"VENTA_REALIZADO": "Documento realizado",
"DOCUMENTO_VENCIDO": "Trámite vencido",
"RESERVA_CONFIRMADA": "Hora confirmada",
}
subject = f"Documenta: {labels[event['evento_tipo']]} - {contact['numero_publico']}"
status_url=f"{get_settings().public_base_url}/estado?codigo={contact['numero_publico']}"
panel_url=f"{get_settings().public_base_url}/panel"
client_name=f"{decrypt_text(contact['nombres_cifrado'])} {decrypt_text(contact['apellidos_cifrado'])}".strip()
config=db.execute(text("SELECT debug_copia_admin FROM configuracion_correo WHERE id=1")).mappings().first()
debug=bool(config and config["debug_copia_admin"]); bcc=[get_settings().admin_error_email] if debug else []
receipt=build_payment_receipt(contact) if event["evento_tipo"] == "VENTA_PAGO_APROBADO" else None
common=[("Folio",contact["numero_publico"]),("Servicio",contact["materia"]),
("Entidad",contact["nombre_fantasia"]),("Dirección",contact["direccion"]),
("Monto",f"{_money(contact['total'])} {contact['moneda']}"),("Fecha",santiago_text(contact["confirmado_en"]))]
if event["evento_tipo"] == "RESERVA_CONFIRMADA":
common.extend([("Hora reservada",santiago_text(contact["reserva_inicio_utc"])),
("Atención",contact["reserva_servicio"]),
("Código de reserva",contact["reserva_codigo"])])
targets=[("CLIENTE",decrypt_text(contact["email_cifrado"]),f"Hola {client_name}. {labels[event['evento_tipo']]} para tu trámite.","Ver seguimiento",status_url,"Ir a mi panel",panel_url)]
if event["evento_tipo"] in ("VENTA_PAGO_APROBADO", "VENTA_REALIZADO", "DOCUMENTO_REALIZADO", "DOCUMENTO_VENCIDO", "RESERVA_CONFIRMADA"):
recipient=(contact["reserva_avisos"] or contact["email_notificaciones"]) if event["evento_tipo"]=="RESERVA_CONFIRMADA" else contact["email_notificaciones"]
targets.append(("NOTARIA",recipient,"Se registró una novedad en una solicitud asignada a la entidad.","Abrir solicitud en el panel",panel_url,"Ver panel",panel_url))
if event["evento_tipo"] in ("VENTA_PAGO_APROBADO", "VENTA_REALIZADO", "DOCUMENTO_REALIZADO", "RESERVA_CONFIRMADA"):
members=db.execute(text("""
SELECT u.id,u.email FROM notaria_usuarios nu JOIN usuarios u ON u.id=nu.usuario_id
WHERE nu.notaria_id=:notary AND nu.activo=TRUE AND u.activo=TRUE
"""),{"notary":contact["notaria_id"]}).mappings().all()
targets.append(("OPERACIONES",contact["email_operaciones"],
"Hay una novedad en un trámite de la entidad.","Abrir panel",panel_url,"Ver panel",panel_url))
targets.extend((f"USUARIO_{m['id']}",m["email"],
"Hay una novedad en un trámite de la entidad.","Abrir panel",panel_url,"Ver panel",panel_url) for m in members)
seen_recipients=set()
failures=[]
for kind,recipient,lead,primary,primary_url,secondary,secondary_url in targets:
if not recipient or recipient.strip().lower() in seen_recipients: continue
seen_recipients.add(recipient.strip().lower())
delivery=db.execute(text("SELECT estado FROM entregas_correo_evento WHERE evento_outbox_id=:event AND destinatario_tipo=:kind"),{"event":event["id"],"kind":kind}).scalar()
if delivery == "ENVIADO": continue
db.execute(text("""INSERT INTO entregas_correo_evento(evento_outbox_id,destinatario_tipo,destinatario_hmac)
VALUES(:event,:kind,:hmac) ON DUPLICATE KEY UPDATE intentos=intentos+1"""),
{"event":event["id"],"kind":kind,"hmac":keyed_hash(recipient,get_settings().access_token_hmac_key)})
detail_rows=list(common)
if kind == "NOTARIA":
detail_rows.extend([("Cliente",client_name),("Orden de compra",contact["buy_order"] or "—"),
("Autorización",contact["codigo_autorizacion"] or "—")])
body=f"{lead}\n"+"\n".join(f"{k}: {v}" for k,v in detail_rows)+f"\nSeguimiento: {status_url}\nPanel: {panel_url}\nModo depuración: {'activo' if debug else 'inactivo'}"
html=_email_html(title=labels[event["evento_tipo"]],lead=lead,rows=detail_rows,action_label=primary,action_url=primary_url,secondary_label=secondary,secondary_url=secondary_url,debug=debug)
attachments=[(f"comprobante-{contact['numero_publico']}.pdf",receipt,"application","pdf")] if receipt else []
try:
send_email(recipient,subject,body,html_body=html,attachments=attachments,bcc=bcc)
db.execute(text("UPDATE entregas_correo_evento SET estado='ENVIADO',enviado_en=UTC_TIMESTAMP(6),ultimo_error_sanitizado=NULL WHERE evento_outbox_id=:event AND destinatario_tipo=:kind"),{"event":event["id"],"kind":kind})
except Exception as exc:
safe=type(exc).__name__; failures.append(f"{kind}:{safe}")
db.execute(text("UPDATE entregas_correo_evento SET estado='ERROR',ultimo_error_sanitizado=:error WHERE evento_outbox_id=:event AND destinatario_tipo=:kind"),{"event":event["id"],"kind":kind,"error":safe})
if failures:
db.execute(text("""UPDATE eventos_outbox SET intentos=intentos+1,ultimo_error_sanitizado=:error,
disponible_en=DATE_ADD(UTC_TIMESTAMP(6),INTERVAL LEAST(3600,30*POW(2,LEAST(intentos,7))) SECOND) WHERE id=:id"""),
{"id":event["id"],"error":";".join(failures)})
continue
db.execute(text("UPDATE eventos_outbox SET procesado_en=UTC_TIMESTAMP(6),ultimo_error_sanitizado=NULL WHERE id=:id"), {"id": event["id"]})
processed += 1
return {"encontrados": len(events), "procesados": processed}