alex_orid/Praxis_Bot
2
0
Fork 0
Code Issues 1 Pull Requests Releases Wiki Activity

logging

Browse Source
This commit is contained in:
Alex Orid 2021-04-27 18:47:51 -04:00
parent 87492e625a
commit b106eaa9dd
29 changed files with 377 additions and 254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
Command_Management_Module.py
View File

@ -9,25 +9,31 @@ from commands.command_base import AbstractCommand
import credentials
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Command_Management_Module():
def __init__(self):
super().__init__()
self.dbCredential: credentials.DB_Credential
def main_test(self):
print("[TEST Module]> test")
praxis_logger_obj.log("[TEST Module]> test")
tempModule = user_module.User_Module()
#tempModule.commands = command_loader.load_commands_new(AbstractCommand.CommandType.Praxis)
print(self.getCommandsList(tempModule.commands))
praxis_logger_obj.log(self.getCommandsList(tempModule.commands))
def getCommandsList(self, targetModuleCommands):
print(type(targetModuleCommands))
praxis_logger_obj.log(type(targetModuleCommands))
commandsList = "\n"
for cmd in targetModuleCommands:
targetCommand = targetModuleCommands[cmd]
print(targetCommand.command)
print(targetCommand.isCommandEnabled)
praxis_logger_obj.log(targetCommand.command)
praxis_logger_obj.log(targetCommand.isCommandEnabled)
return commandsList

16
channel_rewards/channelRewards_loader.py
View File

@ -1,7 +1,13 @@
import importlib
import importlib.util
import inspect
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
import sys
from typing import Dict
@ -10,7 +16,7 @@ from channel_rewards.channelRewards_base import AbstractChannelRewards
#New
def load_rewards(channelRewardsType: AbstractChannelRewards.ChannelRewardsType) -> Dict[str, AbstractChannelRewards]:
print(" -Loading ", channelRewardsType ," ChannelRewards...\n")
praxis_logger_obj.log(" -Loading ", channelRewardsType ," ChannelRewards...\n")
ChannelRewards = compile_and_load(channelRewardsType)
return ChannelRewards
@ -26,10 +32,10 @@ def compile_and_load_file(path: str, channelRewardsType: AbstractChannelRewards.
if inspect.isclass(obj) and name.startswith("ChannelReward"):
ChannelReward_inst = obj()
if channelRewardsType == ChannelReward_inst.get_ChannelRewardType():
print(" ---Successfully loaded %s: %s" % (channelRewardsType, ChannelReward_inst.get_ChannelRewardName()))
praxis_logger_obj.log(" ---Successfully loaded %s: %s" % (channelRewardsType, ChannelReward_inst.get_ChannelRewardName()))
return ChannelReward_inst.get_ChannelRewardName(), ChannelReward_inst
elif channelRewardsType != ChannelReward_inst.get_ChannelRewardType():
print(" -%s ChannelRewardsType did not match: %s for: %s" % (ChannelReward_inst.get_ChannelRewardType(), channelRewardsType, ChannelReward_inst.get_ChannelRewardName()))
praxis_logger_obj.log(" -%s ChannelRewardsType did not match: %s for: %s" % (ChannelReward_inst.get_ChannelRewardType(), channelRewardsType, ChannelReward_inst.get_ChannelRewardName()))
return "", None
@ -40,7 +46,7 @@ def compile_and_load(ChannelRewardType: AbstractChannelRewards.ChannelRewardsTyp
for dirName, subdirList, fileList in os.walk(implementations):
for file in fileList:
name = os.path.join(dirName, file)
print("compiling: %s" % name)
praxis_logger_obj.log("compiling: %s" % name)
name, reward = compile_and_load_file(name, ChannelRewardType)
if reward is not None and reward.ChannelRewardType is ChannelRewardType:
dic[name] = reward
@ -56,7 +62,7 @@ def get_base_dir() -> str:
elif current == 'Praxis_Bot' or current == 'Praxis':
return check_dir(os.path.join(cwd, "channel_rewards"))
else:
print("could not find working directory for Praxis_Bot/channel_rewards")
praxis_logger_obj.log("could not find working directory for Praxis_Bot/channel_rewards")
raise Exception

4
channel_rewards/implemented/ChannelReward_Hydration.py
View File

@ -18,8 +18,6 @@ class ChannelReward_Hydration_v2(AbstractChannelRewards, metaclass=ABCMeta):
self.isChannelRewardEnabled = True
def do_ChannelReward(self, source = AbstractChannelRewards.ChannelRewardsSource.default, user = "User", rewardName = "", rewardPrompt = "", userInput = "", bonusData = None):
#print("sending:",user, 16, "!lights hydration")
self.dothething(user, 16, "!lights hydration", "")
return None
@ -30,7 +28,7 @@ class ChannelReward_Hydration_v2(AbstractChannelRewards, metaclass=ABCMeta):
url = "http://standalone_lights:42069/api/v1/exec_lights?%s" % params
resp = requests.get(url)
if resp.status_code == 200:
print("Got the following message: %s" % resp.text)
praxis_logger_obj.log("Got the following message: %s" % resp.text)
data = loads(resp.text)
msg = data['message']
if msg is not None:

10
chyron_module.py
View File

@ -1,6 +1,10 @@
import config
import utilities_script as utilities
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Chyron_Module():
def __init__(self):
@ -77,7 +81,7 @@ class Chyron_Module():
file = open(real_file_path, "rb")
text = file.read()
#print(text)
#praxis_logger_obj.log(text)
file.close
return text
@ -94,7 +98,7 @@ class ChyronItem():
self.itemComputedString = ""
def setupItem(self, name, title, content):
print("\nSetting up Item {", name,"}[", title, content, "]")
praxis_logger_obj.log("\nSetting up Item {", name,"}[", title, content, "]")
self.itemName = name
self.itemTitle = title
self.itemContent = content
@ -114,6 +118,6 @@ if __name__ == "__main__":
testModule.chyron_stringUpdater()
test = testModule.chyron_computedString + "<<<|"
print(test)
praxis_logger_obj.log(test)
testModule.updateChyronFile()

74
commands/implemented/Command_lights_v2.py
View File

@ -1,9 +1,10 @@
from abc import ABCMeta
import lights_module
from commands.command_base import AbstractCommand
import utilities_script as utility
from json import loads
from urllib.parse import urlencode
import requests
class Command_lights_v2(AbstractCommand, metaclass=ABCMeta):
"""
@ -18,58 +19,27 @@ class Command_lights_v2(AbstractCommand, metaclass=ABCMeta):
self.isCommandEnabled = True
def do_command(self, source = AbstractCommand.CommandSource.default, user = "User", command = "", rest = "", bonusData = None):
returnString = ""
returnString = self.dothething(user, 16, command, rest)
praxis_logger_obj.log(returnString)
tempBool = True
if tempBool == True:
LightModule = lights_module.Lights_Module()
LightModule.main()
#bot.return_message("\nRGB Command Detected!")
tempFix = command + " " + rest
return None
tempParsedMessage = tempFix.split(" ")
sceneCommand = False
if (len(tempParsedMessage)) > 2:
#bot.return_message("RGB Command!")
rgb_r = float(tempParsedMessage[1])
rgb_g = float(tempParsedMessage[2])
rgb_b = float(tempParsedMessage[3])
xy_result = LightModule.rgb_to_xy(rgb_r, rgb_g, rgb_b)
#bot.return_message("got XY")
LightModule.bridge_.set_group(16, "xy", xy_result)
#bot.return_message("sent color to [Lights_Module]")
else:
if "stream" in tempParsedMessage:
sceneCommand = True
LightModule.bridge_.run_scene("Downstairs", "Stream")
elif "normal" in tempParsedMessage:
sceneCommand = True
LightModule.bridge_.run_scene("Downstairs", "Bright")
elif "haxor" in tempParsedMessage:
sceneCommand = True
LightModule.bridge_.run_scene("Downstairs", "hacker vibes")
elif "off" in tempParsedMessage:
sceneCommand = True
LightModule.bridge_.set_group("Downstairs", "on", False)
elif "on" in tempParsedMessage:
sceneCommand = True
LightModule.bridge_.set_group("Downstairs", "on", True)
elif "ravemode" in tempParsedMessage:
sceneCommand = True
LightModule.raveMode()
else:
#bot.return_message("Color Command!")
xy_result = LightModule.color_string_parser(tempParsedMessage)
#bot.return_message("got XY")
LightModule.bridge_.set_group(16, "xy", xy_result)
#bot.return_message("sent color to [Lights_Module]")
#if sceneCommand == True:
#bot.return_message("Scene Command!")
returnString = user + " changed the light's color!"
return returnString
def dothething(self, username, light_group, command, rest):
# todo need to url-escape command and rest
params = urlencode({'user_name': username, 'light_group': light_group, 'command': command, 'rest':rest})
#standalone_lights
url = "http://standalone_lights:42069/api/v1/exec_lights?%s" % params
resp = requests.get(url)
if resp.status_code == 200:
praxis_logger_obj.log("Got the following message: %s" % resp.text)
data = loads(resp.text)
msg = data['message']
if msg is not None:
return msg
# todo send to logger and other relevent services
else:
# todo handle failed requests
pass
def get_help(self):
return self.help

4
commands/implemented/Command_roll_v2.py
View File

@ -47,7 +47,7 @@ class Command_roll_v2(AbstractCommand, metaclass=ABCMeta):
loopBool = False
if roll_type == 1:
print("-rolling...")
praxis_logger_obj.log("-rolling...")
# If roll is in xdx+x format
if loopBool == True:
rolls: list = []
@ -78,7 +78,7 @@ class Command_roll_v2(AbstractCommand, metaclass=ABCMeta):
if roll_type == 2:
print("-fate Rolling....")
praxis_logger_obj.log("-fate Rolling....")
# !roll 4df
# If roll is in xdx+x format
if loopBool == True:

2
commands/implemented/Command_test_v2.py
View File

@ -18,7 +18,7 @@ class Command_test_v2(AbstractCommand, metaclass=ABCMeta):
def do_command(self, source = AbstractCommand.CommandSource.default, user = "User", command = "", rest = "", bonusData = None):
returnString = user + " sent: [ " + command + " ] with: " + rest
#print(returnString)
#praxis_logger_obj.log(returnString)
return returnString
def get_help(self):

16
commands/loader.py
View File

@ -1,7 +1,13 @@
import importlib
import importlib.util
import inspect
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
import sys
from typing import Dict
@ -10,7 +16,7 @@ from commands.command_base import AbstractCommand
#New
def load_commands(commandType: AbstractCommand.CommandType) -> Dict[str, AbstractCommand]:
print(" -Loading ", commandType ," Commands...\n")
praxis_logger_obj.log(" -Loading ", commandType ," Commands...\n")
commands = compile_and_load(commandType)
return commands
@ -26,10 +32,10 @@ def compile_and_load_file(path: str, commandType: AbstractCommand.CommandType):
if inspect.isclass(obj) and name.startswith("Command"):
command_inst = obj()
if commandType == command_inst.get_commandType():
print(" ---Successfully loaded %s: %s" % (commandType, command_inst.get_command()))
praxis_logger_obj.log(" ---Successfully loaded %s: %s" % (commandType, command_inst.get_command()))
return command_inst.get_command(), command_inst
elif commandType != command_inst.get_commandType():
print(" -%s CommandType did not match: %s for: %s" % (command_inst.get_commandType(), commandType, command_inst.get_command()))
praxis_logger_obj.log(" -%s CommandType did not match: %s for: %s" % (command_inst.get_commandType(), commandType, command_inst.get_command()))
return "", None
@ -40,7 +46,7 @@ def compile_and_load(commandType: AbstractCommand.CommandType) -> Dict[str, Abst
for dirName, subdirList, fileList in os.walk(implementations):
for file in fileList:
name = os.path.join(dirName, file)
print("compiling: %s" % name)
praxis_logger_obj.log("compiling: %s" % name)
name, command = compile_and_load_file(name, commandType)
if command is not None and command.command_type is commandType:
dic[name] = command
@ -56,7 +62,7 @@ def get_base_dir() -> str:
elif current == 'Praxis_Bot' or current == 'Praxis':
return check_dir(os.path.join(cwd, "commands"))
else:
print("could not find working directory for Praxis_Bot/commands")
praxis_logger_obj.log("could not find working directory for Praxis_Bot/commands")
raise Exception

36
cooldowns.py
View File

@ -7,6 +7,12 @@ from datetime import timedelta
import time
from time import sleep
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Cooldown_Action:
def __init__(self):
self.tag:str = ""
@ -81,41 +87,41 @@ if __name__ == "__main__":
cdName = "test"
testCD.setupCooldown(cdName, 20, 2)
print("CD Test 0: ")
praxis_logger_obj.log("CD Test 0: ")
for x in range(20):
testCD.actionTrigger("cdName")
sleep(0)
print(testCD.isCooldownActive("cdName"))
print("//Test Done//")
praxis_logger_obj.log(testCD.isCooldownActive("cdName"))
praxis_logger_obj.log("//Test Done//")
sleep(2)
print("CD Test 1: ")
praxis_logger_obj.log("CD Test 1: ")
for x in range(20):
testCD.actionTrigger(cdName)
sleep(0)
print(testCD.isCooldownActive("test"))
print("//Test Done//")
praxis_logger_obj.log(testCD.isCooldownActive("test"))
praxis_logger_obj.log("//Test Done//")
sleep(2)
print("CD Test 2: ")
praxis_logger_obj.log("CD Test 2: ")
for x in range(10):
testCD.actionTrigger(cdName)
sleep(0)
print(testCD.isCooldownActive(cdName))
print("//Test Done//")
praxis_logger_obj.log(testCD.isCooldownActive(cdName))
praxis_logger_obj.log("//Test Done//")
sleep(2)
print("CD Test 3: ")
praxis_logger_obj.log("CD Test 3: ")
for x in range(20):
testCD.actionTrigger(cdName)
sleep(0.05)
print(testCD.isCooldownActive(cdName))
print("//Test Done//")
praxis_logger_obj.log(testCD.isCooldownActive(cdName))
praxis_logger_obj.log("//Test Done//")
sleep(2)
print("CD Test 4: ")
praxis_logger_obj.log("CD Test 4: ")
for x in range(20):
testCD.actionTrigger(cdName)
sleep(0.6)
print(testCD.isCooldownActive(cdName))
print("//Test Done//")
praxis_logger_obj.log(testCD.isCooldownActive(cdName))
praxis_logger_obj.log("//Test Done//")

30
credentials.py
View File

@ -1,7 +1,11 @@
import json
import os
from enum import Enum
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Credential(Enum):
Twitch_Credential = 1
@ -59,7 +63,7 @@ class Credentials_Module():
self.DB_Credentials_List: list = []
def load_credentials(self):
print("Loading credentials...")
praxis_logger_obj.log("Loading credentials...")
fileList = self.list_credential_files()
for file in fileList:
if file.lower().find("twitch") != -1:
@ -111,33 +115,33 @@ class Credentials_Module():
return tobj
def find_Credential(self, credentialType, searchParam: str):
print("Searching for credential named: " + searchParam)
praxis_logger_obj.log("Searching for credential named: " + searchParam)
if credentialType.__name__ == Twitch_Credential.__name__:
print(".\{Twitch Credential Detected}")
praxis_logger_obj.log(".\{Twitch Credential Detected}")
credential_search_function = self.credentialSearchFunctions.get(Credential.Twitch_Credential)
output = credential_search_function(self, searchParam)
return output
elif credentialType.__name__ == Discord_Credential.__name__:
print(".\{Discord Credential Detected}")
praxis_logger_obj.log(".\{Discord Credential Detected}")
credential_search_function = self.credentialSearchFunctions.get(Credential.Twitch_Credential)
output = credential_search_function(self, searchParam)
return output
elif credentialType.__name__ == DB_Credential.__name__:
print(".\{DB Credential Detected}")
praxis_logger_obj.log(".\{DB Credential Detected}")
credential_search_function = self.credentialSearchFunctions.get(Credential.DB_Credential)
output = credential_search_function(self, searchParam)
return output
else:
print(".\{Something else Detected}")
praxis_logger_obj.log(".\{Something else Detected}")
return None
def find_Twitch_Credential(self, searchParam: str):
print("Searching for Twitch Credential named: " + searchParam)
praxis_logger_obj.log("Searching for Twitch Credential named: " + searchParam)
foundSomething = False
tempCert: Twitch_Credential = None
for cert in self.Twitch_Credentials_List:
if cert.username == searchParam:
print("Twitch Credential Found: {" + cert.username + "}")
praxis_logger_obj.log("Twitch Credential Found: {" + cert.username + "}")
tempCert = cert
foundSomething = True
if foundSomething:
@ -146,12 +150,12 @@ class Credentials_Module():
return None
def find_Discord_Credential(self, searchParam: str):
print("Searching for Discord Credential named: " + searchParam)
praxis_logger_obj.log("Searching for Discord Credential named: " + searchParam)
foundSomething = False
tempCert: Discord_Credential = None
for cert in self.Discord_Credentials_List:
if cert.nickname == searchParam:
print("Discord Credential Found: {" + cert.nickname + "}")
praxis_logger_obj.log("Discord Credential Found: {" + cert.nickname + "}")
tempCert = cert
foundSomething = True
if foundSomething:
@ -160,12 +164,12 @@ class Credentials_Module():
return None
def find_DB_Credential(self, searchParam: str):
print("Searching for DB Credential named: " + searchParam)
praxis_logger_obj.log("Searching for DB Credential named: " + searchParam)
foundSomething = False
tempCert: DB_Credential = None
for cert in self.DB_Credentials_List:
if cert.nickname == searchParam:
print("DB Credential Found: {" + cert.nickname + "}")
praxis_logger_obj.log("DB Credential Found: {" + cert.nickname + "}")
tempCert = cert
foundSomething = True
if foundSomething:

4
db.py
View File

@ -24,7 +24,7 @@ class db_module():
if createEngine:
self.engine = create_engine(credential.engine_url)
self.currentWorkingDB = credential.databaseName
print("SQL Engine Created")
praxis_logger_obj.log("SQL Engine Created")
def create_table(self, tableName: str = ""):
pass
@ -45,7 +45,7 @@ class db_module():
# temp = df.query(stmt)
# result = temp.get("response")
#
# # print(result)
# # praxis_logger_obj.log(result)
# i = len(temp.index.values)
#
# if i == 1:

8
help_module.py
View File

@ -7,13 +7,19 @@ import commands.loader as command_loader
import credentials
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Help_Module():
def __init__(self):
super().__init__()
#self.dbCredential: credentials.DB_Credential
def main(self):
print("[Help Module]> help test")
praxis_logger_obj.log("[Help Module]> help test")
self.isCommandEnabled = True
def help_command_response(self, command:AbstractCommand, responseType):

54
lights_module.py
View File

@ -8,13 +8,19 @@ import utilities_script as utilities
import credentials
import config
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Lights_Module():
def __init__(self):
super().__init__()
self.bridge_:Bridge = Bridge('192.168.191.146')
def main(self):
print("\nStarting up [Lights_Module]...")
praxis_logger_obj.log("\nStarting up [Lights_Module]...")
self.bridge_.connect()
self.bridge_.get_api()
@ -24,26 +30,26 @@ class Lights_Module():
groups = self.bridge_.get_group()
groupCount = 0
#print("\n -Listing Lights...")
#praxis_logger_obj.log("\n -Listing Lights...")
for l in light_list:
pass
#print(l.name)
#print("\n -Counting Groups...")
#praxis_logger_obj.log(l.name)
#praxis_logger_obj.log("\n -Counting Groups...")
for g in groups:
#print(g)
#praxis_logger_obj.log(g)
groupCount = int(g)
for gc in range(groupCount):
try:
#print("group n:" + str(gc))
#praxis_logger_obj.log("group n:" + str(gc))
group = self.bridge_.get_group(gc ,'name')
#print(group)
#praxis_logger_obj.log(group)
group_list.append(group)
#print(" --done adding")
#praxis_logger_obj.log(" --done adding")
except:
pass
#print(" --adding failed")
#praxis_logger_obj.log(" --adding failed")
#self.bridge_.set_group(18, "bri", 254) #This is max Brightness
#self.bridge_.set_group(18, "on", True) #This is will turn ON
@ -60,12 +66,12 @@ class Lights_Module():
#sleep(0.1)
#for stuffz in self.bridge_.scenes:
#print(stuffz)
#praxis_logger_obj.log(stuffz)
# This will set the group Downstairs to the Stream scene
#self.bridge_.run_scene("Downstairs", "Stream")
print("-[Lights_Module] Setup Complete")
praxis_logger_obj.log("-[Lights_Module] Setup Complete")
def setLight():
pass
@ -118,22 +124,22 @@ class Lights_Module():
def color_string_parser(self, message):
maxDigits = config.colorParse_maxDigits
print("Searching for color...")
praxis_logger_obj.log("Searching for color...")
xy_color = [0, 0]
for text in message:
#print("testing word")
#praxis_logger_obj.log("testing word")
if "red" in text.lower():
xy_color = self.rgb_to_xy(1,0,0)
print("-found: red")
praxis_logger_obj.log("-found: red")
if "blue" in text.lower():
print("-found: blue")
praxis_logger_obj.log("-found: blue")
xy_color = self.rgb_to_xy(0,0,1)
if "green" in text.lower():
print("-found: green")
praxis_logger_obj.log("-found: green")
xy_color = self.rgb_to_xy(0,1,0)
if "yellow" in text.lower():
print("-found: yellow")
praxis_logger_obj.log("-found: yellow")
xy_color = self.rgb_to_xy(
0.7,
0.64,
@ -141,23 +147,23 @@ class Lights_Module():
if "cyan" in text.lower():
print("-found: cyan")
praxis_logger_obj.log("-found: cyan")
xy_color = self.rgb_to_xy(0,1,1)
if "aquamarine" in text.lower():
print("-found: aquamarine")
praxis_logger_obj.log("-found: aquamarine")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(111,0,254),maxDigits),
round(utilities.rescale_value(218,0,254),maxDigits),
round(utilities.rescale_value(146,0,254),maxDigits))
if "turquoise" in text.lower():
print("-found: turquoise")
praxis_logger_obj.log("-found: turquoise")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(172,0,254),maxDigits),
round(utilities.rescale_value(233,0,254),maxDigits),
round(utilities.rescale_value(232,0,254),maxDigits))
if "orange" in text.lower():
print("-found: orange")
praxis_logger_obj.log("-found: orange")
xy_color = self.rgb_to_xy(
1,
round(utilities.rescale_value(126,0,254),maxDigits),
@ -165,21 +171,21 @@ class Lights_Module():
if "magenta" in text.lower():
print("-found: magenta")
praxis_logger_obj.log("-found: magenta")
xy_color = self.rgb_to_xy(
1,
0,
1)
if "purple" in text.lower():
print("-found: purple")
praxis_logger_obj.log("-found: purple")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(159,0,254),maxDigits),
round(utilities.rescale_value(32,0,254),maxDigits),
round(utilities.rescale_value(239,0,254),maxDigits))
if "violet" in text.lower():
print("-found: violet")
praxis_logger_obj.log("-found: violet")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(237,0,254),maxDigits),
round(utilities.rescale_value(129,0,254),maxDigits),

16
logsStandAlone_Command.log Normal file
View File

@ -0,0 +1,16 @@
INFO:root:Application running!
INFO:root:testLog
INFO:werkzeug: * Restarting with stat
INFO:root:Application running!
INFO:root:testLog
WARNING:werkzeug: * Debugger is active!
INFO:werkzeug: * Debugger PIN: 760-498-562
INFO:werkzeug: * Running on http://0.0.0.0:6009/ (Press CTRL+C to quit)
INFO:root:Application running!
INFO:root:testLog
INFO:werkzeug: * Restarting with stat
INFO:root:Application running!
INFO:root:testLog
WARNING:werkzeug: * Debugger is active!
INFO:werkzeug: * Debugger PIN: 760-498-562
INFO:werkzeug: * Running on http://0.0.0.0:6009/ (Press CTRL+C to quit)

16
main.py
View File

@ -15,6 +15,12 @@ import credentials
import threading
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
testModule_: test_module.Test_Module
userModule_: user_module.User_Module
@ -25,18 +31,18 @@ def main(inputArg):
def test_module_init(dbCert, Empty):
print("-init [TEST Module]")
praxis_logger_obj.log("-init [TEST Module]")
#testModule_.dbCredential = dbCert
testModule_.main()
def user_module_init(dbCert, Empty):
print("-init [USER Module]")
praxis_logger_obj.log("-init [USER Module]")
userModule_.dbCredential = dbCert
userModule_.main()
def thread_main():
if utility.isRunningInDocker() == True:
print("<[DOCKER Detected]>")
praxis_logger_obj.log("<[DOCKER Detected]>")
if not config.skip_splashScreen:
utility.splashScreen()
global credentials_manager
@ -66,11 +72,11 @@ def thread_main():
threads.append(thread_)
thread_.start()
print("---Post Thread Creation Test---\n")
praxis_logger_obj.log("---Post Thread Creation Test---\n")
for t in threads:
t.join()
print("---Point of no return---")
praxis_logger_obj.log("---Point of no return---")
if utility.isRunningInDocker() == False:
input()

13
praxis_logging.py Normal file
View File

@ -0,0 +1,13 @@
import logging
import utilities_script
class praxis_logger():
def init(self, name):
super().__init__()
self.logName = "logs/" + name + ".log"
utilities_script.get_dir("logs")
logging.basicConfig(filename=self.logName, level=logging.DEBUG)
logging.info('Application running!')
def log(self, msg):
logging.info(msg)

20
standalone_channelrewards.py
View File

@ -4,6 +4,12 @@ from flask import request
import channel_rewards.channelRewards_loader as rewards_loader
from channel_rewards.channelRewards_base import AbstractChannelRewards
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
api = flask.Flask(__name__)
# enable/disable this to get web pages of crashes returned
api.config["DEBUG"] = True
@ -12,7 +18,7 @@ loadedRewards = {}
def init():
# todo load entire reward library and cache it here
print("init stuff")
praxis_logger_obj.log("init stuff")
loadedRewards[AbstractChannelRewards.ChannelRewardsType.channelPoints] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.channelPoints)
loadedRewards[AbstractChannelRewards.ChannelRewardsType.twitch_bits] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.twitch_bits)
loadedRewards[AbstractChannelRewards.ChannelRewardsType.twitch_subs] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.twitch_subs)
@ -22,12 +28,12 @@ def is_reward(reward_name, reward_type) -> bool:
#global loadedRewards
tempType = reward_type.replace('ChannelRewardsType.', '')
realTempType = AbstractChannelRewards.ChannelRewardsType.__dict__[tempType]
#print(loadedRewards[realTempType])
#praxis_logger_obj.log(loadedRewards[realTempType])
for reward in loadedRewards[realTempType]:
print("found: ", reward, "type: ", type(reward))
praxis_logger_obj.log("found: ", reward, "type: ", type(reward))
if reward_name == reward:
print("Equal")
praxis_logger_obj.log("Equal")
return True
@ -48,15 +54,15 @@ def handle_reward(source, username, reward_name, reward_type, rewardPrompt, user
return flask.make_response("{\"message\":\"%s\"}" % reward_response, 200, {"Content-Type": "application/json"})
except:
return "None"
#print("Doing a reward")
#praxis_logger_obj.log("Doing a reward")
@api.route('/api/v1/reward', methods=['GET'])
def reward_check():
if 'reward_name' in request.args and 'reward_type' in request.args:
print("reward_name:", request.args['reward_name'],"reward_type:", request.args['reward_type'])
praxis_logger_obj.log("reward_name:", request.args['reward_name'],"reward_type:", request.args['reward_type'])
if is_reward(request.args['reward_name'], request.args['reward_type']):
print("about to send")
praxis_logger_obj.log("about to send")
return flask.make_response('', 200)
else:
return flask.make_response('', 404)

14
standalone_command.py
View File

@ -4,6 +4,12 @@ from flask import request
import commands.loader as command_loader
from commands.command_base import AbstractCommand
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
api = flask.Flask(__name__)
# enable/disable this to get web pages of crashes returned
api.config["DEBUG"] = True
@ -21,9 +27,9 @@ def load_commands():
def is_command(command: str) -> bool:
#print(command)
#praxis_logger_obj.log(command)
for cmd in loadedCommands:
#print(cmd)
#praxis_logger_obj.log(cmd)
if command == cmd:
return True
@ -35,7 +41,7 @@ def is_command(command: str) -> bool:
def handle_command(source, username, command, rest, bonusData):
if command == "!echo":
message = "Got payload [%s]" % rest
#print(message)
#praxis_logger_obj.log(message)
return flask.make_response("{\"message\":\"%s\"}" % message, 200, {"Content-Type": "application/json"})
cmd:AbstractCommand = loadedCommands[command]
@ -43,7 +49,7 @@ def handle_command(source, username, command, rest, bonusData):
cmd_response = cmd.do_command(source, username, command, rest, bonusData)
return flask.make_response("{\"message\":\"%s\"}" % cmd_response, 200, {"Content-Type": "application/json"})
#print("Doing a command")
#praxis_logger_obj.log("Doing a command")
@api.route('/api/v1/command', methods=['GET'])

28
standalone_discord_script.py
View File

@ -25,6 +25,12 @@ import discord.abc
from cooldowns import Cooldown_Module
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Discord_Module(discord.Client):
def __init__(self):
super().__init__()
@ -44,17 +50,17 @@ class Discord_Module(discord.Client):
await self.start(self.discordCredential.token)
def main(self):
print("starting loop")
praxis_logger_obj.log("starting loop")
self.loop.create_task(self.startup())
self.loop.run_forever()
async def on_ready(self):
print('Logged on as', self.user)
praxis_logger_obj.log('Logged on as', self.user)
async def on_message(self, message: discord.Message):
print("{" + message.guild.name + "}[ " + str(message.channel) + " ](" + message.author.display_name + ")> ")
#print(message.author.mention)
print(message.content)
praxis_logger_obj.log("{" + message.guild.name + "}[ " + str(message.channel) + " ](" + message.author.display_name + ")> ")
#praxis_logger_obj.log(message.author.mention)
praxis_logger_obj.log(message.content)
if not await self.isSenderBot(message):
# This will check for the praxis_bot-tts channel and will TTS stuff from there.
@ -82,7 +88,7 @@ class Discord_Module(discord.Client):
if self.cooldownModule.isCooldownActive("discordRateLimit") == False:
await self.exec_command(message, command, rest)
except:
print("something went wrong with a command")
praxis_logger_obj.log("something went wrong with a command")
async def is_command(self, word: str) -> bool:
# todo need to url-escape word
@ -97,7 +103,7 @@ class Discord_Module(discord.Client):
url = "http://standalone_command:6009/api/v1/exec_command?%s" % params
resp = requests.get(url)
if resp.status_code == 200:
print("Got the following message: %s" % resp.text)
praxis_logger_obj.log("Got the following message: %s" % resp.text)
data = loads(resp.text)
msg = data['message']
if msg is not None:
@ -121,15 +127,15 @@ class Discord_Module(discord.Client):
for bot in config.botList:
if message.author.display_name.lower() == bot.lower():
isBot = True
print("<{ bot detected! }> ")
praxis_logger_obj.log("<{ bot detected! }> ")
return isBot
async def isChannel_inConfigList(self, selectedChannel, selectedList):
#print(channel)
#print(selectedList)
#praxis_logger_obj.log(channel)
#praxis_logger_obj.log(selectedList)
is_Self = False
for discordChannel in selectedList:
#print("isSelf: " + str(discordChannel) + " vs " + str(selectedChannel))
#praxis_logger_obj.log("isSelf: " + str(discordChannel) + " vs " + str(selectedChannel))
if discordChannel == selectedChannel:
is_Self = True

73
standalone_lights.py
View File

@ -11,6 +11,12 @@ import config
import flask
from flask import request
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
api = flask.Flask(__name__)
# enable/disable this to get web pages of crashes returned
api.config["DEBUG"] = True
@ -22,7 +28,7 @@ class Lights_Module():
self.bridge_:Bridge = Bridge('192.168.191.146')
def main(self):
print("\nStarting up [Lights_Module]...")
praxis_logger_obj.log("\nStarting up [Lights_Module]...")
self.bridge_.connect()
self.bridge_.get_api()
@ -32,26 +38,26 @@ class Lights_Module():
groups = self.bridge_.get_group()
groupCount = 0
#print("\n -Listing Lights...")
#praxis_logger_obj.log("\n -Listing Lights...")
for l in light_list:
pass
#print(l.name)
#print("\n -Counting Groups...")
#praxis_logger_obj.log(l.name)
#praxis_logger_obj.log("\n -Counting Groups...")
for g in groups:
#print(g)
#praxis_logger_obj.log(g)
groupCount = int(g)
for gc in range(groupCount):
try:
#print("group n:" + str(gc))
#praxis_logger_obj.log("group n:" + str(gc))
group = self.bridge_.get_group(gc ,'name')
#print(group)
#praxis_logger_obj.log(group)
group_list.append(group)
#print(" --done adding")
#praxis_logger_obj.log(" --done adding")
except:
pass
#print(" --adding failed")
#praxis_logger_obj.log(" --adding failed")
#self.bridge_.set_group(18, "bri", 254) #This is max Brightness
#self.bridge_.set_group(18, "on", True) #This is will turn ON
@ -68,13 +74,13 @@ class Lights_Module():
#sleep(0.1)
#for stuffz in self.bridge_.scenes:
#print(stuffz)
#praxis_logger_obj.log(stuffz)
# This will set the group Downstairs to the Stream scene
#self.bridge_.run_scene("Downstairs", "Stream")
self.bridge_.run_scene("Downstairs", "Stream")
print("-[Lights_Module] Setup Complete")
#self.bridge_.run_scene("Downstairs", "Stream")
praxis_logger_obj.log("-[Lights_Module] Setup Complete")
def setLight():
pass
@ -132,22 +138,22 @@ class Lights_Module():
def color_string_parser(self, message):
maxDigits = config.colorParse_maxDigits
print("Searching for color...")
praxis_logger_obj.log("Searching for color...")
xy_color = [0, 0]
for text in message:
#print("testing word")
#praxis_logger_obj.log("testing word")
if "red" in text.lower():
xy_color = self.rgb_to_xy(1,0,0)
print("-found: red")
praxis_logger_obj.log("-found: red")
if "blue" in text.lower():
print("-found: blue")
praxis_logger_obj.log("-found: blue")
xy_color = self.rgb_to_xy(0,0,1)
if "green" in text.lower():
print("-found: green")
praxis_logger_obj.log("-found: green")
xy_color = self.rgb_to_xy(0,1,0)
if "yellow" in text.lower():
print("-found: yellow")
praxis_logger_obj.log("-found: yellow")
xy_color = self.rgb_to_xy(
0.7,
0.64,
@ -155,23 +161,23 @@ class Lights_Module():
if "cyan" in text.lower():
print("-found: cyan")
praxis_logger_obj.log("-found: cyan")
xy_color = self.rgb_to_xy(0,1,1)
if "aquamarine" in text.lower():
print("-found: aquamarine")
praxis_logger_obj.log("-found: aquamarine")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(111,0,254),maxDigits),
round(utilities.rescale_value(218,0,254),maxDigits),
round(utilities.rescale_value(146,0,254),maxDigits))
if "turquoise" in text.lower():
print("-found: turquoise")
praxis_logger_obj.log("-found: turquoise")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(172,0,254),maxDigits),
round(utilities.rescale_value(233,0,254),maxDigits),
round(utilities.rescale_value(232,0,254),maxDigits))
if "orange" in text.lower():
print("-found: orange")
praxis_logger_obj.log("-found: orange")
xy_color = self.rgb_to_xy(
1,
round(utilities.rescale_value(126,0,254),maxDigits),
@ -179,21 +185,21 @@ class Lights_Module():
if "magenta" in text.lower():
print("-found: magenta")
praxis_logger_obj.log("-found: magenta")
xy_color = self.rgb_to_xy(
1,
0,
1)
if "purple" in text.lower():
print("-found: purple")
praxis_logger_obj.log("-found: purple")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(159,0,254),maxDigits),
round(utilities.rescale_value(32,0,254),maxDigits),
round(utilities.rescale_value(239,0,254),maxDigits))
if "violet" in text.lower():
print("-found: violet")
praxis_logger_obj.log("-found: violet")
xy_color = self.rgb_to_xy(
round(utilities.rescale_value(237,0,254),maxDigits),
round(utilities.rescale_value(129,0,254),maxDigits),
@ -208,8 +214,8 @@ def init():
RGB_Lights.main()
def do_lights_command(user="", lightGroup="all", command = "", rest = ""):
returnString = "None"
print("about to do something ......")
returnString = ""
praxis_logger_obj.log("about to do something ......")
#bot.return_message("\nRGB Command Detected!")
if rest is not "":
@ -221,12 +227,12 @@ def do_lights_command(user="", lightGroup="all", command = "", rest = ""):
tempParsedMessage = tempFix.split(" ")
sceneCommand = False
if (len(tempParsedMessage)) > 2:
print("RGB Command!")
praxis_logger_obj.log("RGB Command!")
rgb_r = float(tempParsedMessage[1])
rgb_g = float(tempParsedMessage[2])
rgb_b = float(tempParsedMessage[3])
xy_result = RGB_Lights.rgb_to_xy(rgb_r, rgb_g, rgb_b)
print("got XY")
praxis_logger_obj.log("got XY")
RGB_Lights.bridge_.set_group(16, "xy", xy_result)
#bot.return_message("sent color to [Lights_Module]")
else:
@ -259,11 +265,12 @@ def do_lights_command(user="", lightGroup="all", command = "", rest = ""):
#bot.return_message("sent color to [Lights_Module]")
if sceneCommand == True:
print("Scene Command!")
praxis_logger_obj.log("Scene Command!")
returnString = user + " changed the light's color!"
returnString = user + " changed the lights color!"
praxis_logger_obj.log(returnString)
return returnString
return flask.make_response("{\"message\":\"%s\"}" % returnString, 200, {"Content-Type": "application/json"})
@ -278,7 +285,7 @@ def exec_lights():
if 'command' not in request.args:
return flask.make_response('{\"text\":"Argument \'scene_name\' not in request"}', 400)
print("about to do something ......")
praxis_logger_obj.log("about to do something ......")
RGB_Lights.main()
return do_lights_command(user_name, request.args['light_group'], request.args['command'], request.args['rest'])

54
standalone_twitch_pubsub.py
View File

@ -20,6 +20,12 @@ from uuid import UUID
from cooldowns import Cooldown_Module
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Twitch_Pubsub():
def __init__(self):
super().__init__()
@ -42,14 +48,14 @@ class Twitch_Pubsub():
def get_tokens(self):
self.twitch.authenticate_app(self.target_scope)
for scope_ in self.target_scope:
print(scope_)
praxis_logger_obj.log(scope_)
auth = UserAuthenticator(self.twitch, self.target_scope, force_verify=True)
token, refresh_token = auth.authenticate()
if token is not None: print("found token")
if refresh_token is not None: print("found refresh_token")
print(token)
print(refresh_token)
if token is not None: praxis_logger_obj.log("found token")
if refresh_token is not None: praxis_logger_obj.log("found refresh_token")
praxis_logger_obj.log(token)
praxis_logger_obj.log(refresh_token)
self.twitch.set_user_authentication(token, self.target_scope, refresh_token)
@ -57,12 +63,12 @@ class Twitch_Pubsub():
self.pubsub = PubSub(self.twitch)
#self.pubsub.ping_frequency = 30
self.pubsub.start()
print("started")
praxis_logger_obj.log("started")
def next(self):
user_id = self.twitch.get_users(logins=[config.autoJoin_TwitchChannel])['data'][0]['id']
if user_id is not None: print("found user_id")
print(user_id)
if user_id is not None: praxis_logger_obj.log("found user_id")
praxis_logger_obj.log(user_id)
self.uuid_1 = self.pubsub.listen_whispers(user_id, self.callback_whisper)
self.uuid_2 = self.pubsub.listen_channel_points(user_id, self.callback_channelPoints)
#input('press ENTER to close...')
@ -73,22 +79,22 @@ class Twitch_Pubsub():
self.pubsub.stop()
def callback_whisper(self, uuid: UUID, data: dict) -> None:
print('got callback for UUID ' + str(uuid))
praxis_logger_obj.log('got callback for UUID ' + str(uuid))
pprint(data)
def callback_channelPoints(self, uuid: UUID, data: dict) -> None:
print("Channel Point Redemption")
print('got callback for UUID ' + str(uuid))
praxis_logger_obj.log("Channel Point Redemption")
praxis_logger_obj.log('got callback for UUID ' + str(uuid))
pprint(data)
#print("attempting to get data: ")
#print(data['data']['redemption']['user']['display_name'])
#print(data['data']['redemption']['reward']['title'])
#print(data['data']['redemption']['reward']['prompt'])
#praxis_logger_obj.log("attempting to get data: ")
#praxis_logger_obj.log(data['data']['redemption']['user']['display_name'])
#praxis_logger_obj.log(data['data']['redemption']['reward']['title'])
#praxis_logger_obj.log(data['data']['redemption']['reward']['prompt'])
try:
userinput = data['data']['redemption']['user_input']
except:
userinput = ""
#print(userinput)
#praxis_logger_obj.log(userinput)
self.callback_EXEC(
data['data']['redemption']['user']['display_name'],
data['data']['redemption']['reward']['title'],
@ -98,29 +104,29 @@ class Twitch_Pubsub():
data)
def callback_bits(self, uuid: UUID, data: dict) -> None:
print("Bits Redemption")
print('got callback for UUID ' + str(uuid))
praxis_logger_obj.log("Bits Redemption")
praxis_logger_obj.log('got callback for UUID ' + str(uuid))
pprint(data)
def callback_subs(self, uuid: UUID, data: dict) -> None:
print("Subs Redemption")
print('got callback for UUID ' + str(uuid))
praxis_logger_obj.log("Subs Redemption")
praxis_logger_obj.log('got callback for UUID ' + str(uuid))
pprint(data)
def callback_EXEC(self, sender, rewardName:str, rewardType, rewardPrompt, userInput, raw_data):
try:
is_actionable = self.is_reward(rewardName, rewardType)
if is_actionable:
print("Trying to do the thing")
praxis_logger_obj.log("Trying to do the thing")
if self.cooldownModule.isCooldownActive("twitchChat") == False:
self.exec_reward(sender, rewardName, rewardType, rewardPrompt, userInput, raw_data)
except:
print("something went wrong with a reward")
praxis_logger_obj.log("something went wrong with a reward")
def is_reward(self, rewardName, rewardType):
# todo need to url-escape word
clean_param = urlencode({'reward_name': rewardName, 'reward_type':rewardType})
print(rewardName, rewardType)
praxis_logger_obj.log(rewardName, rewardType)
#standalone_channelrewards
url = "http://standalone_channelrewards:6969/api/v1/reward?%s" % clean_param
resp = requests.get(url)
@ -140,7 +146,7 @@ class Twitch_Pubsub():
url = "http://standalone_channelrewards:6969/api/v1/exec_reward?%s" % params
resp = requests.get(url)
if resp.status_code == 200:
print("Got the following message: %s" % resp.text)
praxis_logger_obj.log("Got the following message: %s" % resp.text)
data = loads(resp.text)
msg = data['message']
if msg is not None:

30
standalone_twitch_script.py
View File

@ -12,6 +12,12 @@ from cooldowns import Cooldown_Module
import commands.command_base
import utilities_script as utility
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Twitch_Module():
def __init__(self):
super().__init__()
@ -33,7 +39,7 @@ class Twitch_Module():
def join_channel(self, credential: credentials.Twitch_Credential, channel_name: str):
channel_name = "#" + channel_name
print("Connecting to Channel: " + channel_name + "...")
praxis_logger_obj.log("Connecting to Channel: " + channel_name + "...")
if credential is None:
credential = self.twitchCredential
@ -47,23 +53,23 @@ class Twitch_Module():
)
self.chat.subscribe(self.twitch_chat)
print("Connected to Channel: ", channel_name)
praxis_logger_obj.log("Connected to Channel: ", channel_name)
def leave_channel(self):
print("Leaving Channel", self.chat.channel)
praxis_logger_obj.log("Leaving Channel", self.chat.channel)
self.chat.irc.leave_channel(self.chat.channel)
self.chat.irc.socket.close()
def send_message(self, message):
isBlocked = self.isChannel_inConfigList(self.chat.channel, config.block_TwitchChannelsMessaging)
# print("isBlocked: " + str(isBlocked) + " for: " + self.chat.channel)
# praxis_logger_obj.log("isBlocked: " + str(isBlocked) + " for: " + self.chat.channel)
#if self.
if utility.contains_slur(message): isBlocked = True
if self.cooldownModule.isCooldownActive(
"twitchChat") == False and not isBlocked and not config.blockAll_TwitchChatChannelsMessaging:
self.chat.send(message)
# print("Sent ChatMSG")
# praxis_logger_obj.log("Sent ChatMSG")
self.cooldownModule.actionTrigger("twitchChat")
def is_command(self, word: str) -> bool:
@ -79,7 +85,7 @@ class Twitch_Module():
url = "http://standalone_command:6009/api/v1/exec_command?%s" % params
resp = requests.get(url)
if resp.status_code == 200:
print("Got the following message: %s" % resp.text)
praxis_logger_obj.log("Got the following message: %s" % resp.text)
data = loads(resp.text)
msg = data['message']
if msg is not None:
@ -93,7 +99,7 @@ class Twitch_Module():
# This reacts to messages
def twitch_chat(self, message: twitch.chat.Message) -> None:
print("[#" + message.channel + "](" + message.sender + ")> " + message.text)
praxis_logger_obj.log("[#" + message.channel + "](" + message.sender + ")> " + message.text)
command, rest = utility.parse_line(message.text)
try:
@ -102,19 +108,19 @@ class Twitch_Module():
if self.cooldownModule.isCooldownActive("twitchChat") == False:
self.exec_command(message ,command, rest)
except:
print("something went wrong with a command")
praxis_logger_obj.log("something went wrong with a command")
def isChannel_inConfigList(self, selectedChannel, selectedList):
# print(channel)
# print(selectedList)
# praxis_logger_obj.log(channel)
# praxis_logger_obj.log(selectedList)
is_Self = False
for twitchChannel in selectedList:
if twitchChannel == selectedChannel:
is_Self = True
# if is_Self:
# print("Is Self")
# praxis_logger_obj.log("Is Self")
# if not is_Self:
# print("Is Not Self")
# praxis_logger_obj.log("Is Not Self")
return is_Self

10
standalone_webSource.py
View File

@ -20,6 +20,12 @@ import utilities_script as utility
import chyron_module
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class webSource_Module():
webSources:Flask = Flask('webSources')
@ -28,7 +34,7 @@ class webSource_Module():
self.dbCredential: credentials.DB_Credential
def main(self, port_=5000):
print("starting up on port: ", port_)
praxis_logger_obj.log("starting up on port: ", port_)
self.webSources.run(host="0.0.0.0", port= port_)
@webSources.route('/')
@ -42,7 +48,7 @@ class webSource_Module():
@webSources.route('/temptext/<filename>/')
def textSource_tempText(filename):
print("trying file: ", filename)
praxis_logger_obj.log("trying file: ", filename)
tempModule = tempText_Module.tempText_Module()
return tempModule.getTempTextFile(filename)

9
tempText_Module.py
View File

@ -1,6 +1,11 @@
import config
import utilities_script as utilities
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class tempText_Module():
def __init__(self):
@ -57,7 +62,7 @@ class tempText_Module():
file = open(real_file_path, "rb")
text = file.read()
#print(text)
#praxis_logger_obj.log(text)
file.close
return text
@ -74,7 +79,7 @@ class tempTextItem():
self.itemComputedString = ""
def setupItem(self, name, title, content):
print("\nSetting up tempTextItem {", name,"}[", title, content, "]")
praxis_logger_obj.log("\nSetting up tempTextItem {", name,"}[", title, content, "]")
self.itemName = name
self.itemTitle = title
self.itemContent = content

8
test_module.py
View File

@ -3,13 +3,19 @@ import db
import credentials
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Test_Module():
def __init__(self):
super().__init__()
self.dbCredential: credentials.DB_Credential
def main(self):
print("[TEST Module]> test")
praxis_logger_obj.log("[TEST Module]> test")
if __name__ == "__main__":

11
tts.py
View File

@ -1,6 +1,11 @@
import datetime
import hashlib
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
import requests
from gtts import gTTS
@ -14,9 +19,9 @@ streamLabsUrl = "https://streamlabs.com/polly/speak"
def tts(inputText: str, *args):
outpath = create_speech_file(inputText)
if utility.isRunningInDocker() == True:
print("Docker Detected, skipping playsound()")
praxis_logger_obj.log("Docker Detected, skipping playsound()")
else:
print("Playing Sound...")
praxis_logger_obj.log("Playing Sound...")
playsound(outpath)
@ -108,6 +113,6 @@ def get_tts_dir():
if __name__ == "__main__":
print("Enter Text: ")
praxis_logger_obj.log("Enter Text: ")
textInput = str(input())
tts(textInput)

20
twitch_generate_credentials.py
View File

@ -10,6 +10,12 @@ from twitchAPI.oauth import UserAuthenticator
from pprint import pprint
from uuid import UUID
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class Twitch_Credential_Maker():
def __init__(self):
@ -21,15 +27,15 @@ class Twitch_Credential_Maker():
def get_tokens(self):
self.twitch.authenticate_app(self.target_scope)
for scope_ in self.target_scope:
print(scope_)
praxis_logger_obj.log(scope_)
auth = UserAuthenticator(self.twitch, self.target_scope, force_verify=True)
token, refresh_token = auth.authenticate()
if token is not None: print("found token")
if refresh_token is not None: print("found refresh_token\n")
print("token: ", token)
print("refresh_token: ", refresh_token)
print("")
if token is not None: praxis_logger_obj.log("found token")
if refresh_token is not None: praxis_logger_obj.log("found refresh_token\n")
praxis_logger_obj.log("token: ", token)
praxis_logger_obj.log("refresh_token: ", refresh_token)
praxis_logger_obj.log("")
return token, refresh_token
@ -45,5 +51,5 @@ if __name__ == "__main__":
#pprint(testModule.twitch.get_users(logins=['thecuriousnerd']))
testModule.get_tokens()
print("Ready to close")
praxis_logger_obj.log("Ready to close")
input()

14
user_module.py
View File

@ -13,6 +13,12 @@ from cooldowns import Cooldown_Module
import utilities_script as utility
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
class User_Module():
def __init__(self):
super().__init__()
@ -27,10 +33,10 @@ class User_Module():
def main(self):
time.sleep(.01)
print("\nWaiting on User input...\n\n")
praxis_logger_obj.log("\nWaiting on User input...\n\n")
if utility.isRunningInDocker() == True:
self.inputLoop = False
print("\nNo User's Input Allowed")
praxis_logger_obj.log("\nNo User's Input Allowed")
while self.inputLoop:
keyboardInput = input()
@ -77,7 +83,7 @@ class User_Module():
command.do_command(self, message)
except Exception as e:
# Undo the following for debug stuff
#print(e)
#praxis_logger_obj.log(e)
pass # we don't care
def eval_commands_SpecialActionCheck(self):
@ -88,7 +94,7 @@ class User_Module():
pass
def return_message(self, returnedMessage):
print(returnedMessage)
praxis_logger_obj.log(returnedMessage)
def tts(self, message):
tts.tts(message)

21
utilities_script.py
View File

@ -1,5 +1,4 @@
from asyncio.tasks import sleep
import os
import sys
import re
import psutil
@ -9,6 +8,12 @@ import time
import config as config
import art
import os
import praxis_logging
praxis_logger_obj = praxis_logging.praxis_logger()
praxis_logger_obj.init(os.path.basename(__file__))
praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__))
clearScreen = lambda: os.system('cls' if os.name == 'nt' else 'clear')
urlMatcher = re.compile("(https?:(/{1,3}|[a-z0-9%])|[a-z0-9.-]+[.](com|net|org|edu|gov|mil|aero|asia|biz|cat|coop|info|int|jobs|mobi|museum|name|post|pro|tel|travel|xxx|ac|ad|ae|af|ag|ai|al|am|an|ao|aq|ar|as|at|au|aw|ax|az|ba|bb|bd|be|bf|bg|bh|bi|bj|bm|bn|bo|br|bs|bt|bv|bw|by|bz|ca|cc|cd|cf|cg|ch|ci|ck|cl|cm|cn|co|cr|cs|cu|cv|cx|cy|cz|dd|de|dj|dk|dm|do|dz|ec|ee|eg|eh|er|es|et|eu|fi|fj|fk|fm|fo|fr|ga|gb|gd|ge|gf|gg|gh|gi|gl|gm|gn|gp|gq|gr|gs|gt|gu|gw|gy|hk|hm|hn|hr|ht|hu|id|ie|il|im|in|io|iq|ir|is|it|je|jm|jo|jp|ke|kg|kh|ki|km|kn|kp|kr|kw|ky|kz|la|lb|lc|li|lk|lr|ls|lt|lu|lv|ly|ma|mc|md|me|mg|mh|mk|ml|mm|mn|mo|mp|mq|mr|ms|mt|mu|mv|mw|mx|my|mz|na|nc|ne|nf|ng|ni|nl|no|np|nr|nu|nz|om|pa|pe|pf|pg|ph|pk|pl|pm|pn|pr|ps|pt|pw|py|qa|re|ro|rs|ru|rw|sa|sb|sc|sd|se|sg|sh|si|sj|Ja|sk|sl|sm|sn|so|sr|ss|st|su|sv|sx|sy|sz|tc|td|tf|tg|th|tj|tk|tl|tm|tn|to|tp|tr|tt|tv|tw|tz|ua|ug|uk|us|uy|uz|va|vc|ve|vg|vi|vn|vu|wf|ws|ye|yt|yu|za|zm|zw))")
@ -22,7 +27,7 @@ def get_args(text: str) -> list:
def does_contain_OnlyNumbers(text):
isJustNumbers = False
print("checking numbers")
praxis_logger_obj.log("checking numbers")
try:
for x in range(10):
if str(x) in str(text):
@ -35,9 +40,9 @@ def does_contain_OnlyNumbers(text):
return isJustNumbers
def rescale_value(value, min, max):
#print("trying Rescale")
#praxis_logger_obj.log("trying Rescale")
returnValue = (value - min) / (max - min)
#print("got ", returnValue)
#praxis_logger_obj.log("got ", returnValue)
return returnValue
def get_dir(selected_dir):
@ -62,7 +67,7 @@ def contains_slur(input: str):
break
if containsSlur:
print("<{ slur detected! }> ")
praxis_logger_obj.log("<{ slur detected! }> ")
return containsSlur
def parse_line(message: str):
@ -131,9 +136,9 @@ def splashScreen():
art.tprint("----------",font="slant")
art.tprint("Praxis Bot",font="graffiti")
art.tprint("----------",font="slant")
print("-Maintained by Alex Orid, TheCuriousNerd.com\nFor help visit discord.gg/thecuriousnerd")
print("ver: " + config.praxisVersion_Alpha + config.praxisVersion_Delta + config.praxisVersion_Omega)
print("\n\n\n")
praxis_logger_obj.log("-Maintained by Alex Orid, TheCuriousNerd.com\nFor help visit discord.gg/thecuriousnerd")
praxis_logger_obj.log("ver: " + config.praxisVersion_Alpha + config.praxisVersion_Delta + config.praxisVersion_Omega)
praxis_logger_obj.log("\n\n\n")
if not config.skip_splashScreenSleep:
time.sleep(3)