lm/myTS3
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Compare commits

base: lm/myTS3:master
Branches Tags
lm/myTS3:master
..
compare: lm/myTS3:d7759ce7d049f395789f2a036ea1e51f25e2ff33
Branches Tags
lm/myTS3:master
2 Commits
master .. d7759ce7d0
Author SHA1 Message Date
lm d7759ce7d0 Rewrite config to use toml instead of .env 2021-09-27 12:06:09 +02:00
lm b248022eaa Rewrite config to tuse omlà instead of .env 2021-09-27 01:45:08 +02:00
8 changed files with 250 additions and 663 deletions
Expand all files Collapse all files
.gitignore
-2
View File
@@ -141,5 +141,3 @@ cython_debug/
# Pycharm
.idea/
prod.toml
log/
Pipfile
-1
View File
@@ -7,7 +7,6 @@ name = "pypi"
ts3 = "*"
pycoingecko = "*"
toml = "*"
requests = "*"
[dev-packages]
Pipfile.lock
Generated
+17 -17
View File
@@ -1,7 +1,7 @@
{
"_meta": {
"hash": {
"sha256": "31ecf8cb0f57cd41d5c094347ce8878eddd6f611790259ee671f31d6620a1bee"
"sha256": "d9f13c4341b606a12b8b0ef95f6bce67911d3f7866a6c4552c342c4a5ae15fdf"
},
"pipfile-spec": 6,
"requires": {
@@ -18,26 +18,26 @@
"default": {
"certifi": {
"hashes": [
"sha256:78884e7c1d4b00ce3cea67b44566851c4343c120abd683433ce934a68ea58872",
"sha256:d62a0163eb4c2344ac042ab2bdf75399a71a2d8c7d47eac2e2ee91b9d6339569"
"sha256:2bbf76fd432960138b3ef6dda3dde0544f27cbf8546c458e60baf371917ba9ee",
"sha256:50b1e4f8446b06f41be7dd6338db18e0990601dce795c2b1686458aa7e8fa7d8"
],
"version": "==2021.10.8"
"version": "==2021.5.30"
},
"charset-normalizer": {
"hashes": [
"sha256:2857e29ff0d34db842cd7ca3230549d1a697f96ee6d3fb071cfa6c7393832597",
"sha256:6881edbebdb17b39b4eaaa821b438bf6eddffb4468cf344f09f89def34a8b1df"
"sha256:5d209c0a931f215cee683b6445e2d77677e7e75e159f78def0db09d68fafcaa6",
"sha256:5ec46d183433dcbd0ab716f2d7f29d8dee50505b3fdb40c6b985c7c4f5a3591f"
],
"markers": "python_version >= '3'",
"version": "==2.0.12"
"version": "==2.0.6"
},
"idna": {
"hashes": [
"sha256:84d9dd047ffa80596e0f246e2eab0b391788b0503584e8945f2368256d2735ff",
"sha256:9d643ff0a55b762d5cdb124b8eaa99c66322e2157b69160bc32796e824360e6d"
"sha256:14475042e284991034cb48e06f6851428fb14c4dc953acd9be9a5e95c7b6dd7a",
"sha256:467fbad99067910785144ce333826c71fb0e63a425657295239737f7ecd125f3"
],
"markers": "python_version >= '3'",
"version": "==3.3"
"version": "==3.2"
},
"pycoingecko": {
"hashes": [
@@ -49,11 +49,11 @@
},
"requests": {
"hashes": [
"sha256:68d7c56fd5a8999887728ef304a6d12edc7be74f1cfa47714fc8b414525c9a61",
"sha256:f22fa1e554c9ddfd16e6e41ac79759e17be9e492b3587efa038054674760e72d"
"sha256:6c1246513ecd5ecd4528a0906f910e8f0f9c6b8ec72030dc9fd154dc1a6efd24",
"sha256:b8aa58f8cf793ffd8782d3d8cb19e66ef36f7aba4353eec859e74678b01b07a7"
],
"index": "pypi",
"version": "==2.27.1"
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4, 3.5'",
"version": "==2.26.0"
},
"toml": {
"hashes": [
@@ -73,11 +73,11 @@
},
"urllib3": {
"hashes": [
"sha256:44ece4d53fb1706f667c9bd1c648f5469a2ec925fcf3a776667042d645472c14",
"sha256:aabaf16477806a5e1dd19aa41f8c2b7950dd3c746362d7e3223dbe6de6ac448e"
"sha256:4987c65554f7a2dbf30c18fd48778ef124af6fab771a377103da0585e2336ece",
"sha256:c4fdf4019605b6e5423637e01bc9fe4daef873709a7973e195ceba0a62bbc844"
],
"markers": "python_version >= '2.7' and python_version not in '3.0, 3.1, 3.2, 3.3, 3.4' and python_version < '4'",
"version": "==1.26.9"
"version": "==1.26.7"
}
},
"develop": {}
myTS3.py
+204 -229
View File
@@ -3,21 +3,21 @@ TBD
"""
__author__ = "Lukas Mahler"
__version__ = "1.0.0"
__date__ = "14.04.2022"
__version__ = "0.0.0"
__date__ = "27.09.2021"
__email__ = "m@hler.eu"
__status__ = "Development"
# Default
import sys
import time
import os.path
import json
# Custom
import ts3
# Self
from src import util, gecko, commands
from src import util, gecko
class TSbot:
@@ -28,45 +28,35 @@ class TSbot:
start the event loop function.
"""
# Initialize from config
self.host = conf["Connection"]["host"]
self.port = conf["Connection"]["port"]
self.sid = conf["Connection"]["sid"]
self.user = conf["Authentication"]["user"]
self.pwd = conf["Authentication"]["pwd"]
self.sgid_admin = conf["Groups"]["admins"]
self.sgid_mute = conf["Groups"]["mute"]
self.crypto = conf["Misc"]["crypto"]
self.nickname = conf["Misc"]["nickname"]
self.whitelisted = conf["Misc"]["whitelisted"]
# Initialize self
self.gecko = gecko.GeckoAPI()
self.log = log
self.myid = None
self.running = True
self.version = __version__
self.started = time.time()
self.gecko = gecko.GeckoAPI()
self.crypto_update = self.started - 1800
self.intro = "<Keep this chat open to use commands>"
self.last_crypto_update = time.time() - 1800
self.pipeOut(f"Trying to connect to: {self.host}:{self.port}")
# Starting the Connection
with ts3.query.TS3Connection(self.host, self.port) as self.bot:
try:
self.bot.login(client_login_name=self.user, client_login_password=self.pwd)
self.bot.use(sid=self.sid)
except ts3.query.TS3QueryError:
self.pipeOut("Invalid login credentials, please check your '.toml' file.", lvl="CRITICAL")
try:
self.bot.clientupdate(client_nickname=self.nickname)
except ts3.query.TS3QueryError as e:
self.pipeOut(e, lvl="WARNING")
except ts3.query.TS3QueryError:
pass
self.pipeOut(f"Successfully connected as: {self.nickname}")
# Start the Bot loop
# Start the Bot
self.loop()
def loop(self):
@@ -78,7 +68,7 @@ class TSbot:
The loop can be stopped setting self.running to False
"""
# Find the instances client id
# Find my client id
me = self.bot.clientfind(pattern=self.nickname)
if len(me) == 1:
self.myid = me[0]["clid"]
@@ -86,7 +76,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 default channel, you can do: '''
''' if you want to move the Bot to a certain channel (instead of the defualt channel, you can do: '''
# self.bot.clientmove(clid=self.myid,cid=129)
# ----------- SUBSCRIBE TO EVENTS -------------
@@ -109,26 +99,20 @@ class TSbot:
time.sleep(5)
# Notify connected admins
commands.notifyAdmin(self)
self.notifyAdmin()
# ----------- LOOP HERE -------------
while self.running:
# self.bot.send_keepalive()
self.pipeOut(f"Waiting for a new Event...", lvl="debug")
self.pipeOut(f"Waiting for a new Event...")
self.bot.version()
# Auto-update crypto price channels every 30 minutes
if self.crypto:
if self.crypto_update + 1800 < time.time():
self.crypto_update = time.time()
if self.last_crypto_update + 1800 < time.time():
self.last_crypto_update = time.time()
commands.ticker(self, "")
# Hijacked for WoW
commands.rwf(self)
else:
pass
self.lookupcommand(".btc", self.myid)
self.lookupcommand(".dot", self.myid)
try:
# This method blocks, but we must sent the keepalive message at
@@ -138,8 +122,6 @@ class TSbot:
except ts3.query.TS3TimeoutError:
pass
# try else!
else:
event_type = event.event
self.pipeOut(f"Got Event | length={len(event[0])} | {event_type}")
@@ -149,22 +131,17 @@ class TSbot:
if 'client_nickname' in event[0]:
displayname = event[0]['client_nickname']
elif 'clid' in event[0]:
displayname = event[0]['clid']
else:
self.pipeOut(event[0])
# displayname = "Unresolved"
continue # can't resolve no clid
displayname = event[0]['clid']
self.pipeOut(f"Client [{displayname}] connected.")
# Check if the connector is a ServerQuery or not
if not self.isqueryclient(event[0]["clid"]):
if not self.isqueryclient(event[0]["client_unique_identifier"]):
self.pipeOut(f"* {event[0]}", lvl="debug")
# Check if the connector is an Admin
if self.isadmin(event[0]["client_database_id"]):
msg = "<Keep this chat open to use commands>"
self.bot.sendtextmessage(targetmode=1, target=event[0]["clid"], msg=msg)
self.bot.sendtextmessage(targetmode=1, target=event[0]["clid"], msg=self.intro)
else:
pass
else:
@@ -182,30 +159,17 @@ class TSbot:
# New text message
elif event_type == "notifytextmessage":
msg = event[0]["msg"].replace("\n", " ")
invkr_name = event[0]["invokername"]
msg = event[0]["msg"]
invkr = event[0]["invokername"]
invkr_id = event[0]["invokerid"]
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}"')
self.pipeOut(f'Message | From: "{invkr}" | Content: "{msg}"')
if msg.startswith("."):
self.lookupcommand(msg, invoker, permission_level)
self.lookupcommand(msg, invkr_id)
# Unknown Event
else:
self.pipeOut(f"Unknown Event: {event.__dict__}", lvl="warning")
# Class Utility --------------------------------------------------------------------------------------------------------
def pipeOut(self, msg, lvl="INFO", reprint=True):
"""
All output pipes through this function.
@@ -213,30 +177,38 @@ class TSbot:
or run in silent log mode. [set reprint to False]
"""
lvl = lvl.upper()
lvln = int(getattr(util.logging, lvl))
lvln = int(getattr(util.logging, lvl.upper()))
self.log.log(lvln, msg)
if reprint and lvln >= self.log.getEffectiveLevel():
print(f"[{time.strftime('%H:%M:%S')}]{f'[{lvl}]':10s} {msg}")
if reprint:
print(f"[{time.strftime('%H:%M:%S')}][{lvl.upper()}] {msg}")
if not self.whitelisted:
if msg == "error id 524: client is flooding":
time.sleep(30)
if lvl == "CRITICAL":
exit(1)
def stop(self, invoker):
def stop(self, invkr_id):
"""
This stops the bot instance by invalidating the event loop.
"""
msg = "Shutdown, bye bye!"
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=msg)
self.pipeOut(msg)
self.running = False
def notifyAdmin(self):
"""
Ping every available admin.
"""
clients = self.bot.clientlist()
clients = [client for client in clients if client["client_type"] != "1"]
for client in clients:
cldbid = client["client_database_id"]
clid = client["clid"]
if self.isadmin(cldbid):
self.bot.sendtextmessage(targetmode=1, target=clid, msg=self.intro)
if not self.whitelisted:
time.sleep(1)
def kickall(self, msg):
"""
TODO
@@ -245,10 +217,13 @@ class TSbot:
clients = self.bot.clientlist()
clients = [client["clid"] for client in clients if client["client_type"] != "1"]
for clid in clients:
try:
self.bot.clientpoke(msg=msg, clid=clid)
# TODO
except:
pass
def poke(self, msg=None, n=1, delay=0.2, usr='all'):
def poke(self, msg=None, n=10, delay=0.2, usr='all'):
"""
ping a single or multiple users for n times including a msg.
"""
@@ -261,20 +236,15 @@ class TSbot:
clients = self.bot.clientlist()
clients = [client["clid"] for client in clients if client["client_type"] != "1"]
else:
try:
clients = self.bot.clientfind(pattern=usr)
clients = [client["clid"] for client in clients]
except ts3.query.TS3QueryError as e:
self.pipeOut(f"Couldnt execute poke, no client found using pattern {usr},"
f"returned error:\n{e}", lvl="ERROR")
return
# Ping them
if len(clients) > 0:
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:
@@ -282,7 +252,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
@@ -318,197 +288,202 @@ class TSbot:
except Exception as e:
self.pipeOut(e, lvl="error")
def isqueryclient(self, clid):
def list(self, what, invkr_id):
"""
Message the invoker of the function either a list of channels or a list of clients.
"""
mydict = {}
if what == "channel":
channels = self.bot.channellist()
for channel in channels:
order = channel["channel_order"]
mydict[order] = [channel["channel_name"], channel["cid"]]
mydict = dict(sorted(mydict.items()))
msg = json.dumps(mydict)
elif what == "clients":
clients = self.bot.clientlist()
for client in clients:
mydict[client["client_nickname"]] = {"clid": client["clid"], "cldbid": client["client_database_id"]}
mydict = dict(sorted(mydict.items()))
msg = json.dumps(mydict)
else:
msg = None
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=msg)
def isqueryclient(self, cluid):
"""
Check if the given client-uid is a query client.
"""
try:
client = self.bot.clientinfo(clid=clid)
except ts3.query.TS3QueryError as e:
self.pipeOut(f"ISQUERY: given clid {clid} returned: {e}", lvl="WARNING")
return True
try:
if client[0]["client_type"] == "1":
self.pipeOut(f"[{clid}] ISQUERY: True")
# client = self.bot.clientlist(uid=uid)
# print(client[0])
# if client[0]["client_type"] == "1":
if cluid == "ServerQuery":
self.pipeOut(f"[{cluid}] ISQUERY: True")
return True
else:
self.pipeOut(f"[{clid}] ISQUERY: False")
self.pipeOut(f"[{cluid}] ISQUERY: False")
return False
except KeyError as e:
self.pipeOut(f"Error [{e}] on client [{client.__dict__}] ")
return True # Da wahrscheinlich ein Query Client
def isadmin(self, cldbid):
"""
Check if the given client-databaseid is an admin.
This is done by resolving the clients groups and check
if any of the groups sgid match a sgid from the sgids from the '.toml' config.
"""
try:
groups = self.bot.servergroupsbyclientid(cldbid=cldbid)
except ts3.query.TS3QueryError as e:
self.pipeOut(e, lvl="ERROR")
return False
self.pipeOut(str(groups.__dict__), lvl="DEBUG") # DEBUG
# [print(group["sgid"]) for group in groups]
for group in groups:
if any(sgid == group['sgid'] for sgid in self.sgid_admin):
# 6 Server Admin/ 13 Operator / 15 Root
# if (group["sgid"] == "6") or (group["sgid"] == "13") or (group["sgid"] == "15"):
if group["sgid"] == "15":
self.pipeOut(f"[{cldbid}] ISADMIN: True")
return True
else:
continue
self.pipeOut(f"[{cldbid}] ISADMIN: False")
return False
def lookupcommand(self, msg, invkr_id):
"""
Every message starting with '.' gets passed and evaluated in this function.
Command Parameter 1 Parameter 2
---------------------------------------------------------
.annoy target message
.kickall message
.test
.btc / .eth
.dot / .ada
.info clid
.list channel/clients
.follow
.rename nickname
.stop / .quit / .q
.pingall message
.admin
"""
commandstring = msg.split(" ")
command = commandstring[0]
parameter = commandstring[1:]
self.pipeOut(f"command: {command} | parameter: {parameter} | invkr_id: {invkr_id}")
if command == ".annoy":
try:
target = parameter[0]
msg = parameter[1]
self.poke(msg=msg, usr=target)
except IndexError:
err = "Please use the command like this: .annoy TARGET MESSAGE"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
pass
elif command == ".kickall":
self.kickall("test") # TODO
elif command == ".test":
cid = self.createChannel("Test")
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=cid)
elif command == ".btc" or command == ".eth":
channelname = f"[cspacerC1]BTC: {self.gecko.getSymbol('Bitcoin', decimal=0)} | " \
f"ETH: {self.gecko.getSymbol('ethereum', decimal=0)}"
try:
self.editChannelname(200, channelname)
except ts3.query.TS3QueryError:
pass
elif command == ".dot" or command == ".ada":
channelname = f"[cspacerC2]DOT: {self.gecko.getSymbol('polkadot', decimal=2)} | " \
f"ADA: {self.gecko.getSymbol('cardano', decimal=2)}"
try:
self.editChannelname(201, channelname)
except ts3.query.TS3QueryError:
pass
elif command == ".info":
pass # TODO
elif command == ".list":
try:
self.list(parameter[0], invkr_id)
except IndexError:
err = "Please use the command like this: .list channel/clients"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
pass
elif command == ".follow":
pass # TODO
elif command == ".rename":
try:
self.nickname = parameter[0]
self.bot.clientupdate(client_nickname=self.nickname)
except IndexError:
err = "Please use the command like this: .rename NAME"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
pass
elif command == ".stop" or command == ".quit" or command == ".q":
self.stop(invkr_id)
elif command == ".pingall":
try:
msg = parameter[0]
self.poke(msg=msg)
except IndexError:
err = "Please use the command like this: .pingall MESSAGE"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
pass
elif command == ".admin":
self.notifyAdmin()
else:
err = f"Unknown Command: [{command}]"
self.bot.sendtextmessage(targetmode=1, target=invkr_id, msg=err)
def printable_clientinfo(self, clid):
"""
Generate printable clientinfo from clid.
"""
usrd = {}
info = self.bot.clientinfo(clid=clid)
usrd = info.__dict__
return usrd
def clid2cldbid(self, clid):
def checkgrp(self):
"""
Convert a given clid to a cldbid.
Print all assigned groups for every connected client.
"""
cluid = self.bot.clientgetuidfromclid(clid=clid)[0]["cluid"]
cldbid = self.bot.clientgetdbidfromuid(cluid=cluid)[0]["cldbid"]
# clients = self.bot.clientlist()
# clients_cldbid = [client["client_database_id"] for client in clients if client["client_type"] != "1"]
clients = self.bot.clientlist(groups=True)
clients_groups = [client["client_servergroups"] for client in clients if client["client_type"] != "1"]
return cldbid
def identifytarget(self, target):
"""
Try to find clid from nickname.
"""
try:
clients = self.bot.clientfind(pattern=target)
clids = [client["clid"] for client in clients]
if len(clids) == 1:
return clids[0]
else:
self.pipeOut(f"Found multiple matching clients using pattern {target}:{clids}")
return
except ts3.query.TS3QueryError as e:
self.pipeOut(f"No client found using pattern [{target}], "
f"returned error:\n{e}", lvl="ERROR")
return
# Commands -------------------------------------------------------------------------------------------------------------
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.
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} | "
f"from: {invoker['name']} | permissionlevel: {permission_level}")
if command == ".admin" or command == ".notifyAdmin":
commands.notifyAdmin(self)
elif command == ".annoy" and permission_level > 0:
commands.annoy(self, invoker, parameter)
elif command == ".follow" and permission_level > 0:
commands.follow(self, invoker, parameter)
elif command == ".help" or command == ".h":
commands.help(self, invoker)
elif command == ".info":
commands.info(self, invoker)
elif command == ".kickall" and permission_level > 0:
self.kickall("test") # TODO
elif command == ".list" and permission_level > 0:
commands.list(self, invoker, parameter)
elif command == ".mute" and permission_level > 0:
commands.mute(self, invoker, parameter)
elif command == ".pingall" and permission_level > 0:
commands.pingall(self, invoker, 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 == ".stop" or command == ".quit" or command == ".q" and permission_level > 0:
commands.quit(self, invoker)
elif command == ".test" and permission_level > 0:
cid = self.createChannel("Test")
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=cid)
elif command == ".ticker":
commands.ticker(self, 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=invoker['id'], msg=err)
self.pipeOut(f"{clients_groups}")
# ----------------------------------------------------------------------------------------------------------------------
def main():
# Start logger
logpath = os.path.dirname(os.path.abspath(__file__)) + r"/log"
log = util.setupLogger(logpath)
log = util.setupLogger()
# Log unhandled exception
sys.excepthook = util.unhandledException
# Load toml config
conf = util.getConf("prod.toml")
# Change loglevel from config
util.changeLevel(log, conf['Log']['level'])
conf = util.getconf("prod.toml")
# Start the Bot Instance
TSbot(conf, log)
src/commands.py
-343
View File
@@ -1,343 +0,0 @@
"""
TBD
"""
__author__ = "Lukas Mahler"
__version__ = "1.0.0"
__date__ = "14.04.2022"
__email__ = "m@hler.eu"
__status__ = "Development"
# Default
import sys
import json
import time
import random
import inspect
# Custom
import ts3
import requests
# Self
from src import util
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=invoker['id'], msg=err)
return
target = parameter[0]
msg = parameter[1]
self.poke(n=10, msg=msg, usr=target)
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=invoker['id'], msg=err)
return
target = parameter[0]
clid = self.identifytarget(target)
cid = self.bot.clientinfo(clid=int(clid))[0]["cid"]
self.bot.clientmove(clid=self.myid, cid=cid)
def help(self, invoker):
"""
"""
# 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=invoker['id'], msg=msg)
def info(self, invoker):
"""
"""
msg = f"Runtime: {util.getRuntime(self.started)}\n" \
f"Version: {self.version}"
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
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=invoker['id'], msg=err)
return
what = parameter[0]
mydict = {}
if what == "channel":
channels = self.bot.channellist()
for channel in channels:
order = channel["channel_order"]
mydict[order] = [channel["channel_name"], channel["cid"]]
mydict = dict(sorted(mydict.items()))
msg = json.dumps(mydict)
elif what == "clients":
clients = self.bot.clientlist()
for client in clients:
mydict[client["client_nickname"]] = {"clid": client["clid"], "cldbid": client["client_database_id"]}
mydict = dict(sorted(mydict.items()))
msg = json.dumps(mydict)
elif what == "commands":
help(self, invoker)
return
elif what == "groups":
groups = self.bot.servergrouplist()
for group in groups:
mydict[group["name"]] = {"sgid": group["sgid"], "sortid": group["sortid"]}
msg = json.dumps(mydict)
else:
msg = f"The parameter [{what}] is not supported."
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=msg)
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=invoker['id'], msg=err)
return
target = parameter[0]
clid = self.identifytarget(target)
cldbid = self.clid2cldbid(clid)
try:
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=invoker['id'], msg=err)
return
def pingall(self, invoker, parameter):
"""
"""
if not parameter:
err = "Please use the command like this: .pingall MESSAGE"
self.bot.sendtextmessage(targetmode=1, target=invoker['id'], msg=err)
return
msg = parameter[0]
self.poke(msg=msg)
def notifyAdmin(self):
"""
Message every available admin.
"""
clients = self.bot.clientlist()
clients = [client for client in clients if client["client_type"] != "1"]
for client in clients:
cldbid = client["client_database_id"]
clid = client["clid"]
if self.isadmin(cldbid):
msg = "<Keep this chat open to use commands>"
self.bot.sendtextmessage(targetmode=1, target=clid, msg=msg)
if not self.whitelisted:
time.sleep(1)
def quit(self, invoker):
"""
"""
self.stop(invoker)
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=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,
this is very specific to my use, you can disable this in the config.
"""
url = "https://raider.io/api/v1/raiding/progression?raid=sepulcher-of-the-first-ones&difficulty=mythic&region=world"
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 "progression" not in data:
self.pipeOut(data, lvl="WARNING")
return
prog_done = 0
prog_max = len(data['progression'])
for i in range(0, prog_max):
if data['progression'][i]['totalGuilds'] > 0:
prog_done = i
guilds_done = data['progression'][prog_done]['totalGuilds']
channelname = f"[cspacerC3]Sepulcher {prog_done+1}/{prog_max} ({guilds_done})"
try:
self.editChannelname(205, channelname)
except ts3.query.TS3QueryError:
pass
def ticker(self, parameter):
"""
Refresh the crypto Ticker channels,
this is very specific to my use, you can disable this in the config.
"""
if parameter:
symbol = parameter[0].lower()
else:
symbol = "all"
if symbol in ["btc", "bitcoin", "eth", "ethereum", "all"]:
channelname = f"[cspacerC1]BTC: {self.gecko.getSymbol('bitcoin', decimal=0)} | " \
f"ETH: {self.gecko.getSymbol('ethereum', decimal=0)}"
try:
self.editChannelname(200, channelname)
except ts3.query.TS3QueryError:
pass
if symbol in ["dot", "polkadot", "ada", "cardano", "all"]:
channelname = f"[cspacerC2]DOT: {self.gecko.getSymbol('polkadot', decimal=2)} | " \
f"ADA: {self.gecko.getSymbol('cardano', decimal=2)}"
try:
self.editChannelname(201, channelname)
except ts3.query.TS3QueryError:
pass
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=invoker['id'], msg=err)
return
target = parameter[0]
clid = self.identifytarget(target)
cldbid = self.clid2cldbid(clid)
try:
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=invoker['id'], msg=err)
return
if __name__ == "__main__":
exit()
src/gecko.py
+2 -8
View File
@@ -3,13 +3,12 @@ TBD
"""
__author__ = "Lukas Mahler"
__version__ = "0.0.1"
__date__ = "04.02.2022"
__version__ = "0.0.0"
__date__ = "27.09.2021"
__email__ = "m@hler.eu"
__status__ = "Development"
# Imports
from datetime import datetime
from pycoingecko import CoinGeckoAPI
@@ -45,12 +44,7 @@ class GeckoAPI:
matching = [x for x in self.coinlist if x["id"] == symbol_id.lower()]
if len(matching) == 1:
try:
resp_dict = self.api.get_price(ids=matching[0]["id"], vs_currencies=curr.lower())
except Exception as e:
with open("gecko_error.log", 'a+') as log:
log.write(f"{datetime.now():%Y-%m-%d %H:%M} | {e}\n")
return "Error"
price = resp_dict[matching[0]["id"]][curr.lower()]
if isinstance(price, int):
price = float(price) # Convert to always have floats
src/template.toml
+1 -9
View File
@@ -1,5 +1,5 @@
[Connection]
host = "127.0.0.1"
host = "example.com"
port = 10011
sid = 1
@@ -7,14 +7,6 @@ sid = 1
user = "exampleuser"
pwd = "password"
[Groups]
admins = ["1"]
mute = "0"
[Misc]
nickname = "myTS3-Bot"
whitelisted = false
crypto = false
[Log]
level = "INFO"
src/util.py
+15 -43
View File
@@ -3,8 +3,8 @@ TBD
"""
__author__ = "Lukas Mahler"
__version__ = "1.0.0"
__date__ = "14.04.2022"
__version__ = "0.0.0"
__date__ = "27.09.2021"
__email__ = "m@hler.eu"
__status__ = "Development"
@@ -14,9 +14,7 @@ import sys
import shutil
import os.path
import logging
import time
from datetime import timedelta
from logging.handlers import RotatingFileHandler
from datetime import date
# Custom
import toml
@@ -34,18 +32,18 @@ def unhandledException(exc_type, exc_value, exc_traceback):
log.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
def getConf(fname):
def getconf(fname):
"""
"""
if fname.endswith(".toml"):
if os.path.isfile(fname):
try:
config = toml.load(fname)
checkConf(config)
return config
data = toml.load(fname)
return data
except ValueError as e:
log.critical(f"The provided '.toml' is probably invalid, returned error:\n{e}")
log.critical(e)
# log.critical("The provided '.toml' seems to be invalid.")
exit(1)
else:
log.critical(f"Couldn't locate the '.toml' file [{fname}].")
@@ -53,53 +51,27 @@ def getConf(fname):
shutil.copy("src/template.toml", fname)
exit(1)
else:
log.critical(f"The provided config file [{fname}] is not a '.toml' file.")
log.critical("The provided config file is not a '.toml' file.")
log.info("Creating a new '.toml' file from template, please edit and restart.")
shutil.copy("src/template.toml", "prod.toml")
exit(1)
def getRuntime(started):
def setupLogger(lvl="DEBUG"):
"""
"""
elapsed = time.time() - started
runtime = str(timedelta(seconds=elapsed)).split(".")[0] # ignore Microseconds
return runtime
def checkConf(config):
"""
TODO check if keys exist
"""
pass
def setupLogger(logpath, lvl="DEBUG"):
"""
Create a rotating log in a log folder
"""
global log # Needed to log exceptions in src\util
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")
log = logging.getLogger()
now = date.today().strftime("%Y-%m-%d")
handler = logging.FileHandler(f"myTS3_{now}.log", encoding="utf-8")
logformat = logging.Formatter("%(asctime)s %(levelname)7s %(message)s", "%d-%m-%Y %H:%M:%S")
handler.setFormatter(logformat)
log.addHandler(handler)
log.setLevel(logging.getLevelName(lvl))
log.setLevel(lvl)
return log
def changeLevel(log, lvl):
"""
Change the loglevel after creation
"""
log.setLevel(logging.getLevelName(lvl))
if __name__ == "__main__":
exit()