# Pool Guy 🏊♂️
A lightweight Twitch bot framework with event subscription and alert handling capabilities.
[](https://www.python.org/downloads/)
[](https://www.gnu.org/licenses/gpl-3.0)
## 🚀 Features
- Event subscription handling
- Custom alert system
- Twitch chat command integration
- WebSocket-based real-time updates
- Support for multiple storage backends (MongoDB, SQLite)
- Customizable command prefix system
- Rate limiting for commands
## 🛠️ Quick Setup
### Install directly from git:
```bash
pip install git+https://github.com/s4w3d0ff/pool-guy.git
```
### Create a configuration file (e.g., config.json):
```json
{
"http_config": {
"client_id": "YOUR_CLIENT_ID",
"client_secret": "YOUR_CLIENT_SECRET",
"redirect_uri": "http://localhost:5000/callback",
"scopes": [
"user:read:chat",
"user:write:chat"
]
},
"ws_config": {
"channels": {"channel.chat.message": [null]},
"queue_skip": ["channel.chat.message"],
"storage_type": "json"
},
"max_retries": 30,
"retry_delay": 10,
"login_browser": {
"chrome": "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe"
}
}
```
### Simple Command Bot:
```python
from poolguy.utils import asyncio
from poolguy.utils import loadJSON, ctxt
from poolguy import CommandBot, Alert, ColorLogger, cmd_rate_limit
logger = ColorLogger(__name__)
class ChannelChatMessageAlert(Alert):
"""channel.chat.message"""
async def process(self):
logger.debug(f'{self.data}')
text = self.data["message"]["text"]
user = {
"user_id": self.data["chatter_user_id"],
"username": self.data["chatter_user_name"]
}
channel = {
"broadcaster_id": self.data["broadcaster_user_id"],
"broadcaster_user_name": self.data["broadcaster_user_name"]
}
logger.info(f'[Chat] {user["username"]}: {text}', 'purple')
if str(user["user_id"]) == str(self.bot.http.user_id):
logger.debug(f'Own message ignored')
return
if self.data["source_broadcaster_user_id"]:
logger.debug(f'Shared chat message ignored')
return
await self.bot.command_check(text, user, channel)
class ExampleBot(CommandBot):
def __init__(self, *args, **kwargs):
"""
# Fetch sensitive data from environment variables
import os
client_id = os.getenv("CLIENT_ID")
client_secret = os.getenv("CLIENT_SECRET")
if not client_id or not client_secret:
raise ValueError("Environment variables CLIENT_ID and CLIENT_SECRET are required")
kwargs['http_config']['client_id'] = client_id
kwargs['http_config']['client_secret'] = client_secret
"""
super().__init__(*args, **kwargs)
async def send_chat(self, message, channel_id=None):
r = await self.http.sendChatMessage(message, channel_id)
if not r[0]['is_sent']:
logger.error(f"Message not sent! Reason: {r[0]['drop_reason']}")
@cmd_rate_limit(calls=1, period=10)
async def cmd_hi(self, user, channel, args):
await self.send_chat(f"Hi, @{user['username']}", channel["broadcaster_id"])
async def my_loop(self):
logger.warning(f'my_loop started')
while self.ws.connected:
await asyncio.sleep(10)
logger.warning(f'my_loop stopped')
async def after_login(self):
await self.add_task(self.my_loop)
if __name__ == '__main__':
import logging
fmat = ctxt('%(asctime)s', 'yellow', style='d') + '-%(levelname)s-' + ctxt('[%(name)s]', 'purple', style='d') + ctxt(' %(message)s', 'green', style='d')
logging.basicConfig(
format=fmat,
datefmt="%I:%M:%S%p",
level=logging.INFO
)
cfg = loadJSON('config.json')
bot = ExampleBot(**cfg, alert_objs={'channel.chat.message': ChannelChatMessageAlert})
asyncio.run(bot.start())
```
More fleshed out example: https://github.com/s4w3d0ff/deezbot
Raw data
{
"_id": null,
"home_page": "https://github.com/s4w3d0ff/pool-guy",
"name": "poolguy",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "twitch, bot, eventsub, websocket, async, alerts, streaming, poolguy",
"author": "s4w3d0ff",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/c7/7a/017cb551c8d7b003f30ba7eb5b44f6456d64852e9af6aaf738a530a16854/poolguy-0.1.3.tar.gz",
"platform": "any",
"description": "# Pool Guy \ud83c\udfca\u200d\u2642\ufe0f\n\nA lightweight Twitch bot framework with event subscription and alert handling capabilities.\n\n[](https://www.python.org/downloads/)\n[](https://www.gnu.org/licenses/gpl-3.0)\n\n## \ud83d\ude80 Features\n\n- Event subscription handling\n- Custom alert system\n- Twitch chat command integration\n- WebSocket-based real-time updates\n- Support for multiple storage backends (MongoDB, SQLite)\n- Customizable command prefix system\n- Rate limiting for commands\n\n## \ud83d\udee0\ufe0f Quick Setup\n\n\n### Install directly from git:\n```bash\npip install git+https://github.com/s4w3d0ff/pool-guy.git\n```\n\n### Create a configuration file (e.g., config.json):\n```json\n{\n \"http_config\": {\n \"client_id\": \"YOUR_CLIENT_ID\",\n \"client_secret\": \"YOUR_CLIENT_SECRET\",\n \"redirect_uri\": \"http://localhost:5000/callback\",\n \"scopes\": [\n \"user:read:chat\",\n \"user:write:chat\"\n ]\n },\n \"ws_config\": {\n \"channels\": {\"channel.chat.message\": [null]}, \n \"queue_skip\": [\"channel.chat.message\"], \n \"storage_type\": \"json\"\n },\n \"max_retries\": 30,\n \"retry_delay\": 10,\n \"login_browser\": {\n \"chrome\": \"C:\\\\Program Files\\\\Google\\\\Chrome\\\\Application\\\\chrome.exe\"\n }\n}\n```\n\n### Simple Command Bot:\n```python\nfrom poolguy.utils import asyncio\nfrom poolguy.utils import loadJSON, ctxt\nfrom poolguy import CommandBot, Alert, ColorLogger, cmd_rate_limit\n\nlogger = ColorLogger(__name__)\n\nclass ChannelChatMessageAlert(Alert):\n \"\"\"channel.chat.message\"\"\"\n async def process(self):\n logger.debug(f'{self.data}')\n text = self.data[\"message\"][\"text\"]\n user = {\n \"user_id\": self.data[\"chatter_user_id\"], \n \"username\": self.data[\"chatter_user_name\"]\n }\n channel = {\n \"broadcaster_id\": self.data[\"broadcaster_user_id\"],\n \"broadcaster_user_name\": self.data[\"broadcaster_user_name\"]\n }\n logger.info(f'[Chat] {user[\"username\"]}: {text}', 'purple')\n if str(user[\"user_id\"]) == str(self.bot.http.user_id):\n logger.debug(f'Own message ignored')\n return\n if self.data[\"source_broadcaster_user_id\"]:\n logger.debug(f'Shared chat message ignored')\n return\n await self.bot.command_check(text, user, channel)\n\nclass ExampleBot(CommandBot):\n def __init__(self, *args, **kwargs):\n \"\"\"\n # Fetch sensitive data from environment variables\n import os\n client_id = os.getenv(\"CLIENT_ID\")\n client_secret = os.getenv(\"CLIENT_SECRET\")\n if not client_id or not client_secret:\n raise ValueError(\"Environment variables CLIENT_ID and CLIENT_SECRET are required\")\n kwargs['http_config']['client_id'] = client_id\n kwargs['http_config']['client_secret'] = client_secret\n \"\"\"\n super().__init__(*args, **kwargs)\n\n async def send_chat(self, message, channel_id=None):\n r = await self.http.sendChatMessage(message, channel_id)\n if not r[0]['is_sent']:\n logger.error(f\"Message not sent! Reason: {r[0]['drop_reason']}\")\n\n @cmd_rate_limit(calls=1, period=10)\n async def cmd_hi(self, user, channel, args):\n await self.send_chat(f\"Hi, @{user['username']}\", channel[\"broadcaster_id\"])\n\n async def my_loop(self):\n logger.warning(f'my_loop started')\n while self.ws.connected:\n await asyncio.sleep(10)\n logger.warning(f'my_loop stopped')\n\n async def after_login(self):\n await self.add_task(self.my_loop)\n\n\nif __name__ == '__main__':\n import logging\n fmat = ctxt('%(asctime)s', 'yellow', style='d') + '-%(levelname)s-' + ctxt('[%(name)s]', 'purple', style='d') + ctxt(' %(message)s', 'green', style='d')\n logging.basicConfig(\n format=fmat,\n datefmt=\"%I:%M:%S%p\",\n level=logging.INFO\n )\n cfg = loadJSON('config.json')\n bot = ExampleBot(**cfg, alert_objs={'channel.chat.message': ChannelChatMessageAlert})\n asyncio.run(bot.start())\n \n```\nMore fleshed out example: https://github.com/s4w3d0ff/deezbot\n",
"bugtrack_url": null,
"license": "GNU General Public License v3 (GPLv3)",
"summary": "A Twitch bot framework with event subscription and alert handling capabilities",
"version": "0.1.3",
"project_urls": {
"Homepage": "https://github.com/s4w3d0ff/pool-guy"
},
"split_keywords": [
"twitch",
" bot",
" eventsub",
" websocket",
" async",
" alerts",
" streaming",
" poolguy"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "65853d7f9bdda90caa7dfc360c7c91aef926ee68971db449d859307f7ca617e0",
"md5": "0c4945b8d4cdf046275a4a2dbdcf38b4",
"sha256": "cc8ffde37a126d79fb58be96c394f3e2d47e72e049158c5dac1a432bda10c3a1"
},
"downloads": -1,
"filename": "poolguy-0.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0c4945b8d4cdf046275a4a2dbdcf38b4",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 37249,
"upload_time": "2025-02-01T21:34:55",
"upload_time_iso_8601": "2025-02-01T21:34:55.058933Z",
"url": "https://files.pythonhosted.org/packages/65/85/3d7f9bdda90caa7dfc360c7c91aef926ee68971db449d859307f7ca617e0/poolguy-0.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "c77a017cb551c8d7b003f30ba7eb5b44f6456d64852e9af6aaf738a530a16854",
"md5": "321ca809a6198c26db1b8f4222df92b7",
"sha256": "ce39b631c63c51c0d602fe015497265c44e0a41733e61c3f8ad8851825db06d2"
},
"downloads": -1,
"filename": "poolguy-0.1.3.tar.gz",
"has_sig": false,
"md5_digest": "321ca809a6198c26db1b8f4222df92b7",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 36568,
"upload_time": "2025-02-01T21:34:56",
"upload_time_iso_8601": "2025-02-01T21:34:56.904362Z",
"url": "https://files.pythonhosted.org/packages/c7/7a/017cb551c8d7b003f30ba7eb5b44f6456d64852e9af6aaf738a530a16854/poolguy-0.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-02-01 21:34:56",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "s4w3d0ff",
"github_project": "pool-guy",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "poolguy"
}
JSON
