tracer-cloud--opensre/integrations/github/issue_comments.py

"""Notify Slack when GitHub issue comments are created."""

from __future__ import annotations

import json
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from urllib import error, parse, request

MAX_COMMENT_PREVIEW_CHARS = 500


@dataclass(frozen=True)
class IssueCommentNotification:
    """Normalized issue comment notification payload."""

    repository: str
    issue_number: int
    issue_title: str
    issue_url: str
    issue_author: str
    comment_author: str
    comment_url: str
    comment_body: str


def _string(value: Any) -> str:
    return value.strip() if isinstance(value, str) else ""


def _truncate_comment(text: str, *, limit: int = MAX_COMMENT_PREVIEW_CHARS) -> str:
    normalized = "\n".join(line.rstrip() for line in text.strip().splitlines()).strip()
    if len(normalized) <= limit:
        return normalized
    return normalized[: limit - 3].rstrip() + "..."


def notification_from_issue_comment_event(
    event: dict[str, Any],
    *,
    repository: str,
) -> IssueCommentNotification | None:
    """Build a Slack notification model from a GitHub issue_comment event."""
    issue = event.get("issue")
    comment = event.get("comment")
    if not isinstance(issue, dict) or not isinstance(comment, dict):
        return None

    # GitHub uses the same event for PR comments; ignore those here.
    if issue.get("pull_request"):
        return None

    issue_number = issue.get("number")
    if not isinstance(issue_number, int):
        return None

    issue_title = _string(issue.get("title"))
    issue_url = _string(issue.get("html_url"))
    comment_url = _string(comment.get("html_url"))
    comment_body = _string(comment.get("body"))
    issue_author = _string((issue.get("user") or {}).get("login"))
    comment_author = _string((comment.get("user") or {}).get("login"))

    if not all((repository, issue_title, issue_url, comment_url, comment_author)):
        return None

    return IssueCommentNotification(
        repository=repository,
        issue_number=issue_number,
        issue_title=issue_title,
        issue_url=issue_url,
        issue_author=issue_author or "unknown",
        comment_author=comment_author,
        comment_url=comment_url,
        comment_body=comment_body,
    )


def build_slack_payload(notification: IssueCommentNotification) -> dict[str, Any]:
    """Render a compact Slack incoming webhook payload."""
    comment_preview = _truncate_comment(notification.comment_body)
    issue_link = f"<{notification.issue_url}|{notification.repository}#{notification.issue_number}>"
    comment_link = f"<{notification.comment_url}|View comment>"

    preview_text = comment_preview or "_No comment body provided._"
    preview_block = "\n".join(f"> {line}" for line in preview_text.splitlines()) or "> "

    text = (
        f"New comment on {notification.repository}#{notification.issue_number} by "
        f"{notification.comment_author}: {notification.issue_title}"
    )

    return {
        "text": text,
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"*New GitHub issue comment*\n{issue_link}: *{notification.issue_title}*"
                    ),
                },
            },
            {
                "type": "context",
                "elements": [
                    {
                        "type": "mrkdwn",
                        "text": (
                            f"*Commenter:* `{notification.comment_author}`"
                            f"    *Issue author:* `{notification.issue_author}`"
                            f"    {comment_link}"
                        ),
                    }
                ],
            },
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": preview_block},
            },
        ],
    }


def send_slack_webhook(payload: dict[str, Any], webhook_url: str) -> None:
    """Send a payload to a Slack incoming webhook."""
    parsed = parse.urlparse(webhook_url)
    if parsed.scheme != "https" or parsed.netloc != "hooks.slack.com":
        raise RuntimeError("Slack webhook URL must target https://hooks.slack.com")
    body = json.dumps(payload).encode("utf-8")
    req = request.Request(
        webhook_url,
        data=body,
        headers={"Content-Type": "application/json; charset=utf-8"},
        method="POST",
    )
    try:
        with _open_slack_webhook(req) as response:
            status_code = getattr(response, "status", response.getcode())
    except error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"Slack webhook failed with HTTP {exc.code}: {detail}") from exc
    except error.URLError as exc:
        raise RuntimeError(f"Slack webhook failed: {exc.reason}") from exc

    if status_code >= 400:
        raise RuntimeError(f"Slack webhook failed with HTTP {status_code}")


def _open_slack_webhook(req: request.Request):
    return request.urlopen(req, timeout=10)  # nosemgrep


def main() -> int:
    """Entrypoint used by the GitHub Actions workflow."""
    event_path = _string(os.getenv("GITHUB_EVENT_PATH"))
    repository = _string(os.getenv("GITHUB_REPOSITORY"))
    webhook_url = _string(
        os.getenv("SLACK_GITHUB_ISSUES_WEBHOOK_URL") or os.getenv("SLACK_WEBHOOK_URL")
    )

    if not event_path:
        print("Missing GITHUB_EVENT_PATH.", file=sys.stderr)
        return 1
    if not repository:
        print("Missing GITHUB_REPOSITORY.", file=sys.stderr)
        return 1
    if not webhook_url:
        print("Skipped: Slack webhook is not configured.")
        return 0

    event = json.loads(Path(event_path).read_text(encoding="utf-8"))
    notification = notification_from_issue_comment_event(event, repository=repository)
    if notification is None:
        print("Skipped: event is not a normal issue comment.")
        return 0

    payload = build_slack_payload(notification)
    send_slack_webhook(payload, webhook_url)
    print(f"Posted Slack notification for {notification.repository}#{notification.issue_number}.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())