release 1.0

This commit is contained in:
Lukas 2022-04-14 17:05:39 +02:00
parent b22d16e364
commit a8566dfdde
4 changed files with 181 additions and 92 deletions

143
myTS3.py
View File

@ -3,8 +3,8 @@ TBD
"""
__author__ = "Lukas Mahler"
__version__ = "0.0.0"
__date__ = "20.03.2022"
__version__ = "1.0.0"
__date__ = "14.04.2022"
__email__ = "m@hler.eu"
__status__ = "Development"
@ -86,7 +86,7 @@ class TSbot:
self.pipeOut("Can't find my own client id. terminating...", lvl="critical")
exit(1)
''' if you want to move the Bot to a certain channel (instead of the defualt channel, you can do: '''
''' if you want to move the Bot to a certain channel (instead of the default channel, you can do: '''
# self.bot.clientmove(clid=self.myid,cid=129)
# ----------- SUBSCRIBE TO EVENTS -------------
@ -114,7 +114,7 @@ class TSbot:
# ----------- LOOP HERE -------------
while self.running:
# self.bot.send_keepalive()
self.pipeOut(f"Waiting for a new Event...")
self.pipeOut(f"Waiting for a new Event...", lvl="debug")
self.bot.version()
# Auto-update crypto price channels every 30 minutes
@ -183,11 +183,22 @@ class TSbot:
# New text message
elif event_type == "notifytextmessage":
msg = event[0]["msg"].replace("\n", " ")
invkr = event[0]["invokername"]
invkr_name = event[0]["invokername"]
invkr_id = event[0]["invokerid"]
self.pipeOut(f'Message | From: "{invkr}" | Content: "{msg}"')
invkr_uid = event[0]["invokeruid"]
invkr_dbid = self.bot.clientgetdbidfromuid(cluid=invkr_uid)[0]["cldbid"]
permission_level = 1 if self.isadmin(invkr_dbid) else 0
invoker = {'name': invkr_name,
'id': invkr_id,
'uid': invkr_uid,
'dbid': invkr_dbid}
self.pipeOut(f'Message | From: "{invkr_name}" | '
f'Permissionlevel: {permission_level} | Content: "{msg}"')
if msg.startswith("."):
self.lookupcommand(msg, invkr_id)
self.lookupcommand(msg, invoker, permission_level)
# Unknown Event
else:
@ -206,19 +217,23 @@ class TSbot:
lvln = int(getattr(util.logging, lvl))
self.log.log(lvln, msg)
if reprint:
if reprint and lvln >= self.log.getEffectiveLevel():
print(f"[{time.strftime('%H:%M:%S')}]{f'[{lvl}]':10s} {msg}")
if not self.whitelisted:
if msg == "error id 524: client is flooding":
time.sleep(30)
if lvl == "CRITICAL":
exit(1)
def stop(self, invkr_id):
def stop(self, invoker):
"""
This stops the bot instance by invalidating the event loop.
"""
msg = "Shutdown, bye bye!"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=msg)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
self.pipeOut(msg)
self.running = False
@ -259,7 +274,7 @@ class TSbot:
for clid in clients:
data = self.printable_clientinfo(clid)
self.pipeOut(f"{data}", lvl="debug")
self.pipeOut(f"{data}", lvl="DEBUG")
i = 0
while n == -1 or i < n:
@ -267,7 +282,7 @@ class TSbot:
try:
self.bot.clientpoke(msg=msg, clid=clid)
except Exception as e:
self.pipeOut(e, lvl="warning")
self.pipeOut(e, lvl="WARNING")
pass
time.sleep(delay)
i += 1
@ -310,7 +325,7 @@ class TSbot:
try:
client = self.bot.clientinfo(clid=clid)
except ts3.query.TS3QueryError as e:
self.pipeOut(f"given clid {clid} returned error:\n{e}", lvl="ERROR")
self.pipeOut(f"ISQUERY: given clid {clid} returned: {e}", lvl="WARNING")
return True
try:
if client[0]["client_type"] == "1":
@ -387,91 +402,96 @@ class TSbot:
# Commands -------------------------------------------------------------------------------------------------------------
def lookupcommand(self, msg, invkr_id):
def lookupcommand(self, msg, invoker, permission_level):
"""
Every message starting with '.' gets passed and evaluated in this function.
Commands in this function are sorted alphabetically.
Parameters in brackets are optional.
Command Parameter 1 Parameter 2
---------------------------------------------------------
.admin / .notifyAdmin
.annoy target message
.follow target
.help / .h
.info clid
.kickall message
.list channel/clients/commands/groups
.mute target
.pingall message
.rename nickname
.rwf
.roll
.stop / .quit / .q
.test
.ticker (symbol)
.unmute target
Currently only 2 permission levels are implemented [0 = Everyone, 1 = Admins]
Command Parameter 1 Parameter 2 Permissionlevel
--------------------------------------------------------------------------------------
.admin / .notifyAdmin 0
.annoy target message 1
.follow target 1
.help / .h 0
.info clid 0
.kickall message 1
.list channel/clients/commands/groups 1
.mute target 1
.pingall message 1
.rename nickname 1
.rmb amount 0
.roll 0
.rwf 0
.stop / .quit / .q 1
.test 1
.ticker (symbol) 0
.unmute target 1
"""
commandstring = msg.split(" ")
command = commandstring[0]
parameter = commandstring[1:]
self.pipeOut(f"command: {command} | parameter: {parameter} | invkr_id: {invkr_id}")
# ??? passable = {"self": self, "invkr_id": invkr_id, "parameter": parameter}
self.pipeOut(f"command: {command} | parameter: {parameter} | "
f"from: {invoker['name']} | permissionlevel: {permission_level}")
if command == ".admin" or command == ".notifyAdmin":
commands.notifyAdmin(self)
elif command == ".annoy":
commands.annoy(self, invkr_id, parameter)
elif command == ".annoy" and permission_level > 0:
commands.annoy(self, invoker, parameter)
elif command == ".follow":
commands.follow(self, invkr_id, parameter)
elif command == ".follow" and permission_level > 0:
commands.follow(self, invoker, parameter)
elif command == ".help" or command == ".h":
commands.help(self, invkr_id)
commands.help(self, invoker)
elif command == ".info":
commands.info(self, invkr_id)
commands.info(self, invoker)
elif command == ".kickall":
elif command == ".kickall" and permission_level > 0:
self.kickall("test") # TODO
elif command == ".list":
commands.list(self, invkr_id, parameter)
elif command == ".list" and permission_level > 0:
commands.list(self, invoker, parameter)
elif command == ".mute":
commands.mute(self, invkr_id, parameter)
elif command == ".mute" and permission_level > 0:
commands.mute(self, invoker, parameter)
elif command == ".pingall":
commands.pingall(self, invkr_id, parameter)
elif command == ".pingall" and permission_level > 0:
commands.pingall(self, invoker, parameter)
elif command == ".rename":
commands.rename(self, invkr_id, parameter)
elif command == ".rename" and permission_level > 0:
commands.rename(self, invoker, parameter)
elif command == ".rmb":
commands.rmb(self, invoker, parameter)
elif command == ".roll":
commands.roll(self, invoker)
elif command == ".rwf":
commands.rwf(self)
elif command == ".roll":
pass # TODO needs permission for everyone + targemode implementation
elif command == ".stop" or command == ".quit" or command == ".q" and permission_level > 0:
commands.quit(self, invoker)
elif command == ".stop" or command == ".quit" or command == ".q":
commands.quit(self, invkr_id)
elif command == ".test":
elif command == ".test" and permission_level > 0:
cid = self.createChannel("Test")
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=cid)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=cid)
elif command == ".ticker":
commands.ticker(self, parameter)
elif command == ".unmute":
commands.unmute(self, invkr_id, parameter)
elif command == ".unmute" and permission_level > 0:
commands.unmute(self, invoker, parameter)
else:
err = f"Unknown Command: [{command}]"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
# ----------------------------------------------------------------------------------------------------------------------
@ -487,6 +507,9 @@ def main():
# Load toml config
conf = util.getConf("prod.toml")
# Change loglevel from config
util.changeLevel(log, conf['Log']['level'])
# Start the Bot Instance
TSbot(conf, log)

View File

@ -3,7 +3,7 @@ TBD
"""
__author__ = "Lukas Mahler"
__version__ = "0.0.3"
__version__ = "1.0.0"
__date__ = "14.04.2022"
__email__ = "m@hler.eu"
__status__ = "Development"
@ -13,6 +13,7 @@ __status__ = "Development"
import sys
import json
import time
import random
import inspect
# Custom
@ -23,14 +24,14 @@ import requests
from src import util
def annoy(self, invkr_id, parameter):
def annoy(self, invoker, parameter):
"""
"""
if not len(parameter) != 1:
err = "Please use the command like this: .annoy TARGET MESSAGE"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
target = parameter[0]
@ -38,14 +39,14 @@ def annoy(self, invkr_id, parameter):
self.poke(n=10, msg=msg, usr=target)
def follow(self, invkr_id, parameter):
def follow(self, invoker, parameter):
"""
TODO sticky folgen
"""
if not parameter:
err = "Please use the command like this: .follow TARGET"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
target = parameter[0]
@ -54,7 +55,7 @@ def follow(self, invkr_id, parameter):
self.bot.clientmove(clid=self.myid, cid=cid)
def help(self, invkr_id):
def help(self, invoker):
"""
"""
@ -62,27 +63,27 @@ def help(self, invkr_id):
# Find all functions in this submodule
cmds = [f[0] for f in inspect.getmembers(sys.modules[__name__], inspect.isfunction)]
msg = f"List of commands:\n---------------------\n.{f'{chr(10)}.'.join(cmds)}"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=msg)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
def info(self, invkr_id):
def info(self, invoker):
"""
"""
msg = f"Runtime: {util.getRuntime(self.started)}\n" \
f"Version: {self.version}"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=msg)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
def list(self, invkr_id, parameter):
def list(self, invoker, parameter):
"""
Message the invoker of the function either a list of channels, clients, commands or server groups.
"""
if not parameter:
err = "Please use the command like this: .list channel/clients/commands/groups/"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
what = parameter[0]
@ -105,7 +106,7 @@ def list(self, invkr_id, parameter):
msg = json.dumps(mydict)
elif what == "commands":
help(self, invkr_id)
help(self, invoker)
return
elif what == "groups":
@ -117,17 +118,17 @@ def list(self, invkr_id, parameter):
else:
msg = f"The parameter [{what}] is not supported."
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=msg)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
def mute(self, invkr_id, parameter):
def mute(self, invoker, parameter):
"""
Assign the mute group to a user.
"""
if not parameter:
err = "Please use the command like this: .mute TARGET"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
target = parameter[0]
@ -138,18 +139,18 @@ def mute(self, invkr_id, parameter):
self.bot.servergroupaddclient(sgid=self.sgid_mute, cldbid=cldbid)
except ts3.query.TS3QueryError as e:
err = f"Failed to add to group: [{e}]"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
def pingall(self, invkr_id, parameter):
def pingall(self, invoker, parameter):
"""
"""
if not parameter:
err = "Please use the command like this: .pingall MESSAGE"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
msg = parameter[0]
@ -174,27 +175,82 @@ def notifyAdmin(self):
time.sleep(1)
def quit(self, invkr_id):
def quit(self, invoker):
"""
"""
self.stop(invkr_id)
self.stop(invoker)
def rename(self, invkr_id, parameter):
def rename(self, invoker, parameter):
"""
Rename the Bot to the given Nickname
"""
if not parameter:
err = "Please use the command like this: .rename NICKNAME"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
self.nickname = parameter[0]
self.bot.clientupdate(client_nickname=self.nickname)
def rmb(self, invoker, parameter):
"""
Convert rmb (Chinese Yuan) to eur (Euro)
"""
if not parameter:
err = "Please use the command like this: .rmb AMOUNT"
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
url = "https://www.floatrates.com/daily/cny.json"
req = requests.get(url)
if req.status_code != 200:
self.pipeOut(req.status_code, lvl="WARNING")
return
data = json.loads(req.text)
if not data:
return
if "eur" not in data:
self.pipeOut(data, lvl="WARNING")
return
try:
conversion_rate = float(data['eur']['inverseRate'])
except KeyError:
conversion_rate = 6.94 # Static
if "k" in parameter[0]:
rmb_val = float(parameter[0].replace("k", "000"))
else:
rmb_val = float(parameter[0])
eur = rmb_val / conversion_rate
msg = f"[{rmb_val:.0f}] rmb is approx. [{eur:.0f}] eur / conversion rate [{conversion_rate:.2f}]"
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
def roll(self, invoker):
cid_invoker = self.bot.clientinfo(clid=int(invoker['id']))[0]["cid"]
cid_self = self.bot.clientinfo(clid=self.myid)[0]["cid"]
if cid_self != cid_invoker:
self.bot.clientmove(clid=self.myid, cid=cid_invoker)
rolled = random.randint(0, 100)
msg = f"Rolled a [{rolled:3d}] by {invoker['name']}"
self.bot.sendtextmessage(targetmode=2, target=cid_invoker, msg=msg)
def rwf(self):
"""
Refresh the WoW progress channel,
@ -204,7 +260,7 @@ def rwf(self):
req = requests.get(url)
if req.status_code != 200:
self.pipeOut(req.status_code)
self.pipeOut(req.status_code, lvl="WARNING")
return
data = json.loads(req.text)
@ -213,7 +269,7 @@ def rwf(self):
return
if "progression" not in data:
self.pipeOut(data)
self.pipeOut(data, lvl="WARNING")
return
prog_done = 0
@ -261,14 +317,14 @@ def ticker(self, parameter):
pass
def unmute(self, invkr_id, parameter):
def unmute(self, invoker, parameter):
"""
Remove the mute group to a user.
"""
if not parameter:
err = "Please use the command like this: .unmute TARGET"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
target = parameter[0]
@ -279,7 +335,7 @@ def unmute(self, invkr_id, parameter):
self.bot.servergroupdelclient(sgid=self.sgid_mute, cldbid=cldbid)
except ts3.query.TS3QueryError as e:
err = f"Failed to remove from group: [{e}]"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return

View File

@ -14,4 +14,7 @@ mute = "0"
[Misc]
nickname = "myTS3-Bot"
whitelisted = false
crypto = false
crypto = false
[Log]
level = "INFO"

View File

@ -3,8 +3,8 @@ TBD
"""
__author__ = "Lukas Mahler"
__version__ = "0.0.0"
__date__ = "13.10.2021"
__version__ = "1.0.0"
__date__ = "14.04.2022"
__email__ = "m@hler.eu"
__status__ = "Development"
@ -82,17 +82,24 @@ def setupLogger(logpath, lvl="DEBUG"):
"""
global log # Needed to log exceptions in src\util
log = logging.getLogger()
log = logging.getLogger("mylog")
if not os.path.exists(logpath):
os.makedirs(logpath)
handler = RotatingFileHandler(logpath + r"/myTS3.log", encoding='utf-8', maxBytes=1*1024*1024, backupCount=10)
logformat = logging.Formatter("%(asctime)s %(levelname)8s %(message)s", "%Y-%m-%d %H:%M:%S")
handler.setFormatter(logformat)
log.addHandler(handler)
log.setLevel(lvl)
log.setLevel(logging.getLevelName(lvl))
return log
def changeLevel(log, lvl):
"""
Change the loglevel after creation
"""
log.setLevel(logging.getLevelName(lvl))
if __name__ == "__main__":
exit()