-
-
Notifications
You must be signed in to change notification settings - Fork 160
/
utils.py
180 lines (154 loc) · 5.35 KB
/
utils.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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
import os
import subprocess
import psutil
import yaml
from telegram import Bot
from btb_manager_telegram import logger, scheduler, settings
def setup_root_path_constant():
if settings.ROOT_PATH is None:
logger.info("No root_path was specified. Aborting.")
exit(-1)
else:
settings.ROOT_PATH = os.path.join(settings.ROOT_PATH, "")
def setup_telegram_constants():
logger.info("Retrieving Telegram token and user_id from apprise.yml file.")
telegram_url = None
yaml_file_path = f"{settings.ROOT_PATH}config/apprise.yml"
if os.path.exists(yaml_file_path):
with open(yaml_file_path) as f:
try:
parsed_urls = yaml.load(f, Loader=yaml.FullLoader)["urls"]
except Exception:
logger.error(
"Unable to correctly read apprise.yml file. Make sure it is correctly set up. Aborting."
)
exit(-1)
for url in parsed_urls:
if url.startswith("tgram"):
telegram_url = url.split("//")[1]
if not telegram_url:
logger.error(
"No telegram configuration was found in your apprise.yml file. Aborting."
)
exit(-1)
else:
logger.error(
f'Unable to find apprise.yml file at "{yaml_file_path}". Aborting.'
)
exit(-1)
try:
settings.TOKEN = telegram_url.split("/")[0]
settings.USER_ID = telegram_url.split("/")[1]
logger.info(
f"Successfully retrieved Telegram configuration. "
f"The bot will only respond to user with user_id {settings.USER_ID}"
)
except Exception:
logger.error(
"No user_id has been set in the yaml configuration, anyone would be able to control your bot. Aborting."
)
exit(-1)
def text_4096_cutter(m_list):
message = [""]
index = 0
for mes in m_list:
if len(message[index]) + len(mes) <= 4096:
message[index] += mes
else:
message.append(mes)
index += 1
return message
def find_process():
return any(
"binance_trade_bot" in proc.name()
or "binance_trade_bot" in " ".join(proc.cmdline())
for proc in psutil.process_iter()
)
def find_and_kill_process():
try:
for proc in psutil.process_iter():
if "binance_trade_bot" in proc.name() or "binance_trade_bot" in " ".join(
proc.cmdline()
):
proc.terminate()
proc.wait()
except Exception as e:
logger.info(f"ERROR: {e}")
def is_tg_bot_update_available():
try:
proc = subprocess.Popen(
["bash", "-c", "git remote update && git status -uno"],
stdout=subprocess.PIPE,
)
output, _ = proc.communicate()
re = "Your branch is behind" in str(output)
except Exception:
re = None
return re
def is_btb_bot_update_available():
try:
proc = subprocess.Popen(
[
"bash",
"-c",
f"cd {settings.ROOT_PATH} && git remote update && git status -uno",
],
stdout=subprocess.PIPE,
)
output, _ = proc.communicate()
re = "Your branch is behind" in str(output)
except Exception:
re = None
return re
def update_checker():
logger.info("Checking for updates.")
if settings.TG_UPDATE_BROADCASTED_BEFORE is False:
if is_tg_bot_update_available():
logger.info("BTB Manager Telegram update found.")
message = (
"⚠ An update for _BTB Manager Telegram_ is available\.\n\n"
"Please update by going to *🛠 Maintenance* and pressing the *Update Telegram Bot* button\."
)
settings.TG_UPDATE_BROADCASTED_BEFORE = True
bot = Bot(settings.TOKEN)
bot.send_message(settings.USER_ID, message, parse_mode="MarkdownV2")
scheduler.enter(
60 * 60 * 12,
1,
update_reminder,
("_*Reminder*_:\n\n" + message,),
)
if settings.BTB_UPDATE_BROADCASTED_BEFORE is False:
if is_btb_bot_update_available():
logger.info("Binance Trade Bot update found.")
message = (
"⚠ An update for _Binance Trade Bot_ is available\.\n\n"
"Please update by going to *🛠 Maintenance* and pressing the *Update Binance Trade Bot* button\."
)
settings.BTB_UPDATE_BROADCASTED_BEFORE = True
bot = Bot(settings.TOKEN)
bot.send_message(settings.USER_ID, message, parse_mode="MarkdownV2")
scheduler.enter(
60 * 60 * 12,
1,
update_reminder,
("_*Reminder*_:\n\n" + message,),
)
if (
settings.TG_UPDATE_BROADCASTED_BEFORE is False
or settings.BTB_UPDATE_BROADCASTED_BEFORE is False
):
scheduler.enter(
60 * 60,
1,
update_checker,
)
def update_reminder(self, message):
logger.info(f"Reminding user: {message}")
bot = Bot(settings.TOKEN)
bot.send_message(settings.USER_ID, message, parse_mode="MarkdownV2")
scheduler.enter(
60 * 60 * 12,
1,
update_reminder,
)