first test of channels class

This commit is contained in:
deathbybandaid 2019-05-18 19:17:04 -04:00
parent 76e52c1ae5
commit 590888ba33
10 changed files with 209 additions and 223 deletions

View File

@ -5,14 +5,74 @@ This is the SpiceBot Logs system.
This Class stores logs in an easy to access manner
"""
import sopel
import re
class BotChannels():
"""This Logs all channels known to the server"""
def __init__(self):
self.channel_lock = False
self.SpiceBot_Channels = {
"list": {}
"list": {},
"InitialProcess": False
}
def channel_list_request(bot):
bot.write(['LIST'])
def channel_list_recieve_start(self):
self.channel_lock = True
def channel_list_recieve_input(self, trigger):
channel, _, topic = trigger.args[1:]
if channel.lower() not in self.SpiceBot_Channels['list'].keys():
self.SpiceBot_Channels['list'][channel.lower()] = dict()
self.SpiceBot_Channels['list'][channel.lower()]['name'] = channel
self.SpiceBot_Channels['list'][channel.lower()]['topic'] = self.topic_compile(topic)
self.channel_lock = True
def channel_list_recieve_finish(self):
self.channel_lock.release()
if not self.SpiceBot_Channels['InitialProcess']:
self.SpiceBot_Channels['InitialProcess'] = True
self.channel_lock = False
def topic_compile(self, topic):
actual_topic = re.compile(r'^\[\+[a-zA-Z]+\] (.*)')
topic = re.sub(actual_topic, r'\1', topic)
return topic
def bot_part_empty(bot):
"""Don't stay in empty channels"""
ignorepartlist = []
if bot.config.core.logging_channel:
ignorepartlist.append(bot.config.core.logging_channel)
for channel in bot.channels.keys():
if len(bot.channels[channel].privileges.keys()) == 1 and channel not in ignorepartlist and channel.startswith("#"):
bot.part(channel, "Leaving Empty Channel")
if channel.lower() in botchannels.SpiceBot_Channels['list']:
del botchannels.SpiceBot_Channels['list'][channel.lower()]
def join_all_channels(bot):
if bot.config.SpiceBot_Channels.joinall:
for channel in botchannels.SpiceBot_Channels['list'].keys():
if channel.startswith("#"):
if channel not in bot.channels.keys() and channel not in bot.config.SpiceBot_Channels.chanignore:
bot.write(('JOIN', bot.nick, botchannels.SpiceBot_Channels['list'][channel]['name']))
if channel not in bot.channels.keys() and bot.config.SpiceBot_Channels.operadmin:
bot.write(('SAJOIN', bot.nick, botchannels.SpiceBot_Channels['list'][channel]['name']))
def chanadmin_all_channels(bot):
# Chan ADMIN +a
for channel in bot.channels.keys():
if channel.startswith("#"):
if channel not in bot.config.SpiceBot_Channels.chanignore:
if bot.config.SpiceBot_Channels.operadmin:
if not bot.channels[channel].privileges[bot.nick] < sopel.module.ADMIN:
bot.write(('SAMODE', channel, "+a", bot.nick))
else:
bot.part(channel)
botchannels = BotChannels()

View File

@ -1,128 +0,0 @@
#!/usr/bin/env python
# coding=utf-8
from __future__ import unicode_literals, absolute_import, print_function, division
# sopel imports
import sopel.module
from sopel.config.types import StaticSection, ValidatedAttribute, ListAttribute
from sopel_modules.SpiceBot.Logs import botlogs
from sopel_modules.SpiceBot.Events import botevents
from sopel_modules.SpiceBot_SBTools import join_all_channels, chanadmin_all_channels, channel_list_current
import spicemanip
import time
class SpiceBot_Channels_MainSection(StaticSection):
announcenew = ValidatedAttribute('announcenew', default=False)
joinall = ValidatedAttribute('joinall', default=False)
operadmin = ValidatedAttribute('operadmin', default=False)
chanignore = ListAttribute('chanignore')
def configure(config):
config.define_section("SpiceBot_Channels", SpiceBot_Channels_MainSection, validate=False)
config.SpiceBot_Channels.configure_setting('announcenew', 'SpiceBot_Channels Announce New Channels')
config.SpiceBot_Channels.configure_setting('joinall', 'SpiceBot_Channels JOIN New Channels')
config.SpiceBot_Channels.configure_setting('operadmin', 'SpiceBot_Channels OPER ADMIN MODE')
config.SpiceBot_Channels.configure_setting('chanignore', 'SpiceBot_Channels Ignore JOIN for channels')
def setup(bot):
botlogs.log('SpiceBot_Channels', "Starting setup procedure")
botevents.startup_add([botevents.BOT_CHANNELS])
bot.config.define_section("SpiceBot_Channels", SpiceBot_Channels_MainSection, validate=False)
if "SpiceBot_Channels" not in bot.memory:
bot.memory["SpiceBot_Channels"] = {"channels": {}, "InitialProcess": False, "ProcessLock": False}
def shutdown(bot):
if "SpiceBot_Channels" in bot.memory:
del bot.memory["SpiceBot_Channels"]
@sopel.module.event(botevents.BOT_CONNECTED)
@sopel.module.rule('.*')
def trigger_channel_list_initial(bot, trigger):
if bot.config.SpiceBot_Channels.operadmin:
bot.write(('SAMODE', bot.nick, '+q'))
bot_part_empty(bot)
bot.write(['LIST'])
bot.memory['SpiceBot_Channels']['ProcessLock'] = True
botlogs.log('SpiceBot_Channels', "Initial Channel list populating")
starttime = time.time()
while not bot.memory['SpiceBot_Channels']['InitialProcess']:
timesince = time.time() - starttime
if timesince < 60:
pass
else:
botlogs.log('SpiceBot_Channels', "Initial Channel list populating Timed Out")
bot.memory['SpiceBot_Channels']['InitialProcess'] = True
channel_list_current(bot)
foundchannelcount = len(bot.memory['SpiceBot_Channels']['channels'].keys())
botlogs.log('SpiceBot_Channels', "Channel listing finished! " + str(foundchannelcount) + " channel(s) found.")
join_all_channels(bot)
chanadmin_all_channels(bot)
if "*" in bot.memory['SpiceBot_Channels']['channels']:
del bot.memory['SpiceBot_Channels']['channels']["*"]
bot_part_empty(bot)
botevents.trigger(bot, botevents.BOT_CHANNELS, "SpiceBot_Channels")
@sopel.module.event(botevents.BOT_CHANNELS)
@sopel.module.rule('.*')
def trigger_channel_list_recurring(bot, trigger):
while True:
try:
time.sleep(1800)
bot_part_empty(bot)
oldlist = list(bot.memory['SpiceBot_Channels']['channels'].keys())
bot.write(['LIST'])
bot.memory['SpiceBot_Channels']['ProcessLock'] = True
while bot.memory['SpiceBot_Channels']['ProcessLock']:
pass
newlist = [item.lower() for item in oldlist if item.lower() not in list(bot.memory['SpiceBot_Channels']['channels'].keys())]
if "*" in newlist:
newlist.remove("*")
if len(newlist) and bot.config.SpiceBot_Channels.announcenew:
bot.osd(["The Following channel(s) are new:", spicemanip.main(newlist, 'andlist')], bot.channels.keys())
join_all_channels(bot)
chanadmin_all_channels(bot)
if "*" in bot.memory['SpiceBot_Channels']['channels']:
del bot.memory['SpiceBot_Channels']['channels']["*"]
bot_part_empty(bot)
except KeyError:
return
def bot_part_empty(bot):
"""Don't stay in empty channels"""
ignorepartlist = []
if bot.config.core.logging_channel:
ignorepartlist.append(bot.config.core.logging_channel)
for channel in bot.channels.keys():
if len(bot.channels[channel].privileges.keys()) == 1 and channel not in ignorepartlist and channel.startswith("#"):
bot.part(channel, "Leaving Empty Channel")
if channel.lower() in bot.memory['SpiceBot_Channels']['channels']:
del bot.memory['SpiceBot_Channels']['channels'][channel.lower()]

View File

@ -1,32 +0,0 @@
# coding=utf-8
from __future__ import unicode_literals, absolute_import, division, print_function
import sopel.module
from sopel_modules.SpiceBot_SBTools import topic_compile
@sopel.module.event('321')
@sopel.module.rule('.*')
def watch_chanlist_start(bot, trigger):
bot.memory['SpiceBot_Channels']['ProcessLock'] = True
@sopel.module.event('322')
@sopel.module.rule('.*')
def watch_chanlist_populate(bot, trigger):
channel, _, topic = trigger.args[1:]
if channel.lower() not in bot.memory['SpiceBot_Channels']['channels']:
bot.memory['SpiceBot_Channels']['channels'][channel.lower()] = dict()
bot.memory['SpiceBot_Channels']['channels'][channel.lower()]['name'] = channel
bot.memory['SpiceBot_Channels']['channels'][channel.lower()]['topic'] = topic_compile(topic)
bot.memory['SpiceBot_Channels']['ProcessLock'] = True
@sopel.module.event('323')
@sopel.module.rule('.*')
def watch_chanlist_complete(bot, trigger):
bot.memory['SpiceBot_Channels']['ProcessLock'] = False
if not bot.memory['SpiceBot_Channels']['InitialProcess']:
bot.memory['SpiceBot_Channels']['InitialProcess'] = True

View File

@ -7,11 +7,8 @@ import sopel.module
import spicemanip
from sopel_modules.SpiceBot.Events import botevents
from sopel_modules.SpiceBot_SBTools import (
sopel_triggerargs, inlist, channel_privs,
join_all_channels, channel_list_current,
)
from .Channels import bot_part_empty
from sopel_modules.SpiceBot.Channels import botchannels
from sopel_modules.SpiceBot_SBTools import sopel_triggerargs, inlist, channel_privs
@botevents.check_ready([botevents.BOT_LOADED])
@ -27,8 +24,6 @@ def nickname_comand_channels(bot, trigger):
triggerargs = spicemanip.main(triggerargs, '2+', 'list')
channel_list_current(bot)
if commandused == 'list':
chanlist = spicemanip.main(bot.channels.keys(), 'andlist')
bot.osd("You can find me in " + chanlist)
@ -36,14 +31,14 @@ def nickname_comand_channels(bot, trigger):
elif commandused == 'total':
botcount = len(bot.channels.keys())
servercount = len(bot.memory['SpiceBot_Channels']['channels'].keys())
servercount = len(botchannels.SpiceBot_Channels['list'].keys())
bot.osd("I am in " + str(botcount) + " of " + str(servercount) + " channel(s) available on this server.")
return
elif commandused == 'random':
channel = spicemanip.main(bot.memory['SpiceBot_Channels']['channels'], 'random')
topic = bot.memory['SpiceBot_Channels']['channels'][channel]['topic']
msg = ["Random channel for you: {}.".format(bot.memory['SpiceBot_Channels']['channels'][channel]['name'])]
channel = spicemanip.main(botchannels.SpiceBot_Channels['list'], 'random')
topic = botchannels.SpiceBot_Channels['list'][channel]['topic']
msg = ["Random channel for you: {}.".format(botchannels.SpiceBot_Channels['list'][channel]['name'])]
if topic and not topic.isspace():
msg.append("The topic is: {}".format(topic))
else:
@ -55,17 +50,15 @@ def nickname_comand_channels(bot, trigger):
if not trigger.admin:
bot.osd("You do not have permission to update the channel listing.")
return
bot_part_empty(bot)
bot.write(['LIST'])
botchannels.bot_part_empty(bot)
botchannels.channel_list_request(bot)
bot.osd(["[SpiceBot_Channels]", "I am now updating the channel listing for this server."])
bot.memory['SpiceBot_Channels']['ProcessLock'] = True
while bot.memory['SpiceBot_Channels']['ProcessLock']:
while botchannels.channel_lock:
pass
join_all_channels(bot)
foundchannelcount = len(bot.memory['SpiceBot_Channels']['channels'].keys())
botchannels.join_all_channels(bot)
foundchannelcount = len(botchannels.SpiceBot_Channels['list'].keys())
bot.osd("[SpiceBot_Channels]", "Channel listing finished!", str(foundchannelcount) + " channel(s) found.")
bot_part_empty(bot)
botchannels.bot_part_empty(bot)
return
elif commandused == 'topic':
@ -73,11 +66,11 @@ def nickname_comand_channels(bot, trigger):
bot.osd("Channel name input missing.")
return
channel = spicemanip.main(triggerargs, 1)
if not inlist(bot, channel.lower(), bot.memory['SpiceBot_Channels']['channels'].keys()):
if not inlist(bot, channel.lower(), botchannels.SpiceBot_Channels['list'].keys()):
bot.osd("Channel name {} not valid.".format(channel))
return
topic = bot.memory['SpiceBot_Channels']['channels'][channel.lower()]['topic']
channel = bot.memory['SpiceBot_Channels']['channels'][channel.lower()]['name']
topic = botchannels.SpiceBot_Channels['list'][channel.lower()]['topic']
channel = botchannels.SpiceBot_Channels['list'][channel.lower()]['name']
bot.osd("Topic for {}: {}".format(channel, topic))
return
@ -90,7 +83,7 @@ def nickname_comand_channels(bot, trigger):
channel = trigger.sender
else:
channel = spicemanip.main(triggerargs, 1)
if not inlist(bot, channel.lower(), bot.memory['SpiceBot_Channels']['channels'].keys()):
if not inlist(bot, channel.lower(), botchannels.SpiceBot_Channels['list'].keys()):
bot.osd("Channel name {} not valid.".format(channel))
return
if not inlist(bot, channel.lower(), bot.channels.keys()):
@ -110,7 +103,7 @@ def nickname_comand_channels(bot, trigger):
# Users List
if commandused == 'users':
channel = spicemanip.main(triggerargs, 1)
if not inlist(bot, channel.lower(), bot.memory['SpiceBot_Channels']['channels'].keys()):
if not inlist(bot, channel.lower(), botchannels.SpiceBot_Channels['list'].keys()):
bot.osd("Channel name {} not valid.".format(channel))
return
if not inlist(bot, channel.lower(), bot.channels.keys()):

View File

@ -3,6 +3,4 @@
"""
from __future__ import unicode_literals, absolute_import, print_function, division
from .Channels import *
from .Channels_Command import *
from .Channels_Parsing import *

View File

@ -0,0 +1,121 @@
#!/usr/bin/env python
# coding=utf-8
from __future__ import unicode_literals, absolute_import, print_function, division
# sopel imports
import sopel.module
from sopel.config.types import StaticSection, ValidatedAttribute, ListAttribute
from sopel_modules.SpiceBot.Logs import botlogs
from sopel_modules.SpiceBot.Events import botevents
from sopel_modules.SpiceBot.Channels import botchannels
import spicemanip
import time
class SpiceBot_Channels_MainSection(StaticSection):
announcenew = ValidatedAttribute('announcenew', default=False)
joinall = ValidatedAttribute('joinall', default=False)
operadmin = ValidatedAttribute('operadmin', default=False)
chanignore = ListAttribute('chanignore')
def configure(config):
config.define_section("SpiceBot_Channels", SpiceBot_Channels_MainSection, validate=False)
config.SpiceBot_Channels.configure_setting('announcenew', 'SpiceBot_Channels Announce New Channels')
config.SpiceBot_Channels.configure_setting('joinall', 'SpiceBot_Channels JOIN New Channels')
config.SpiceBot_Channels.configure_setting('operadmin', 'SpiceBot_Channels OPER ADMIN MODE')
config.SpiceBot_Channels.configure_setting('chanignore', 'SpiceBot_Channels Ignore JOIN for channels')
def setup(bot):
botlogs.log('SpiceBot_Channels', "Starting setup procedure")
botevents.startup_add([botevents.BOT_CHANNELS])
bot.config.define_section("SpiceBot_Channels", SpiceBot_Channels_MainSection, validate=False)
@sopel.module.event('001')
@sopel.module.rule('.*')
def unkickable_bot(bot, trigger):
if bot.config.SpiceBot_Channels.operadmin:
bot.write(('SAMODE', bot.nick, '+q'))
@sopel.module.event('001')
@sopel.module.rule('.*')
def request_channels_list_initial(bot, trigger):
botchannels.bot_part_empty(bot)
botlogs.log('SpiceBot_Channels', "Sending Request for all server channels")
botchannels.channel_list_request(bot)
starttime = time.time()
# wait 60 seconds for initial population of information
while not botchannels.SpiceBot_Channels['InitialProcess']:
if time.time() - starttime >= 60:
botlogs.log('SpiceBot_Channels', "Initial Channel list populating Timed Out")
botchannels.SpiceBot_Channels['InitialProcess'] = True
else:
pass
foundchannelcount = len(botchannels.SpiceBot_Channels['list'].keys())
botlogs.log('SpiceBot_Channels', "Channel listing finished! " + str(foundchannelcount) + " channel(s) found.")
botchannels.join_all_channels(bot)
botchannels.chanadmin_all_channels(bot)
if "*" in botchannels.SpiceBot_Channels['list'].keys():
del botchannels.SpiceBot_Channels['list']["*"]
botchannels.bot_part_empty(bot)
botevents.trigger(bot, botevents.BOT_CHANNELS, "SpiceBot_Channels")
@sopel.module.event('321')
@sopel.module.rule('.*')
def watch_chanlist_start(bot, trigger):
botchannels.channel_list_recieve_start()
@sopel.module.event('322')
@sopel.module.rule('.*')
def watch_chanlist_populate(bot, trigger):
botchannels.channel_list_recieve_input(trigger)
@sopel.module.event('323')
@sopel.module.rule('.*')
def watch_chanlist_complete(bot, trigger):
botchannels.channel_list_recieve_finish()
@sopel.module.event(botevents.BOT_CHANNELS)
@sopel.module.rule('.*')
def trigger_channel_list_recurring(bot, trigger):
while True:
time.sleep(1800)
botchannels.bot_part_empty(bot)
oldlist = list(botchannels.SpiceBot_Channels['list'].keys())
botchannels.channel_list_request(bot)
while botchannels.channel_lock:
pass
newlist = [item.lower() for item in oldlist if item.lower() not in list(botchannels.SpiceBot_Channels['list'].keys())]
if "*" in newlist:
newlist.remove("*")
if len(newlist) and bot.config.SpiceBot_Channels.announcenew:
bot.osd(["The Following channel(s) are new:", spicemanip.main(newlist, 'andlist')], bot.channels.keys())
botchannels.join_all_channels(bot)
botchannels.chanadmin_all_channels(bot)
if "*" in botchannels.SpiceBot_Channels['list'].keys():
del botchannels.SpiceBot_Channels['list']["*"]
botchannels.bot_part_empty(bot)

View File

@ -0,0 +1,6 @@
# coding=utf-8
"""Event Parsing from IRC
"""
from __future__ import unicode_literals, absolute_import, print_function, division
from .Channels_Parsing import *

View File

@ -6,6 +6,7 @@ import sopel.module
import spicemanip
from sopel_modules.SpiceBot.Channels import botchannels
from sopel_modules.SpiceBot_SBTools import sopel_triggerargs, command_permissions_check, inlist
@ -41,7 +42,7 @@ def bot_command_process(bot, trigger):
target = spicemanip.main(triggerargs, 1)
if (target not in ['allchans', 'allnicks']
and not inlist(bot, target.lower(), bot.memory['SpiceBot_Channels']['channels'].keys())
and not inlist(bot, target.lower(), botchannels.SpiceBot_Channels['list'].keys())
and not inlist(bot, target.lower(), bot.users)):
bot.osd("Channel/nick name {} not valid.".format(target))
return

View File

@ -270,40 +270,6 @@ def channel_privs(bot, channel, privtype):
return privlist
def join_all_channels(bot):
if bot.config.SpiceBot_Channels.joinall:
for channel in bot.memory['SpiceBot_Channels']['channels'].keys():
if channel.startswith("#"):
if channel not in bot.channels.keys() and channel not in bot.config.SpiceBot_Channels.chanignore:
bot.write(('JOIN', bot.nick, bot.memory['SpiceBot_Channels']['channels'][channel]['name']))
if channel not in bot.channels.keys() and bot.config.SpiceBot_Channels.operadmin:
bot.write(('SAJOIN', bot.nick, bot.memory['SpiceBot_Channels']['channels'][channel]['name']))
def chanadmin_all_channels(bot):
# Chan ADMIN +a
for channel in bot.channels.keys():
if channel.startswith("#"):
if channel not in bot.config.SpiceBot_Channels.chanignore:
if bot.config.SpiceBot_Channels.operadmin:
if not bot.channels[channel].privileges[bot.nick] < sopel.module.ADMIN:
bot.write(('SAMODE', channel, "+a", bot.nick))
else:
bot.part(channel)
def channel_list_current(bot):
newlist = [item for item in bot.channels.keys() if item.lower() not in bot.memory['SpiceBot_Channels']['channels']]
for channel in newlist:
topic = bot.channels[channel].topic
bot.memory['SpiceBot_Channels']['channels'][str(channel).lower()] = dict()
bot.memory['SpiceBot_Channels']['channels'][str(channel).lower()]['name'] = str(channel)
bot.memory['SpiceBot_Channels']['channels'][str(channel).lower()]['topic'] = topic_compile(topic)
if "*" in bot.memory['SpiceBot_Channels']:
bot.memory['SpiceBot_Channels'].remove("*")
"""Environment Functions"""

View File

@ -6,6 +6,7 @@ import sopel.module
from sopel_modules.SpiceBot.Logs import botlogs
from sopel_modules.SpiceBot.Events import botevents
from sopel_modules.SpiceBot.Channels import botchannels
from sopel_modules.SpiceBot_SBTools import humanized_time
import time
@ -42,7 +43,7 @@ def bot_startup_monologue_commands(bot, trigger):
def bot_startup_monologue_channels(bot, trigger):
botcount = len(bot.channels.keys())
servercount = len(bot.memory['SpiceBot_Channels']['channels'].keys())
servercount = len(botchannels.SpiceBot_Channels['list'].keys())
bot.memory['SpiceBot_StartupMonologue'].append("I am in " + str(botcount) + " of " + str(servercount) + " channel(s) available on this server.")
botevents.trigger(bot, botevents.BOT_STARTUPMONOLOGUE_CHANNELS, "SpiceBot_StartupMonologue")