-
Notifications
You must be signed in to change notification settings - Fork 0
/
tg-spigot.py
127 lines (101 loc) · 4.05 KB
/
tg-spigot.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
import os
import time
import subprocess
import threading
import logging as log
import sys
import signal
from incoming import from_tg
from outgoing import to_tg
log.basicConfig(
format='%(asctime)s [%(levelname)s %(threadName)s] %(message)s',
handlers=[log.FileHandler('data/bot.log'), log.StreamHandler(sys.stdout)],
level=log.DEBUG,
)
# remove "Starting new HTTPS connection" etc.
log.getLogger('urllib3').setLevel(log.INFO)
def log_exit(sig, *args):
log.info(f"bot stopped: {sig} {signal.strsignal(sig)}\n\n")
os._exit(0)
for sig in (signal.SIGABRT, signal.SIGINT, signal.SIGTERM):
signal.signal(sig, log_exit)
bot_dir = os.path.dirname(os.path.realpath(__file__))
os.chdir(bot_dir)
def get_cfg():
if os.path.exists('config.py'):
global cfg
import config as cfg
else:
with open('config.py', 'w') as f:
f.write(
'\n'.join(
[
"# Spigot server's path without slash at the end",
'spigot_path = ""',
"# Telegram bot's token",
'token = ""',
'# tg username (w/o @) of the user who will be able to send commands',
'tg_admin_username = "wanours"',
'# minecraft username of admin',
'mc_admin_username = "okarkalic"',
'# ID of chat which will get messages from Minecraft (user or group)',
'chat_id = 0',
'# ID of chat which will get the entire content of the spigot console (user or group)',
'console_id = 0',
'# @username of channel with login/logout messages',
'channel_id = "@"',
'# whole words triggering admin mention, separated by vert lines',
'triggers = "admin|admina"',
'# request timeout in seconds (appears that max is 50 or so)',
'timeout = 300',
]
)
)
log.warning('config genereated - fill it and restart')
exit()
def from_tg_loop():
while True:
formatted_lines = from_tg()
if formatted_lines:
# screen -X won't work with other users, it's a bug which has been fixed after
# something like 10 years, after I reminded the screen developers of that bug in 2020.
# The fix hasn't been released yet so I must use sudo to execute commands in other users' screen
subprocess.run(['sudo', '-u', 'mc', '/home/mc/command.sh', formatted_lines])
def to_tg_loop():
mc_log_file = cfg.spigot_path + '/tg/temp.log'
mc_chat_file = cfg.spigot_path + '/tg/chat.temp.log'
while True:
with open(mc_log_file, 'r+') as l, open(mc_chat_file, 'r+') as c:
mc_log = l.read()
mc_chat = c.read()
l.truncate(0)
c.truncate(0)
if mc_log or mc_chat:
to_tg(mc_log, mc_chat)
time.sleep(0.2)
if __name__ == '__main__':
log.info(" ")
log.info(" BOT STARTED")
log.info(" ")
get_cfg()
# reset online users (user count will be inaccurate)
with open('data/users.txt', 'w') as file:
pass
def handle_exception(exc_type, exc_value, exc_traceback, thread=None):
if issubclass(exc_type, KeyboardInterrupt):
return
in_thread = f" in thread {thread}" if thread else ""
log.error(
f"Uncaught exception{in_thread}",
exc_info=(exc_type, exc_value, exc_traceback),
)
def handle_threading_exception(args):
handle_exception(args.exc_type, args.exc_value, args.exc_traceback, args.thread)
sys.excepthook = handle_exception
threading.excepthook = handle_threading_exception
in_thread = threading.Thread(target=from_tg_loop)
in_thread.name = "from_tg"
out_thread = threading.Thread(target=to_tg_loop)
out_thread.name = "to_tg"
in_thread.start()
out_thread.start()