# aptbot
+--------
A chatbot for twitch.tv
+
+## Dependencies
+
+--------------
+
+* Python (any >=3.7 version should work)
+* python-dotenv
+* urllib3
+
+## Install
+
+----------
+
+It is highly recommended you install and run aptbot in a virtual environment.
+
+Clone this repository `git clone https://github.com/atheridis/aptbot.git`,
+change to the directory `cd aptbot`, then install the package `pip install .`.
+
+## First Steps
+
+----------
+
+### Adding an account
+
+After installing, you can add an account using `aptbot --add-account "account_name"`.
+A directory will be created in `~/.config/aptbot/` on Linux
+or `%APPDATA%\aptbot\accounts\` on Windows with the twitch name of that account.
+
+Each account directory should contain a `main.py` file
+with the most minimal code being:
+
+```python
+from aptbot import Bot, Message, Commands
+
+# Runs at the beginning, when the account connects.
+# Can be used for an infinite loop within the account,
+# so the bot can send messages, even when chat isn't moving.
+def start(bot: Bot, message: Message, stop_event):
+ pass
+
+# Gets ran every time the IRC channel sends a message.
+# This can either be a message from a user
+# a raid, when a mod deletes a message, etc.
+def main(bot: Bot, message: Message):
+ pass
+```
+
+When you add an account using `aptbot --add-account "account_name"`
+or `aptbot -a "account_name"`,
+the following python file called `main.py` will be created:
+
+```python
+import time
+from threading import Event
+
+from aptbot import Bot, Commands, Message
+
+
+# Starts when the bot is enabled
+# Sends the message "Hello, world!" every 2 minutes to the channel
+# stop_event is set to True, when you disable the account with aptbot -d "account_name"
+def start(bot: Bot, message: Message, stop_event: Event):
+ while not stop_event.is_set():
+ bot.send_message(message.channel, "Hello, world!")
+ time.sleep(120)
+
+
+def main(bot: Bot, message: Message):
+ # Check whether the message sent is a message by a user in chat
+ # and not some notification.
+ if message.command == Commands.PRIVMSG:
+ # Check the content of the message and if the first word is '!hello'
+ # send a reply by creating a new thread.
+ # You can also use `message.nick` instead of `message.tags['display-name']`
+ # but then the message sent back
+ # will contain the name of the user in all lowercase
+ if message.value.split()[0] == "!hello":
+ bot.send_message(
+ message.channel,
+ f"hello {message.tags['display-name']}",
+ reply=message.tags["id"],
+ )
+```
+
+### Activating your bot
+
+Once you have an account set up you can enable the bot by using `aptbot --enable`.
+To stop the bot from running use `aptbot --disable` on another terminal.
+
+After testing you would want to run `aptbot --enable` as a daemon, the easiest way
+to do that on Linux is to use `nohup aptbot --enable &`. This will leave a `nohup.out`
+file in the directory you're in, which may not be ideal. If you wish to have no output
+other than the logs provided in `~/.cache/aptbot/logs/aptbot.log` you may use
+`nohup aptbot --enable </dev/null >/dev/null 2>&1 &`. You are now free to control
+aptbot through any terminal. Type `aptbot --help` to see all available commands.
+
+### More than one file
+
+You can import modules from the same directory that the `main.py` files are in,
+but if you want `aptbot --update` to work on them you will need to reload the modules
+due to how python handles module imports. Some examples will be coming soon.
+
+## BUGS
+
+-------
+
+* Editing any of the directory names in `~/.config/aptbot/`
+or `%APPDATA%\aptbot\accounts\` while the bot is running, may cause weird behaviour.
def leave_channel(self, channel: str):
self._send_command(f"{Commands.PART.value} #{channel}")
- self._connected_channels.remove(channel)
+ try:
+ self._connected_channels.remove(channel)
+ except KeyError as e:
+ logger.exception("Account isn't enabled")
+ logger.exception(e)
def send_message(self, channel: str, text: Union[list[str], str], reply=None):
if reply:
import sys
import time
import traceback
-from threading import Thread
+from dataclasses import dataclass
+from threading import Event, Thread
from types import ModuleType
from dotenv import load_dotenv
from . import args_logic
from .args import parse_arguments
from .bot import Bot, Message
-from .constants import (
- CONFIG_LOGS,
- CONFIG_PATH,
- LOCALHOST,
- LOGGING_DICT,
- MAIN_FILE_NAME,
- PORT,
-)
+from .constants import (CONFIG_LOGS, CONFIG_PATH, LOCALHOST, LOGGING_DICT,
+ MAIN_FILE_NAME, PORT)
logging.config.dictConfig(LOGGING_DICT)
logger = logging.getLogger(__name__)
load_dotenv()
-def handle_message(bot: Bot, modules: dict[str, ModuleType]):
- logger.debug("in handle_message thread")
+@dataclass(frozen=True)
+class Channel:
+ name: str
+ module: ModuleType
+ thread: Thread
+ stop_event: Event
+
+ def __hash__(self):
+ return hash(self.name)
+
+ def __eq__(self, other):
+ return self.name == other.name
+
+
+def handle_message(bot: Bot, channels: set[Channel]):
while True:
messages = bot.get_messages()
for message in messages:
if not message.channel:
continue
- try:
- method = Thread(
- target=modules[message.channel].main,
- args=(
- bot,
- message,
- ),
- )
- except KeyError:
- pass
- else:
- method.daemon = True
- method.start()
+ for channel in channels:
+ if channel.name != message.channel:
+ continue
+ Thread(
+ target=channel.module.main, args=(bot, message), daemon=True
+ ).start()
+ break
-def start(bot: Bot, modules: dict[str, ModuleType]):
- load_modules(modules)
+def run_start(channels: set[Channel]):
+ for channel in channels:
+ channel.thread.start()
+
+
+def disable_channel(channel_name: str, channels: set[Channel]):
+ for channel in channels:
+ if channel.name != channel_name:
+ continue
+ channel.stop_event.set()
+ logger.debug(f"Event set for {channel}")
+ channels.remove(channel)
+ logger.debug(f"{channel} removed")
+ break
+
+
+def enable(bot: Bot, channels: set[Channel]):
+ load_modules(bot, channels)
message_handler_thread = Thread(
target=handle_message,
args=(
bot,
- modules,
+ channels,
),
+ daemon=True,
)
- message_handler_thread.daemon = True
message_handler_thread.start()
- for channel in modules:
- update_channel = Thread(
- target=modules[channel].start,
- args=(
- bot,
- Message({}, "", None, channel, ""),
- ),
- )
- update_channel.daemon = True
- update_channel.start()
-def load_modules(modules: dict[str, ModuleType]):
- modules.clear()
- channels = [
+def load_modules(bot: Bot, channels: set[Channel]):
+ old_channels = channels.copy()
+ channels.clear()
+ channel_names = [
c
for c in os.listdir(CONFIG_PATH)
if os.path.isdir(os.path.join(CONFIG_PATH, c))
]
- channels = filter(lambda x: not x.startswith("."), channels)
- for channel in channels:
- account_path = os.path.join(CONFIG_PATH, channel)
- sys.path.append(account_path)
- module_path = os.path.join(account_path, MAIN_FILE_NAME)
+ channel_names = filter(lambda x: not x.startswith("."), channel_names)
+ for channel_name in channel_names:
+ channel_path = os.path.join(CONFIG_PATH, channel_name)
+ sys.path.append(channel_path)
+ module_path = os.path.join(channel_path, MAIN_FILE_NAME)
spec = importlib.util.spec_from_file_location(
"main",
module_path,
)
if not spec or not spec.loader:
- logger.warning(f"Problem loading for {channel}")
- sys.path.remove(account_path)
+ logger.warning(f"Problem loading for {channel_path}")
+ sys.path.remove(channel_path)
continue
module = importlib.util.module_from_spec(spec)
try:
logger.exception(f"Problem Loading Module: {e}")
logger.exception(traceback.format_exc())
else:
- modules[channel] = module
- sys.path.remove(account_path)
+ for channel in old_channels:
+ if channel_name != channel.name:
+ continue
+ stop_event = channel.stop_event
+ thread = channel.thread
+ logger.debug(
+ f"Copied stop_event and thread for account: {channel.name}"
+ )
+ break
+ else:
+ stop_event = Event()
+ thread = Thread(
+ target=module.start,
+ args=(bot, Message(channel=channel_name), stop_event),
+ daemon=True,
+ )
+ logger.debug(
+ f"Created stop event and thread for account: {channel_name}"
+ )
+ channels.add(
+ Channel(
+ name=channel_name,
+ module=module,
+ thread=thread,
+ stop_event=stop_event,
+ ),
+ )
+ sys.path.remove(channel_path)
+ new_channels = channels.difference(old_channels)
+ run_start(new_channels)
def initialize(bot: Bot):
logger.error("Twitch couldn't authenticate your credentials")
time.sleep(3)
sys.exit(1)
- modules = {}
+ initialize(bot)
+ channels: set[Channel] = set()
message_loop = Thread(
- target=start,
+ target=enable,
args=(
bot,
- modules,
+ channels,
),
+ daemon=True,
)
- message_loop.daemon = True
message_loop.start()
s = socket.socket()
s.bind((LOCALHOST, PORT))
s.listen(5)
- initialize(bot)
while True:
c, _ = s.accept()
s.close()
sys.exit()
elif args_logic.BotCommands.UPDATE.value in command:
- load_modules(modules)
+ load_modules(bot, channels)
elif args_logic.BotCommands.PART.value in command:
+ disable_channel(channel, channels)
bot.leave_channel(channel)
time.sleep(1)
+import time
+from threading import Event
+
from aptbot import Bot, Commands, Message
-def start(bot: Bot, message: Message):
- pass
+# Starts when the bot is enabled
+# Sends the message "Hello, world!" every 2 minutes to the channel
+# stop_event is set to True, when you disable the account with aptbot -d "account_name"
+def start(bot: Bot, message: Message, stop_event: Event):
+ while not stop_event.is_set():
+ bot.send_message(message.channel, "Hello, world!")
+ time.sleep(120)
def main(bot: Bot, message: Message):
Add an account using `aptbot --add-account "account_name"`.
A directory will be created in `~/.config/aptbot/` on Linux
-or `%APPDATA%\aptbot\` on Windows with the twitch id of that account.
+or `%APPDATA%\aptbot\` on Windows with the twitch name of that account.
The contents of each example directory here should mimic the contents of the
-directory with the twitch id.
+directory with the twitch name.
Each account directory should contain a `main.py` file
with the most minimal code being:
```python
from aptbot import Bot, Message, Commands
-# Gets ran at the beginning, when the account connects.
+# Runs at the beginning, when the account connects.
# Can be used for an infinite loop within the account,
# so the bot send messages, even when chat is dead.
def start(bot: Bot, message: Message):
setuptools.setup(
name="aptbot",
- version="0.2.0",
+ version="0.2.1",
author="Georgios Atheridis",
author_email="atheridis@tutamail.com",
description="A chatbot for twitch.tv",
"python-dotenv",
"urllib3",
],
- python_requires=">=3.9",
+ python_requires=">=3.7",
)
projects / personal / aptbot.git / commitdiff