1
0

Merge pull request #4 from yamaks2306/dev

Fixed bug with angle brackets
This commit is contained in:
yamaks2306 2022-07-15 17:24:08 +03:00 committed by GitHub
commit 30a477852c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 137 additions and 76 deletions

2
.gitignore vendored Normal file

@ -0,0 +1,2 @@
__pycache__/
.testenvs

6
constant.py Normal file

@ -0,0 +1,6 @@
ICON = {
"failure": "🔴",
"cancelled": "",
"success": "🟢",
}

@ -1,2 +1,2 @@
#!/bin/bash
python /send_message.py
python /main.py

46
environments.py Normal file

@ -0,0 +1,46 @@
class Environment:
'''Class containing all the necessary variables for building and sending notification'''
def __init__(
self,
chat_id,
token,
status,
commit,
include_commit_info,
docker_tags,
custom_message,
github_workflow,
github_repository,
github_sha,
github_actor,
tag
):
self.chat_id = chat_id
self.token = token
self.status = status
self.commit = self.__fix_commit_message(commit)
self.include_commit_info = include_commit_info
self.docker_tags = docker_tags
self.custom_message = custom_message
self.github_workflow = github_workflow
self.github_repository = github_repository
self.github_sha = github_sha
self.github_actor = github_actor
self.tag = tag
def get_link(self):
'''Get link for sending message'''
return f'https://api.telegram.org/bot{self.token}/sendMessage'
def get_commit_link(self):
'''Get link to commit'''
return f'https://github.com/{self.github_repository}/commit/{self.github_sha}'
def __fix_commit_message(self, message):
'''Replacing unsupported characters'''
replace_dict = {'<':'(', '>':')', '\n':'\n '}
result = ''.join(i if i not in replace_dict else replace_dict[i] for i in message)
return result
def get_version(self):
'''GITHUB_REF contain "refs/tags/v0.0.2" or "refs/heads/main". If second part is "tags",
return tag with version number, else - None'''
tags = self.tag.split("/")
return tags[2] if tags[1] == "tags" else None

39
main.py Normal file

@ -0,0 +1,39 @@
import os
from message import Message
from environments import Environment
chat_id = os.getenv('INPUT_CHAT_ID')
if chat_id == "":
raise SystemExit('Variable chat_id is required. Exit.')
token = os.getenv('INPUT_TOKEN')
if token == "":
raise SystemExit('Variable chat_id is required. Exit.')
status = os.getenv('INPUT_STATUS')
commit = os.getenv('INPUT_COMMIT_MESSAGE')
include_commit_info = os.getenv('INPUT_INCLUDE_COMMIT_INFO')
docker_tags = os.getenv('INPUT_DOCKER_TAGS')
custom_message = os.getenv('INPUT_MESSAGE')
#GitHub environment variables
github_workflow = os.getenv('GITHUB_WORKFLOW')
github_repository = os.getenv('GITHUB_REPOSITORY')
github_sha = os.getenv('GITHUB_SHA')
github_actor = os.getenv('GITHUB_ACTOR')
tag = os.getenv('GITHUB_REF')
envs = Environment(
chat_id,
token,
status,
commit,
include_commit_info,
docker_tags,
custom_message,
github_workflow,
github_repository,
github_sha,
github_actor,
tag
)
message = Message(envs)
message.send_message()

43
message.py Normal file

@ -0,0 +1,43 @@
import requests
import constant
import environments
class Message:
'''A class that builds and sends messages'''
def __init__(self, envs: environments.Environment):
self.envs = envs
def __build_message(self):
'''Building message from different parts'''
message = f'''{constant.ICON[self.envs.status]} {self.envs.status}: <b>{self.envs.github_workflow}</b>\n
Repository: <b>{self.envs.github_repository}</b>'''
version = self.envs.get_version()
if version is not None:
message = f'''{message}
Version: <b>{version}</b>'''
if self.envs.docker_tags != "":
message = f'''{message}
Docker image tags: <b>{self.envs.docker_tags}</b>'''
if self.envs.include_commit_info == "true":
message = f'''{message}
Author: <b>{self.envs.github_actor}</b>
Commit message: <b>{self.envs.commit}</b>
<a href="{self.envs.get_commit_link()}">See changes</a>'''
return message
def send_message(self):
'''Sending message to telegram chat'''
message = self.__build_message() if self.envs.custom_message == "" else self.envs.custom_message
parameters = {
'chat_id': self.envs.chat_id,
'text': message,
'parse_mode': 'HTML',
'disable_web_page_preview': True
}
request = requests.get(self.envs.get_link(), params = parameters)
request.raise_for_status()

@ -1,75 +0,0 @@
import os
import requests
chat_id = os.getenv('INPUT_CHAT_ID')
if chat_id == "":
raise SystemExit('Variable chat_id is required. Exit.')
token = os.getenv('INPUT_TOKEN')
if token == "":
raise SystemExit('Variable chat_id is required. Exit.')
status = os.getenv('INPUT_STATUS')
commit = os.getenv('INPUT_COMMIT_MESSAGE')
include_commit_info = os.getenv('INPUT_INCLUDE_COMMIT_INFO')
docker_tags = os.getenv('INPUT_DOCKER_TAGS')
custom_message = os.getenv('INPUT_MESSAGE')
link=f'https://api.telegram.org/bot{token}/sendMessage'
icon = {
"failure": "🔴",
"cancelled": "",
"success": "🟢",
}
#GitHub environment variables
github_workflow = os.getenv('GITHUB_WORKFLOW')
github_repository = os.getenv('GITHUB_REPOSITORY')
github_sha = os.getenv('GITHUB_SHA')
github_actor = os.getenv('GITHUB_ACTOR')
tag = os.getenv('GITHUB_REF')
commit_link = f'https://github.com/{github_repository}/commit/{github_sha}'
def get_version(tag_string):
'''GITHUB_REF contain "refs/tags/v0.0.2" or "refs/heads/main". If second part is "tags",
return tag with version number, else - None'''
tags = tag_string.split("/")
return tags[2] if tags[1] == "tags" else None
def build_message():
'''Building message from different parts'''
message = f'''{icon[status]} {status}: <b>{github_workflow}</b>\n
Repository: <b>{github_repository}</b>'''
version = get_version(tag)
if version is not None:
message = f'''{message}
Version: <b>{version}</b>'''
if docker_tags != "":
message = f'''{message}
Docker image tags: <b>{docker_tags}</b>'''
if include_commit_info == "true":
message = f'''{message}
Author: <b>{github_actor}</b>
Commit message: <b>{commit}</b>
<a href="{commit_link}">See changes</a>'''
return message
def send_message():
'''Sending message to telegram chat'''
message = build_message() if custom_message == "" else custom_message
parameters = {
'chat_id': chat_id,
'text': message,
'parse_mode': 'HTML',
'disable_web_page_preview': True
}
request = requests.get(link, params = parameters)
request.raise_for_status()
send_message()