Coverage for src / documint_mcp / connectors / slack.py: 0%
97 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-24 23:27 -0400
1"""Slack notification connector using ``slack-sdk``.
3Gracefully degrades to a no-op when the SDK is not installed or when
4the required environment variables (``SLACK_BOT_TOKEN``, ``SLACK_CHANNEL``)
5are not set.
6"""
8from __future__ import annotations
10import logging
11import os
12from typing import Any
14from documint_mcp.models import DriftFinding, DocPatch, FindingSeverity, VerificationRun
16from .base import ConnectorBase
18logger = logging.getLogger(__name__)
20# ---------------------------------------------------------------------------
21# Optional import: slack-sdk is not a hard dependency
22# ---------------------------------------------------------------------------
23try:
24 from slack_sdk.web.async_client import AsyncWebClient
26 _HAS_SLACK_SDK = True
27except ImportError: # pragma: no cover
28 _HAS_SLACK_SDK = False
29 AsyncWebClient = None # type: ignore[assignment, misc]
31# ---------------------------------------------------------------------------
32# Block Kit helpers
33# ---------------------------------------------------------------------------
34_SEVERITY_EMOJI: dict[str, str] = {
35 "high": ":red_circle:",
36 "medium": ":large_orange_circle:",
37 "low": ":large_yellow_circle:",
38}
41def _severity_badge(severity: str) -> str:
42 emoji = _SEVERITY_EMOJI.get(severity, ":white_circle:")
43 return f"{emoji} {severity.upper()}"
46def _truncate(text: str, limit: int = 2900) -> str:
47 """Slack blocks have a 3000-char text limit; leave margin."""
48 if len(text) <= limit:
49 return text
50 return text[:limit] + "\n... (truncated)"
53def _drift_alert_blocks(
54 project_id: str,
55 findings: list[DriftFinding],
56) -> list[dict[str, Any]]:
57 """Build Block Kit blocks for a drift alert message."""
58 high = sum(1 for f in findings if f.severity == FindingSeverity.HIGH)
59 medium = sum(1 for f in findings if f.severity == FindingSeverity.MEDIUM)
60 low = sum(1 for f in findings if f.severity == FindingSeverity.LOW)
62 blocks: list[dict[str, Any]] = [
63 {
64 "type": "header",
65 "text": {
66 "type": "plain_text",
67 "text": f"Doc Drift Detected -- {project_id}",
68 "emoji": True,
69 },
70 },
71 {
72 "type": "section",
73 "text": {
74 "type": "mrkdwn",
75 "text": (
76 f"*{len(findings)}* finding{'s' if len(findings) != 1 else ''} detected\n"
77 f"{_SEVERITY_EMOJI['high']} {high} high "
78 f"{_SEVERITY_EMOJI['medium']} {medium} medium "
79 f"{_SEVERITY_EMOJI['low']} {low} low"
80 ),
81 },
82 },
83 {"type": "divider"},
84 ]
86 for finding in findings[:10]:
87 badge = _severity_badge(finding.severity.value)
88 stale_docs = ", ".join(f"`{p}`" for p in finding.doc_paths[:3])
89 if len(finding.doc_paths) > 3:
90 stale_docs += f" +{len(finding.doc_paths) - 3} more"
92 section_text = (
93 f"{badge} *{finding.artifact_id}*\n"
94 f"{finding.summary}\n"
95 f"Stale docs: {stale_docs or '_none_'}"
96 )
98 blocks.append(
99 {
100 "type": "section",
101 "text": {"type": "mrkdwn", "text": _truncate(section_text)},
102 "accessory": {
103 "type": "overflow",
104 "action_id": f"drift_actions_{finding.id}",
105 "options": [
106 {
107 "text": {"type": "plain_text", "text": "Approve Patch"},
108 "value": f"approve_patch|{finding.id}",
109 },
110 {
111 "text": {"type": "plain_text", "text": "Show Diff"},
112 "value": f"show_diff|{finding.id}",
113 },
114 {
115 "text": {"type": "plain_text", "text": "Dismiss"},
116 "value": f"dismiss|{finding.id}",
117 },
118 ],
119 },
120 }
121 )
123 if len(findings) > 10:
124 blocks.append(
125 {
126 "type": "context",
127 "elements": [
128 {
129 "type": "mrkdwn",
130 "text": f"_Showing 10 of {len(findings)} findings. View all in the dashboard._",
131 }
132 ],
133 }
134 )
136 return blocks
139def _patch_preview_blocks(
140 project_id: str,
141 patch: DocPatch,
142) -> list[dict[str, Any]]:
143 """Build Block Kit blocks for a patch preview thread reply."""
144 diff_preview = _truncate(patch.preview_markdown, limit=2800)
146 blocks: list[dict[str, Any]] = [
147 {
148 "type": "section",
149 "text": {
150 "type": "mrkdwn",
151 "text": (
152 f"*Patch Preview* -- `{patch.target_path}`\n"
153 f"_{patch.summary}_"
154 ),
155 },
156 },
157 {
158 "type": "section",
159 "text": {"type": "mrkdwn", "text": f"```\n{diff_preview}\n```"},
160 },
161 ]
163 if patch.citations:
164 cite_lines = "\n".join(
165 f"- `{c.path}` @ `{c.ref}`: {c.note}" for c in patch.citations[:5]
166 )
167 blocks.append(
168 {
169 "type": "context",
170 "elements": [
171 {"type": "mrkdwn", "text": f"*Citations:*\n{cite_lines}"},
172 ],
173 }
174 )
176 blocks.append(
177 {
178 "type": "actions",
179 "elements": [
180 {
181 "type": "button",
182 "text": {"type": "plain_text", "text": "Approve"},
183 "style": "primary",
184 "action_id": f"approve_patch_{patch.id}",
185 "value": f"{project_id}|{patch.id}",
186 },
187 {
188 "type": "button",
189 "text": {"type": "plain_text", "text": "Reject"},
190 "style": "danger",
191 "action_id": f"reject_patch_{patch.id}",
192 "value": f"{project_id}|{patch.id}",
193 },
194 ],
195 }
196 )
198 return blocks
201def _digest_blocks(
202 project_id: str,
203 report: VerificationRun,
204) -> list[dict[str, Any]]:
205 """Build Block Kit blocks for a periodic health digest."""
206 high = sum(1 for f in report.findings if f.severity == FindingSeverity.HIGH)
207 medium = sum(1 for f in report.findings if f.severity == FindingSeverity.MEDIUM)
208 low = sum(1 for f in report.findings if f.severity == FindingSeverity.LOW)
210 status_icon = ":white_check_mark:" if report.findings_count == 0 else ":warning:"
212 blocks: list[dict[str, Any]] = [
213 {
214 "type": "header",
215 "text": {
216 "type": "plain_text",
217 "text": f"Documentation Health Digest -- {project_id}",
218 "emoji": True,
219 },
220 },
221 {
222 "type": "section",
223 "fields": [
224 {"type": "mrkdwn", "text": f"*Status:*\n{status_icon} {report.status.value}"},
225 {"type": "mrkdwn", "text": f"*Total findings:*\n{report.findings_count}"},
226 {"type": "mrkdwn", "text": f"*High:* {high}"},
227 {"type": "mrkdwn", "text": f"*Medium:* {medium}"},
228 {"type": "mrkdwn", "text": f"*Low:* {low}"},
229 {
230 "type": "mrkdwn",
231 "text": f"*Signal:*\n{report.signal.type.value} @ `{report.signal.ref[:12]}`",
232 },
233 ],
234 },
235 ]
237 if report.completed_at:
238 blocks.append(
239 {
240 "type": "context",
241 "elements": [
242 {
243 "type": "mrkdwn",
244 "text": f"Completed {report.completed_at.isoformat()}",
245 }
246 ],
247 }
248 )
250 return blocks
253# ---------------------------------------------------------------------------
254# Connector implementation
255# ---------------------------------------------------------------------------
258class SlackConnector(ConnectorBase):
259 """Posts drift alerts, patch previews, and digests to a Slack channel.
261 Requires:
262 - ``slack-sdk`` installed (``pip install slack-sdk``)
263 - ``SLACK_BOT_TOKEN`` env var with a valid Bot User OAuth Token
264 - ``SLACK_CHANNEL`` env var with the target channel ID or name
266 If any of the above are missing the connector silently degrades to
267 a no-op, logging a warning on first call.
268 """
270 def __init__(self) -> None:
271 self._token: str | None = os.getenv("SLACK_BOT_TOKEN")
272 self._channel: str | None = os.getenv("SLACK_CHANNEL")
273 self._client: Any | None = None
274 self._warned = False
276 if _HAS_SLACK_SDK and self._token:
277 self._client = AsyncWebClient(token=self._token)
279 @property
280 def is_configured(self) -> bool:
281 return self._client is not None and bool(self._channel)
283 def _warn_once(self, reason: str) -> None:
284 if not self._warned:
285 logger.warning("SlackConnector disabled: %s", reason)
286 self._warned = True
288 # -- drift alert ---------------------------------------------------------
290 async def send_drift_alert(
291 self,
292 project_id: str,
293 findings: list[DriftFinding],
294 ) -> None:
295 if not self.is_configured:
296 self._warn_once(
297 "slack-sdk not installed"
298 if not _HAS_SLACK_SDK
299 else "SLACK_BOT_TOKEN or SLACK_CHANNEL not set"
300 )
301 return
303 if not findings:
304 return
306 blocks = _drift_alert_blocks(project_id, findings)
307 fallback = f"Doc drift detected: {len(findings)} finding(s) in {project_id}"
309 try:
310 await self._client.chat_postMessage( # type: ignore[union-attr]
311 channel=self._channel,
312 text=fallback,
313 blocks=blocks,
314 )
315 except Exception:
316 logger.exception("Failed to send Slack drift alert for %s", project_id)
318 # -- patch preview -------------------------------------------------------
320 async def send_patch_preview(
321 self,
322 project_id: str,
323 patch: DocPatch,
324 ) -> None:
325 if not self.is_configured:
326 self._warn_once(
327 "slack-sdk not installed"
328 if not _HAS_SLACK_SDK
329 else "SLACK_BOT_TOKEN or SLACK_CHANNEL not set"
330 )
331 return
333 blocks = _patch_preview_blocks(project_id, patch)
334 fallback = f"Patch preview for {patch.target_path} in {project_id}"
336 try:
337 await self._client.chat_postMessage( # type: ignore[union-attr]
338 channel=self._channel,
339 text=fallback,
340 blocks=blocks,
341 )
342 except Exception:
343 logger.exception("Failed to send Slack patch preview for %s", project_id)
345 # -- digest --------------------------------------------------------------
347 async def send_digest(
348 self,
349 project_id: str,
350 report: VerificationRun,
351 ) -> None:
352 if not self.is_configured:
353 self._warn_once(
354 "slack-sdk not installed"
355 if not _HAS_SLACK_SDK
356 else "SLACK_BOT_TOKEN or SLACK_CHANNEL not set"
357 )
358 return
360 blocks = _digest_blocks(project_id, report)
361 fallback = f"Doc health digest for {project_id}: {report.findings_count} findings"
363 try:
364 await self._client.chat_postMessage( # type: ignore[union-attr]
365 channel=self._channel,
366 text=fallback,
367 blocks=blocks,
368 )
369 except Exception:
370 logger.exception("Failed to send Slack digest for %s", project_id)