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
|
import os, requests
from typing import List, Tuple, AnyStr
def _get_vars(params: List[Tuple], kw: dict) -> List[AnyStr]:
result = []
for kw_name, env_name in params:
env_name = f'MY_{env_name}'
if kw_name in kw:
result.append(kw[kw_name])
elif env_name in os.environ:
result.append(os.environ[env_name])
else:
raise RuntimeError("missing parameter %s or variable %s" % (kw_name, env_name))
return result
def telegram_notify(text: str, parse_mode: str = None, **kwargs):
chat_id, token = _get_vars([
('chat_id', 'TELEGRAM_NOTIFY_CHAT_ID'),
('token', 'TELEGRAM_NOTIFY_TOKEN')
], kwargs)
data = {
'chat_id': chat_id,
'text': text
}
if parse_mode is not None:
data['parse_mode'] = parse_mode
r = requests.post('https://api.telegram.org/bot%s/sendMessage' % token, data=data)
if r.status_code != 200:
raise RuntimeError("telegram returned %d" % r.status_code)
|
