"""Alert delivery service — sends leak notifications to configured channels.
Public API:
send_new_leak_alerts(company_id, leaks, scan_result)
Currently supports Slack. Extensible for email, SMS, etc.
"""
from __future__ import annotations
from typing import Any, Dict, List
from .slack import SlackAlertService
__all__ = ["SlackAlertService"]
def send_new_leak_alerts(
company_id: str,
leaks: List[Dict[str, Any]],
scan_result: Dict[str, Any],
) -> Dict[str, Any]:
"""Send alerts for newly detected leaks to configured channels.
Returns: { 'slack_sent': N, 'slack_failed': str|null, 'total_sent': N }
"""
from app.models import db, Company, Connector
company = db.session.get(Company, company_id)
if company is None:
return {"error": "Company not found"}
# Only alert on new leaks
new_leaks = [l for l in leaks if l.get("status") == "new"]
if not new_leaks:
return {"slack_sent": 0, "slack_failed": None, "total_sent": 0}
result = {"slack_sent": 0, "slack_failed": None, "total_sent": 0}
# Check if Slack connector is available
slack_conn = Connector.query.filter_by(
company_id=company_id,
service="slack",
status="connected",
).first()
if slack_conn:
try:
alert_service = SlackAlertService(company_id, slack_conn)
sent = alert_service.send_digest(new_leaks, scan_result)
result["slack_sent"] = sent
result["total_sent"] = sent
except Exception as exc:
result["slack_failed"] = str(exc)
return result