def parse_arguments() -> argparse.Namespace:
arg_parser = argparse.ArgumentParser(
prog="aptbot",
- description="A chat bot for twitch.tv"
+ description="A chat bot for twitch.tv",
)
arg_parser.add_argument(
"-a",
"--add-account",
type=str,
- help=f"Add an account to connect with the bot"
+ help=f"Add an account to connect with the bot",
)
arg_parser.add_argument(
"-d",
"--disable-account",
type=str,
- help=f"Disable an account from the bot"
+ help=f"Disable an account from the bot",
)
arg_parser.add_argument(
"-s",
"--send-message",
type=str,
- help=f"Send a message to skgyorugo"
+ help=f"Send a message to skgyorugo",
)
arg_parser.add_argument(
"--enable",
default=False,
action="store_true",
- help=f"Enable the bot"
+ help=f"Enable the bot",
)
arg_parser.add_argument(
"--disable",
default=False,
action="store_true",
- help=f"Disable the bot"
+ help=f"Disable the bot",
)
arg_parser.add_argument(
"--update",
default=False,
action="store_true",
- help=f"Update the bot"
+ help=f"Update the bot",
)
return arg_parser.parse_args()
import logging
import re
import socket
+import sys
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
self._nick = nick
self._oauth_token = oauth_token
self._connected_channels = set()
+ self._buffered_messages = []
def _send_command(self, command: str):
if "PASS" not in command:
- logger.debug(f"< {command}")
+ logger.info(f"< {command}")
self._irc.send((command + "\r\n").encode())
- def connect(self):
+ def connect(self) -> bool:
+ self._connect()
+ connected = self._connected()
+ return connected
+
+ def _connect(self) -> None:
self._irc = socket.socket()
self._irc.connect((self._server, self._port))
+ logger.debug("Connecting...")
self._send_command(f"PASS oauth:{self._oauth_token}")
self._send_command(f"NICK {self._nick}")
self._send_command(
@staticmethod
def _parse_message(received_msg: str) -> Message:
split = re.search(
- r"(?:(?:@(.+))\s)?:(?:(?:(\w+)!\w+@\w+\.)?.+)\s(\w+)\s\#(\w+)\s:?(.+)?",
+ r"(?:@(.+)\s)?:(?:(?:(\w+)!\w+@\w+\.)?.+)\s(\w+)\s(?:\#(\w+)|\*)\s:?(.+)?",
received_msg,
)
tag_value = Bot._replace_escaped_space_in_tags(tag.split("=")[1])
tags[tag_name] = " ".join(tag_value.split())
- nick = split[2]
+ nick = split[2] if split[2] else ""
try:
command = Commands[split[3]]
except KeyError:
return Message()
- channel = split[4]
- value = " ".join(split[5].split())
+ channel = split[4] if split[4] else ""
+ try:
+ value = " ".join(split[5].split())
+ except AttributeError:
+ value = ""
return Message(
tags=tags,
)
def _handle_message(self, received_msg: str) -> Message:
- logger.debug(received_msg)
+ logger.info(f"> {received_msg}")
if received_msg == "PING :tmi.twitch.tv":
self._send_command("PONG :tmi.twitch.tv")
return Message()
return Message()
return Bot._parse_message(received_msg)
- def receive_messages(self) -> list[Message]:
- messages = []
- while True:
+ def _receive_messages(self) -> bytes:
+ for _ in range(10):
try:
received_msgs = self._irc.recv(2048)
except ConnectionResetError as e:
self._restart_connection()
else:
break
- for received_msgs in received_msgs.decode("utf-8").split("\r\n"):
- messages.append(self._handle_message(received_msgs))
+ else:
+ logger.error("Unable to connect to twitch. Exiting")
+ sys.exit(1)
+ return received_msgs
+
+ def _connected(self) -> bool:
+ received_msgs = self._receive_messages()
+ for received_msg in received_msgs.decode("utf-8").split("\r\n"):
+ self._buffered_messages.append(self._handle_message(received_msg))
+ if self._buffered_messages[0] == Message(
+ {},
+ "",
+ Commands.NOTICE,
+ "",
+ "Login authentication failed",
+ ):
+ logger.debug(f"Not connected")
+ return False
+ logger.debug(f"Connected")
+ return True
+
+ def get_messages(self) -> list[Message]:
+ messages = []
+ messages.extend(self._buffered_messages)
+ self._buffered_messages = []
+ received_msgs = self._receive_messages()
+ for received_msg in received_msgs.decode("utf-8").split("\r\n"):
+ messages.append(self._handle_message(received_msg))
return messages
def disconnect(self) -> None:
+ logger.debug("Disconnecting...")
self._irc.close()
- def _restart_connection(self) -> None:
+ def _restart_connection(self):
self.disconnect()
time.sleep(5)
- self.connect()
+ self._connect()
self.join_channels(self._connected_channels)
time.sleep(2)
print("Your OS is not supported", file=sys.stderr)
sys.exit(1)
+MAIN_FILE_NAME = "main.py"
+
PORT = 26538
LOCALHOST = "127.0.0.1"
from . import args_logic
from .args import parse_arguments
from .bot import Bot, Message
-from .constants import CONFIG_LOGS, CONFIG_PATH, LOCALHOST, LOGGING_DICT, PORT
+from .constants import (
+ CONFIG_LOGS,
+ CONFIG_PATH,
+ LOCALHOST,
+ LOGGING_DICT,
+ MAIN_FILE_NAME,
+ PORT,
+)
logging.config.dictConfig(LOGGING_DICT)
logger = logging.getLogger(__name__)
def handle_message(bot: Bot, modules: dict[str, ModuleType]):
+ logger.debug("in handle_message thread")
while True:
- messages = bot.receive_messages()
+ messages = bot.get_messages()
for message in messages:
if not message.channel:
continue
]
channels = filter(lambda x: not x.startswith("."), channels)
for channel in channels:
- account_path = os.path.join(CONFIG_PATH, f"{channel}")
+ account_path = os.path.join(CONFIG_PATH, channel)
sys.path.append(account_path)
- module_path = os.path.join(account_path, "main.py")
+ module_path = os.path.join(account_path, MAIN_FILE_NAME)
spec = importlib.util.spec_from_file_location(
"main",
module_path,
)
if not spec or not spec.loader:
- print("Problem loading spec")
+ logger.warning(f"Problem loading for {channel}")
sys.path.remove(account_path)
continue
module = importlib.util.module_from_spec(spec)
def initialize(bot: Bot):
+ logger.debug("Initializing...")
channels = [
c
for c in os.listdir(CONFIG_PATH)
if NICK and OAUTH:
bot = Bot(NICK, OAUTH)
else:
- print(
- "Please set the environment variables:\nAPTBOT_NICK\nAPTBOT_PASS",
- file=sys.stderr,
+ logger.error(
+ "The environment variables:\nAPTBOT_NICK\nAPTBOT_PASS\nare not set."
)
time.sleep(3)
sys.exit(1)
- bot.connect()
+ if not bot.connect():
+ logger.error("Twitch couldn't authenticate your credentials")
+ time.sleep(3)
+ sys.exit(1)
modules = {}
message_loop = Thread(
target=start,
time.sleep(1)
-def send(func):
- def inner(*args, **kwargs):
- s = socket.socket()
- s.connect((LOCALHOST, PORT))
- func(s, *args, **kwargs)
- s.close()
-
- return inner
-
-
def main():
argsv = parse_arguments()
os.makedirs(CONFIG_PATH, exist_ok=True)
+attrs==21.4.0
black==22.3.0
click==8.1.3
flake8==4.0.1
flake8-black==0.3.3
+iniconfig==1.1.1
mccabe==0.6.1
mypy-extensions==0.4.3
+packaging==21.3
pathspec==0.9.0
platformdirs==2.5.2
+pluggy==1.0.0
+py==1.11.0
pycodestyle==2.8.0
pyflakes==2.4.0
+pyparsing==3.0.9
+pytest==7.1.2
python-dotenv==0.20.0
tomli==2.0.1
urllib3==1.26.9
-import os
-import urllib3
import json
-from datetime import datetime
-from typing import Optional, Union
+import os
from dataclasses import dataclass
+from datetime import datetime
from enum import Enum
-# from dotenv import load_dotenv
+from typing import Optional, Union
+
+import urllib3
+from dotenv import load_dotenv
-# load_dotenv()
+load_dotenv()
NICK = os.getenv("APTBOT_NICK")
OAUTH = os.getenv("APTBOT_OAUTH")
projects / personal / aptbot.git / commitdiff