import asyncio import email.message import json import logging import smtplib import ssl from typing import Any import httpx from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from tenacity import AsyncRetrying, RetryError, stop_after_attempt, wait_exponential from app.config import settings logger = logging.getLogger(__name__) EMAIL_SUBJECT = "[Segunor Intel Grid] Novo processo: {process_number}" EMAIL_BODY = """Novo processo detectado para {company}. Processo: {process_number} Tribunal: {court} Data: {date} Espécie: {type} Fonte: {source} NIF: {nif} Detetado em: {created_at} """ async def _load_configs( session: AsyncSession, company_id: str ) -> list[dict[str, Any]]: """Load enabled alert configs that apply to this company (specific or global).""" rows = ( await session.execute( text( """ SELECT id::text, company_id::text, type, target, enabled FROM alerts_config WHERE enabled = TRUE AND (company_id IS NULL OR company_id = :cid) """ ), {"cid": company_id}, ) ).all() return [ {"id": r[0], "company_id": r[1], "type": r[2], "target": r[3], "enabled": r[4]} for r in rows ] async def _log_dispatch( session: AsyncSession, *, process_id: str, alert_config_id: str, status: str, response: str | None, ) -> None: await session.execute( text( """ INSERT INTO alerts_log (process_id, alert_config_id, status, response) VALUES (:pid, :cid, :st, :resp) """ ), {"pid": process_id, "cid": alert_config_id, "st": status, "resp": (response or "")[:2000] or None}, ) def _build_payload(process_row: dict[str, Any]) -> dict[str, Any]: return { "company": process_row.get("company_name") or "", "nif": process_row.get("company_nif"), "process_number": process_row.get("process_number") or "", "court": process_row.get("tribunal") or "", "date": (process_row.get("date_filed").isoformat() if process_row.get("date_filed") else ""), "type": process_row.get("species"), "source": process_row.get("source") or "", "created_at": (process_row.get("first_seen_at").isoformat() if process_row.get("first_seen_at") else ""), } def _send_smtp(to_addr: str, subject: str, body: str) -> str: """Blocking SMTP send. Returns a short status string on success or raises. Suporta os dois modos: STARTTLS (tipicamente 587) e TLS implícito (465), que é o do mail.segunor.pt. Com SMTP_SSL_IMPLICIT ligado a ligação nasce já cifrada e não se chama starttls(). """ if not settings.SMTP_HOST: raise RuntimeError("SMTP not configured") msg = email.message.EmailMessage() msg["From"] = settings.SMTP_FROM or settings.SMTP_USER or "noreply@localhost" msg["To"] = to_addr msg["Subject"] = subject msg.set_content(body) ctx = ssl.create_default_context() if settings.SMTP_SSL_IMPLICIT: with smtplib.SMTP_SSL( settings.SMTP_HOST, settings.SMTP_PORT, timeout=20, context=ctx ) as s: if settings.SMTP_USER: s.login(settings.SMTP_USER, settings.SMTP_PASSWORD) s.send_message(msg) return f"smtp ok (ssl) to {to_addr}" with smtplib.SMTP(settings.SMTP_HOST, settings.SMTP_PORT, timeout=20) as s: if settings.SMTP_STARTTLS: s.starttls(context=ctx) if settings.SMTP_USER: s.login(settings.SMTP_USER, settings.SMTP_PASSWORD) s.send_message(msg) return f"smtp ok to {to_addr}" async def send_email_raw(target: str, subject: str, body: str) -> tuple[bool, str]: """Envio com assunto e corpo já montados. O `send_email_alert` formata a partir do template de processos; os avisos de insolvência têm corpo próprio (prazo, administrador, base do cálculo) e precisam de passar ao lado desse template. """ if not settings.SMTP_HOST: return False, "SMTP_HOST not configured; email skipped" try: resp = await asyncio.to_thread(_send_smtp, target, subject, body) return True, resp except Exception as e: logger.warning("email para %s falhou: %s", target, e) return False, f"{type(e).__name__}: {e}" async def send_email_alert(target: str, payload: dict[str, Any]) -> tuple[bool, str]: if not settings.SMTP_HOST: return False, "SMTP_HOST not configured; email skipped" subject = EMAIL_SUBJECT.format(**payload) body = EMAIL_BODY.format(**{**payload, "nif": payload.get("nif") or "—"}) try: resp = await asyncio.to_thread(_send_smtp, target, subject, body) return True, resp except Exception as e: logger.warning("email alert to %s failed: %s", target, e) return False, f"{type(e).__name__}: {e}" async def send_webhook_alert(target: str, payload: dict[str, Any]) -> tuple[bool, str]: body = json.dumps(payload, default=str) try: async for attempt in AsyncRetrying( stop=stop_after_attempt(settings.WEBHOOK_RETRIES), wait=wait_exponential(multiplier=1, min=1, max=8), reraise=True, ): with attempt: async with httpx.AsyncClient(timeout=settings.WEBHOOK_TIMEOUT_S) as c: r = await c.post( target, content=body, headers={"Content-Type": "application/json"}, ) r.raise_for_status() return True, f"{r.status_code} {r.text[:200]}" except RetryError as e: # pragma: no cover return False, f"retry exhausted: {e}" except Exception as e: return False, f"{type(e).__name__}: {e}" return False, "unknown" # pragma: no cover async def trigger_alerts( session: AsyncSession, process_id: str, process_row: dict[str, Any], ) -> None: """Dispatch alerts for a newly discovered process. Failures are logged but never raise into the caller's transaction.""" company_id = str(process_row.get("company_id")) if not company_id: return try: configs = await _load_configs(session, company_id) except Exception as e: logger.warning("load alerts_config failed: %s", e) return if not configs: return payload = _build_payload(process_row) logger.info( "trigger_alerts: process=%s configs=%d", process_row.get("process_number"), len(configs) ) for cfg in configs: if cfg["type"] == "email": ok, resp = await send_email_alert(cfg["target"], payload) elif cfg["type"] == "webhook": ok, resp = await send_webhook_alert(cfg["target"], payload) else: ok, resp = False, f"unknown type {cfg['type']}" try: await _log_dispatch( session, process_id=process_id, alert_config_id=cfg["id"], status="sent" if ok else "failed", response=resp, ) except Exception as e: logger.warning("alerts_log insert failed: %s", e)