forked from danigm/gitlab-telegram-bot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
executable file
·148 lines (118 loc) · 4.83 KB
/
app.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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
#!/usr/bin/env python3
import json
from flask import Flask
from flask import request
from flask import jsonify
from bot import Bot
import jinja2
import os
from config import AUTH_MSG
app = Flask(__name__)
class GitlabBot(Bot):
def __init__(self):
try:
self.authmsg = AUTH_MSG
except:
raise Exception("The authorization messsage file is invalid")
super(GitlabBot, self).__init__()
self.chats = {}
try:
chats = open('chats', 'r').read()
self.chats = json.loads(chats)
except:
open('chats', 'w').write(json.dumps(self.chats))
self.send_to_all('Hi !')
def text_recv(self, txt, chatid):
''' registering chats '''
txt = txt.strip()
if txt.startswith('/'):
txt = txt[1:]
if txt == self.authmsg:
if str(chatid) in self.chats:
self.reply(
chatid, "\U0001F60E Вы уже авторизованы.")
else:
self.reply(chatid, "\U0001F60E Авторизация прошла успешно!")
self.chats[chatid] = True
open('chats', 'w').write(json.dumps(self.chats))
elif txt == 'shutupbot':
del self.chats[chatid]
self.reply(chatid, "\U0001F63F Извиись.")
open('chats', 'w').write(json.dumps(self.chats))
else:
self.reply(chatid, "\U0001F612 Нет доступа.")
def send_to_all(self, msg):
print(self.chats)
for c in self.chats:
self.reply(c, msg)
b = GitlabBot()
environment = jinja2.Environment()
template = environment.from_string(
"[{{name}} - {{title}}]({{link}}) \n\n{{source_branch}} -> {{target_branch}} \n\n👨💻️: {{user}} \n{{comment}}")
def requestStatus(status):
if status == 'open':
return '⭐️ Новый МР'
elif status == 'close':
return '🪳 МР удален'
elif status == 'update':
return '✍🏻 Внесены изменения'
elif status == 'reopen':
return '🤡 МР вновь открыт'
elif status == 'merge':
return '🚣🏾 Залито'
def templateRender(type,title,data):
if type=='mr':
return title+"\n"+template.render(name=data['project']['name'],
link=data['object_attributes']['url'],
title=data['object_attributes']['title'],
source_branch=data['object_attributes']['source_branch'],
target_branch=data['object_attributes']['target_branch'],
user=data['user']['name']
)
else:
return title+"\n"+template.render(name=data['project']['name'],
link=data['object_attributes']['url'],
title=data['merge_request']['title'],
source_branch=data['merge_request']['source_branch'],
target_branch=data['merge_request']['target_branch'],
user=data['user']['name'],
comment='Комментарий: '+data['object_attributes']['note']
)
@app.route("/", methods=['GET', 'POST'])
def webhook():
data = request.json
# json contains an attribute that differenciates between the types, see
# https://docs.gitlab.com/ce/user/project/integrations/webhooks.html
# for more infos
kind = data['object_kind']
if kind == 'merge_request':
msg = templateRender('mr',requestStatus(data['object_attributes']['action']), data)
elif kind == 'push':
msg = generatePushMsg(data)
elif kind == 'note':
msg = generateCommentMsg(data)
# TODO: мб добавить обработку других событий
b.send_to_all(msg)
return jsonify({'status': 'ok'})
def generatePushMsg(data):
msg = '✍🏻 [{0}]({1}) - {2} ({3} новый коммит) \n'\
.format(data['project']['name'], data['object_attributes']['url'], data['object_attributes']['title'],data['total_commits_count'])
for commit in data['commits']:
msg = msg + '----------------------------------------------------------------\n'
msg = msg + commit['message'].rstrip()
msg = msg + '\n' + commit['url'].replace("_", "\_") + '\n'
msg = msg + '----------------------------------------------------------------\n'
return msg
def generateCommentMsg(data):
print(data)
ntype = data['object_attributes']['noteable_type']
if ntype == 'Commit':
msg = templateRender('comment','🪃 Новый коммит', data)
elif ntype == 'MergeRequest':
msg = templateRender('comment','🪃 Новый комментарий', data)
elif ntype == 'Issue':
msg = templateRender('comment','🪃', data)
return msg
if __name__ == "__main__":
b.run_threaded()
app.run(host='0.0.0.0', port=10111)