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
|
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import datetime as dt
import logging
import sys
import time
from contextlib import suppress
import ts3
from .exceptions import DateException
__all__ = ['GroupAssigner']
class GroupAssigner:
def __init__(self, host: str, port: (str, int), user: str, password: str, sid: (str, int), gid: (str, int),
date: dt.date, delta: dt.timedelta):
# host config
self.host = host
self.port = port
self.user = user
self.pw = password
self.sid = sid
# group
self.gid = gid
# start date and delta
self.sleepstart = date - dt.timedelta(days=1)
self.startdate = date
self.enddate = date + delta
self.delta = delta
def __connect(self):
""" establish query connection and return connection handler """
# connect to the telnet interface
self.conn = ts3.query.TS3Connection(self.host, self.port)
# login
self.conn.login(client_login_name=self.user, client_login_password=self.pw)
# select specified sid
self.conn.use(sid=self.sid)
def __disconnect(self):
""" gracefully logout and disconnect the connection, this should only be called if the exit is intentional """
try:
self.conn.logout()
self.conn.quit()
# attribute error if disconnect is called prior to the connection being established
except AttributeError:
pass
# broad exception if something unexpected happens
except ts3.TS3Error as TS3Error:
# log exception
logging.error(TS3Error)
# exit
sys.exit()
def __checkdate(self):
""" method to check if the current date is still in the configured date range """
now = dt.date.today()
# check if target date is in the configured range
if self.startdate <= now <= self.enddate:
logging.debug('target date within configured date range')
return True
# if date range is exceeded shutdown gracefully
logging.info('the current date exceeds the configured date range -- exiting')
self.__disconnect()
def __start_sleepstart(self):
""" method to check if the process is eligible to sleepstart """
now = dt.date.today()
# start date already reached proceed
if self.startdate <= now:
logging.debug('start date is already reached -- not eligible to sleepstart continue')
# if startdate within the next 24h proceed to sleepstart
elif now >= self.startdate - dt.timedelta(days=1):
# convert both dates to datetime
starttime = dt.datetime.combine(self.startdate, dt.time(hour=0, minute=0, second=0))
now = dt.datetime.now()
# calculate remaining time delta
remaindelta = starttime - now
logging.debug(f'target date will be reached in {remaindelta.seconds} seconds -- sleeping')
time.sleep(remaindelta.seconds + 1)
else:
# if the date is too far back raise DateException
raise DateException('target date is too far in the future')
def __notifycliententerview(self, data: dict):
"""
event thrown if a client connects to the server
:param data: dictionary containing users info
"""
# return all non voice clients without reasonid 0
if data['client_type'] != '0' or data['reasonid'] != '0':
return
cldbid = data['client_database_id']
user_grps = data['client_servergroups'].split(sep=',')
msg = '{client_nickname}:{client_database_id} connected - member of {client_servergroups}'
logging.debug(msg.format(**data))
# only try to add nonmembers to group
if str(self.gid) not in user_grps:
try:
# Usage: servergroupaddclient sgid={groupID} cldbid={clientDBID}
cmd = self.conn.servergroupaddclient(sgid=self.gid, cldbid=cldbid)
if cmd.error['id'] != '0':
logging.error(cmd.data[0].decode("utf-8"))
# log process
logging.info('{client_nickname}:{client_database_id} added to {gid}'.format(**data, gid=self.gid))
# log possible key errors while the teamspeak 5 client is not fully released
except KeyError as err:
logging.error([err, data])
def __handle_event(self, event: str, data: dict):
""" event handler which separates events to their specific handlers """
# check if event is still eligible
self.__checkdate()
# client enter events
if event == "notifycliententerview":
self.__notifycliententerview(data)
def start(self):
""" main entry point to start the bot """
# eol to start process ahead of time
self.__start_sleepstart()
# proceed only if target date is inside the date range
if self.__checkdate():
try:
# init connection
self.__connect()
# break if credentials are invalid
except ts3.query.TS3QueryError as err:
# log error
logging.error(err)
self.__disconnect()
# start processing
self.__main()
def __main(self):
""" bots main loop """
# register for "server" notify event
self.conn.servernotifyregister(event="server")
# start listening and processing
while True:
self.conn.send_keepalive()
# suppress TimeoutError exceptions
with suppress(ts3.query.TS3TimeoutError):
# wait for events
event = self.conn.wait_for_event(timeout=60)
# handover event to eventhandler
self.__handle_event(event.event, event.parsed[0])
|
