"""Slack alert delivery for leak notifications.
Builds Slack blocks and sends via the existing Slack connector's bot_token.
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List, Optional
from app.models import Connector
logger = logging.getLogger(__name__)
# Severity emoji mapping
SEVERITY_EMOJI = {
"critical": "🔴",
"high": "🟠",
"medium": "🟡",
"low": "🟢",
}
# Max blocks before truncating
MAX_LEAK_ITEMS = 10
class SlackAlertService:
"""Sends leak digest messages to a company's Slack workspace."""
def __init__(self, company_id: str, connector: Connector):
self.company_id = company_id
self.connector = connector
self.bot_token = (
(connector.config or {}).get("bot_token")
or (connector.config or {}).get("access_token")
or ""
)
# -- Channel target --------------------------------------------------------
def _resolve_channel(self) -> str:
"""Get the target channel from alert settings.
Priority:
1. Company settings leak_alerts.slack_channel
2. ConnectedAccounts settings
3. Default to first public channel
"""
from app.models import Company
company = __import__("app.models", fromlist=["db"]).db.session.get(
__import__("app.models", fromlist=["Company"]).Company,
self.company_id,
)
if company is None:
return ""
settings = company.settings_json or {}
alert_settings = settings.get("leak_alerts") or {}
channel = alert_settings.get("slack_channel")
if channel:
return channel
# Fallback: #general or first public channel
return "general"
# -- Message builder -------------------------------------------------------
def _build_digest_blocks(
self,
new_leaks: List[Dict[str, Any]],
scan_result: Dict[str, Any],
) -> List[Dict[str, Any]]:
"""Build Slack blocks for a leak digest message."""
from slack_sdk.blocks import (
HeaderBlock,
SectionBlock,
DividerBlock,
ActionsBlock,
ButtonElement,
)
# Can't import slack_sdk blocks at class level — do it here
# Actually let's just build raw blocks since we may not have slack_sdk installed
blocks: List[Dict[str, Any]] = []
# Header
critical_count = sum(
1 for l in new_leaks if l.get("severity") == "critical"
)
high_count = sum(
1 for l in new_leaks if l.get("severity") == "high"
)
total = len(new_leaks)
if critical_count > 0:
header_text = (
f"🚨 {critical_count} Critical Leak(s) Detected"
)
elif high_count > 0:
header_text = (
f"⚠️ {total} Revenue Leak(s) Detected"
)
else:
header_text = (
f"📊 {total} Revenue Leak(s) Detected"
)
blocks.append({
"type": "header",
"text": {
"type": "plain_text",
"text": header_text,
"emoji": True,
},
})
# Summary line
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f"Auto-scan found *{total}* new leak(s). "
f"*{critical_count}* critical, *"
f"{high_count}* high. "
f"Total estimated loss: "
f"${self._total_estimated_loss(new_leaks):,.0f}"
),
},
})
blocks.append({"type": "divider"})
# Individual leaks (up to MAX_LEAK_ITEMS)
sorted_leaks = sorted(
new_leaks,
key=lambda l: {
"critical": 0,
"high": 1,
"medium": 2,
"low": 3,
}.get(l.get("severity", "medium"), 4),
)
for leak in sorted_leaks[:MAX_LEAK_ITEMS]:
emoji = SEVERITY_EMOJI.get(
leak.get("severity", "medium"), "⚪"
)
loss = leak.get("estimated_loss")
loss_str = (
f"${loss:,.0f}" if loss is not None else "N/A"
)
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f"{emoji} *{leak.get('severity', 'medium').upper()}* "
f"— *{leak.get('source', 'Unknown')}*\n"
f"{leak.get('description', '')}\n"
f"Estimated loss: {loss_str}"
),
},
})
if total > MAX_LEAK_ITEMS:
blocks.append({
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f"*...and {total - MAX_LEAK_ITEMS} more*"
),
},
})
blocks.append({"type": "divider"})
# View in app button
blocks.append({
"type": "actions",
"elements": [
{
"type": "button",
"text": {
"type": "plain_text",
"text": "View Revenue Leaks Dashboard",
"emoji": True,
},
"url": self._app_url("/leaks"),
"action_id": "view_leaks",
}
],
})
return blocks
def _total_estimated_loss(
self, leaks: List[Dict[str, Any]]
) -> float:
return sum(
l.get("estimated_loss") or 0
for l in leaks
)
def _app_url(self, path: str) -> str:
"""Build URL to the app."""
from flask import current_app
base_url = current_app.config.get(
"APP_URL", "https://commandsovereignty.com"
)
return f"{base_url}{path}"
# -- Send ------------------------------------------------------------------
def send_digest(
self,
new_leaks: List[Dict[str, Any]],
scan_result: Dict[str, Any],
) -> int:
"""Send digest message. Returns number of messages sent."""
import requests
channel = self._resolve_channel()
if not channel:
logger.warning(
"No Slack channel configured for alerts on company %s",
self.company_id,
)
return 0
blocks = self._build_digest_blocks(new_leaks, scan_result)
# Send via Slack Web API
response = requests.post(
"https://slack.com/api/chat.postMessage",
headers={
"Authorization": f"Bearer {self.bot_token}",
"Content-Type": "application/json",
},
json={
"channel": f"#{channel}" if not channel.startswith("#") else channel,
"blocks": blocks,
},
timeout=10,
)
data = response.json()
if not data.get("ok"):
error = data.get("error", "unknown")
logger.error(
"Slack alert failed for company %s: %s",
self.company_id,
error,
)
raise ValueError(f"Slack API error: {error}")
logger.info(
"Sent Slack alert to %s for company %s — %d leaks",
channel,
self.company_id,
len(new_leaks),
)
return 1