-
Notifications
You must be signed in to change notification settings - Fork 967
/
slack.py
108 lines (89 loc) · 3.66 KB
/
slack.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
"""The module describes the ``slack`` error plugin plugin."""
import logging
import uuid
from typing import (
Any,
Dict,
)
import requests
from galaxy.util import string_as_bool
from .base_git import BaseGitPlugin
log = logging.getLogger(__name__)
class SlackPlugin(BaseGitPlugin):
"""Send error report to Sentry."""
plugin_type = "slack"
def __init__(self, **kwargs):
self.app = kwargs["app"]
self.redact_user_details_in_bugreport = self.app.config.redact_user_details_in_bugreport
self.verbose = string_as_bool(kwargs.get("verbose", False))
self.user_submission = string_as_bool(kwargs.get("user_submission", False))
self.webhook_url = kwargs.get("webhook_url", "https://localhost/")
def _append_issue(self, *args, **kwargs):
pass
def _create_issue(self, *args, **kwargs):
pass
def _fill_issue_cache(self, *args, **kwargs):
pass
def submit_report(self, dataset, job, tool, **kwargs):
history_id_encoded = self.app.security.encode_id(dataset.history_id)
history_view_link = self.app.url_for("/histories/view", id=history_id_encoded, qualified=True)
error_report_id = str(uuid.uuid4())[0:13]
title = self._generate_error_title(job)
blocks: Dict[str, Any] = {
"blocks": [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"You have a new Galaxy bug report:\n*<{history_view_link}|{title}>*",
},
},
]
}
if "message" in kwargs and kwargs["message"] is not None:
message = kwargs["message"].strip().replace("\n", "\n> ")
blocks["blocks"].extend(
[{"type": "section", "text": {"type": "mrkdwn", "text": f"*Message*\n> {message}```"}}]
)
blocks["blocks"].extend(
[
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Error Report ID:*\n`{error_report_id}`"},
{"type": "mrkdwn", "text": f"*ID:*\n{job.id}"},
{"type": "mrkdwn", "text": f"*Destination ID:*\n{job.destination_id}"},
{"type": "mrkdwn", "text": f"*Exit Code:*\n{job.exit_code}"},
{"type": "mrkdwn", "text": f"*Handler:*\n{job.handler}"},
{"type": "mrkdwn", "text": f"*Tool Version:*\n{job.tool_version}"},
{"type": "mrkdwn", "text": f"*User:*\n{job.get_user().id}"},
],
},
{"type": "divider"},
{
"type": "section",
"text": {"type": "mrkdwn", "text": f"*Command Line:*\n```\n{job.command_line[0:2800]}\n```"},
},
]
)
if job.stdout:
blocks["blocks"].extend(
[
{
"type": "section",
"text": {"type": "mrkdwn", "text": f"*Stdout*\n```\n{job.stdout.strip()[0:2800]}\n```"},
}
]
)
if job.stderr:
blocks["blocks"].extend(
[
{
"type": "section",
"text": {"type": "mrkdwn", "text": f"*Stderr*\n```\n{job.stderr.strip()[0:2800]}\n```"},
}
]
)
requests.post(self.webhook_url, json=blocks)
return (f"Sent report to Slack with ID: {error_report_id}", "success")
__all__ = ("SlackPlugin",)