Raiza.dev/EliteBot
4
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity

Reformatting #2

Merged
ComputerTech merged 5 commits from :master into master 2024-02-18 15:26:21 +01:00
Conversation 1 Commits 5 Files changed 5 +42 -31
5 changed files with 42 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
plugins/commands.py
View file

@ -1,9 +1,10 @@
from src.plugin_base import PluginBase
from src.channel_manager import ChannelManager
import sys
class Plugin(PluginBase):
from src.channel_manager import ChannelManager
from src.plugin_base import PluginBase
class Plugin(PluginBase):
def handle_message(self, source_nick, channel, message):
message_parts = message.split()
self.channel_manager = ChannelManager()
@ -26,12 +27,11 @@ class Plugin(PluginBase):
self.bot.ircsend(f'PART {message_parts[1]}')
self.channel_manager.remove_channel(message_parts[1])
elif message_parts[0] == "!quit":
if len(message_parts) == 0:
quit_message = 'EliteBot!'
else:
quit_message = message[len(message_parts[0])+1:]
quit_message = message[len(message_parts[0]) + 1:]
self.bot.ircsend(f'QUIT :{quit_message}')
self.bot.ircsock.close()
self.bot.connected = False

1
requirements.txt
View file

@ -1 +1,2 @@
colorama==0.4.6
PyYAML==6.0.1

26
src/bot.py
View file

@ -1,19 +1,22 @@
#!/usr/bin/env python3
import importlib.util
import inspect
import json
import os
import socket
import ssl
import time
import json
import yaml
import inspect
import sys
import os
import importlib.util
import time
import yaml
from src.channel_manager import ChannelManager
from src.logger import Logger
from src.plugin_base import PluginBase
from src.sasl import handle_sasl, handle_authenticate, handle_903
class Bot:
def __init__(self, config_file):
self.config = self.load_config(config_file)
@ -51,11 +54,11 @@ class Bot:
def load_plugins(self):
self.plugins = []
plugin_folder = "./plugins"
plugin_folder = './plugins'
for filename in os.listdir(plugin_folder):
if filename.endswith('.py'):
filepath = os.path.join(plugin_folder, filename)
spec = importlib.util.spec_from_file_location("module.name", filepath)
spec = importlib.util.spec_from_file_location('module.name', filepath)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
for name, obj in inspect.getmembers(module):
@ -122,9 +125,10 @@ class Bot:
try:
self.ircsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if str(self.config["Connection"].get("Port"))[:1] == '+':
if str(self.config['Connection'].get('Port'))[:1] == '+':
context = ssl.create_default_context()
self.ircsock = context.wrap_socket(self.ircsock, server_hostname=self.config["Connection"].get("Hostname"))
self.ircsock = context.wrap_socket(self.ircsock,
server_hostname=self.config['Connection'].get('Hostname'))
port = int(self.config['Connection'].get('Port')[1:])
else:
port = int(self.config['Connection'].get('Port'))
@ -135,7 +139,7 @@ class Bot:
self.ircsock.connect_ex((self.config['Connection'].get('Hostname'), port))
self.ircsend(f'NICK {self.config["Connection"].get("Nick")}')
self.ircsend(f'USER {self.config["Connection"].get("Ident")} * * :{self.config["Connection"].get("Name")}')
if self.config["SASL"].get("UseSASL"):
if self.config['SASL'].get('UseSASL'):
self.ircsend('CAP REQ :sasl')
except Exception as e:
self.logger.error(f"Error establishing connection: {e}")

11
src/channel_manager.py
View file

@ -4,12 +4,13 @@ import json
import os
from os import path
class ChannelManager:
def __init__(self):
self.channels = self._load_channels()
def _load_channels(self):
os.makedirs("data", exist_ok=True)
os.makedirs('data', exist_ok=True)
if not path.exists('data/channels.json'):
with open('data/channels.json', 'w') as f:
json.dump([], f)
@ -18,10 +19,10 @@ class ChannelManager:
with open('data/channels.json', 'r') as f:
return json.load(f)
except json.JSONDecodeError as e:
print(f"Error decoding JSON: {e}")
print(f'Error decoding JSON: {e}')
return []
except Exception as e:
print(f"Error loading channels: {e}")
print(f'Error loading channels: {e}')
return []
def save_channel(self, channel):
@ -37,12 +38,12 @@ class ChannelManager:
self._write_channels()
def _write_channels(self):
os.makedirs("data", exist_ok=True)
os.makedirs('data', exist_ok=True)
try:
with open('data/channels.json', 'w') as f:
json.dump(self.channels, f)
except Exception as e:
print(f"Error saving channels: {e}")
print(f'Error saving channels: {e}')
def get_channels(self):
return self.channels

11
src/sasl.py
View file

@ -4,6 +4,7 @@ import base64
NULL_BYTE = '\x00'
ENCODING = 'UTF-8'
def handle_sasl(config, ircsend):
"""
Handles SASL authentication by sending an AUTHENTICATE command.
@ -14,6 +15,7 @@ def handle_sasl(config, ircsend):
"""
ircsend('AUTHENTICATE PLAIN')
def handle_authenticate(args, config, ircsend):
"""
Handles the AUTHENTICATE command response.
@ -24,12 +26,15 @@ def handle_authenticate(args, config, ircsend):
ircsend (function): Function to send IRC commands
"""
if args[0] == '+':
if "SASLNick" in config['SASL'] and "SASLPassword" in config['SASL']:
authpass = f"{config['SASL']['SASLNick']}{NULL_BYTE}{config['SASL']['SASLNick']}{NULL_BYTE}{config['SASL']['SASLPassword']}"
if 'SASLNick' in config['SASL'] and 'SASLPassword' in config['SASL']:
authpass = (f'{config["SASL"]["SASLNick"]}{NULL_BYTE}'
f'{config["SASL"]["SASLNick"]}{NULL_BYTE}'
f'{config["SASL"]["SASLPassword"]}')
ap_encoded = base64.b64encode(authpass.encode(ENCODING)).decode(ENCODING)
ircsend(f'AUTHENTICATE {ap_encoded}')
else:
raise KeyError("SASLNICK and/or SASLPASS not found in config")
raise KeyError('SASLNICK and/or SASLPASS not found in config')
def handle_903(ircsend):
"""