1
0
mirror of https://github.com/vimagick/dockerfiles.git synced 2024-12-23 01:39:27 +02:00
dockerfiles/urlwatch/data/reporters.py

70 lines
2.4 KiB
Python
Raw Normal View History

2024-02-26 07:34:21 +02:00
class SlackReporter(TextReporter):
"""Send a message to a Slack channel"""
__kind__ = 'slack'
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.max_length = self.config.get('max_message_length', 40000)
def submit(self):
webhook_url = self.config['webhook_url']
text = '\n'.join(super().submit())
if not text:
logger.debug('Not calling {} API (no changes)'.format(self.__kind__))
return
result = None
for chunk in chunkstring(text, self.max_length, numbering=True):
res = self.submit_chunk(webhook_url, chunk)
if res.status_code != requests.codes.ok or res is None:
result = res
return result
def submit_chunk(self, webhook_url, text):
logger.debug("Sending {} request with text: {}".format(self.__kind__, text))
post_data = self.prepare_post_data(text)
result = requests.post(webhook_url, json=post_data)
try:
if result.status_code == requests.codes.ok:
logger.info("{} response: ok".format(self.__kind__))
else:
logger.error("{} error: {}".format(self.__kind__, result.text))
except ValueError:
logger.error(
"Failed to parse {} response. HTTP status code: {}, content: {}".format(self.__kind__,
result.status_code,
result.content))
return result
def prepare_post_data(self, text):
if self.config.get('rich_text', False):
return {
"blocks": [
{
"type": "rich_text",
"elements": [
{
"type": "rich_text_preformatted",
"elements": [
{"type": "text", "text": text}
]
}
]
}
]
}
else:
return {"text": text}
class MattermostReporter(SlackReporter):
"""Send a message to a Mattermost channel"""
__kind__ = 'mattermost'
def prepare_post_data(self, text):
return {"text": text}