Skip to content

Commit

Permalink
Added new announce_join and ascii_cog bot_features
Browse files Browse the repository at this point in the history
  • Loading branch information
@minimike86
minimike86 committed Oct 29, 2024
1 parent fa5636a commit 6fb10f7
Show file tree
Hide file tree
Showing 4 changed files with 123 additions and 58 deletions.
7 changes: 6 additions & 1 deletion tests/config/test_bot_config.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,14 @@ def test_config(monkeypatch):
assert bot_initial_channels[0].get("login") == "msec"

# Bot Features Vars
assert BOT_CONFIG.get("bot_features").get("enable_commands") is False
assert BOT_CONFIG.get("bot_features").get("announce_join") is True
assert BOT_CONFIG.get("bot_features").get("enable_commands") is True
assert BOT_CONFIG.get("bot_features").get("enable_psclient") is False
assert BOT_CONFIG.get("bot_features").get("enable_esclient") is False
assert (
BOT_CONFIG.get("bot_features").get("cogs").get("ascii_cog").get("enable_ascii_cog")
is True
)
assert (
BOT_CONFIG.get("bot_features").get("cogs").get("rce_cog").get("enable_rce_cog")
is False
Expand Down
6 changes: 5 additions & 1 deletion twitchrce/config/bot_config.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,10 +56,14 @@ def get_bot_config(self) -> dict:
"bot_initial_channels": self.BOT_INITIAL_CHANNELS,
},
"bot_features": {
"enable_commands": False,
"announce_join": True,
"enable_commands": True,
"enable_psclient": False,
"enable_esclient": False,
"cogs": {
"ascii_cog": {
"enable_ascii_cog": True,
},
"rce_cog": {
"enable_rce_cog": False,
"cmd_allow_list": self.CMD_ALLOW_LIST,
Expand Down
143 changes: 92 additions & 51 deletions twitchrce/custom_bot.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,10 @@ async def event_channel_joined(channel):
logger.info(
f"{Fore.LIGHTWHITE_EX}Connected_channels: {Fore.LIGHTCYAN_EX}{self.connected_channels}{Fore.LIGHTWHITE_EX}!{Style.RESET_ALL}"
)
if self.config.get_bot_config().get("bot_features").get("announce_join"):
await channel.send(
f"has joined the chat! PowerUpL EntropyWins PowerUpR"
)
if self.config.get_bot_config().get("bot_features").get("enable_esclient"):
# Side effect of joining channel it should start listening to event subscriptions
broadcasters: List[User] = await self.fetch_users(names=[channel.name])
Expand All @@ -121,15 +125,15 @@ async def event_channel_joined(channel):

@self.event()
async def event_join(channel, user):
logger.info(
f"{Fore.MAGENTA}JOIN{Fore.WHITE} is received from Twitch for user {Fore.CYAN}{user.name}{Fore.WHITE} in channel {Fore.CYAN}{channel}{Fore.WHITE}!{Style.RESET_ALL}"
logger.debug(
f"{Fore.MAGENTA}JOIN{Fore.WHITE} is received from Twitch for user {Fore.CYAN}{user.name}{Fore.WHITE} in channel {Fore.CYAN}{channel.name}{Fore.WHITE}!{Style.RESET_ALL}"
)
pass

@self.event()
async def event_part(user):
logger.info(
f"{Fore.MAGENTA}PART{Fore.WHITE} is received from Twitch for user {Fore.CYAN}{user.name}{Fore.WHITE} in channel {Fore.CYAN}{channel}{Fore.WHITE}!{Style.RESET_ALL}"
logger.debug(
f"{Fore.MAGENTA}PART{Fore.WHITE} is received from Twitch for user {Fore.CYAN}{user.name}{Fore.WHITE} in channel {Fore.CYAN}{channel.name}{Fore.WHITE}!{Style.RESET_ALL}"
)
pass

Expand Down Expand Up @@ -170,8 +174,11 @@ async def event_eventsub_notification_followV2(
) -> None:
"""event triggered when someone follows the channel"""
logger.info(
f"{Fore.RED}[{payload.data.broadcaster.name}]{Fore.BLUE}[Follow]{Fore.RED}[EventSub]: "
f"{payload.data.user.name} [{payload.data.user.id}]{Style.RESET_ALL}"
f"{Fore.LIGHTWHITE_EX}[{Fore.LIGHTRED_EX}{payload.data.broadcaster.name}{Fore.LIGHTWHITE_EX}]"
f"{Fore.LIGHTBLUE_EX}[{Fore.LIGHTRED_EX}Follow{Fore.LIGHTWHITE_EX}]"
f"{Fore.LIGHTBLUE_EX}[{Fore.YELLOW}EventSub{Fore.LIGHTWHITE_EX}]: "
f"{Fore.LIGHTYELLOW_EX}{payload.data.user.name} {Fore.LIGHTWHITE_EX}"
f"[{Fore.LIGHTYELLOW_EX}{payload.data.user.id}{Fore.LIGHTWHITE_EX}]{Style.RESET_ALL}"
)
await self.get_channel(payload.data.broadcaster.name).send(
f"Thank you {payload.data.user.name} for following the channel!"
Expand Down Expand Up @@ -653,7 +660,9 @@ async def update_bot_http_token(self, token):

async def __validate__(self):
validate_result = await self._http.validate(token=self.config.BOT_OAUTH_TOKEN)
print(f"{Fore.GREEN}Validation complete: {validate_result}{Style.RESET_ALL}")
logger.info(
f"{Fore.GREEN}Validation complete: {validate_result}{Style.RESET_ALL}"
)

async def __psclient_init__(self) -> None:
if self.config.get_bot_config().get("bot_features").get("enable_psclient"):
Expand Down Expand Up @@ -709,7 +718,7 @@ async def detect_bot_spam(message: twitchio.Message) -> bool:
>= 1
or str(message.content).lower().count("order management panel") >= 1
) and str(message.content).lower().count("dogehype") >= 1:
print("Bot detected")
logger.info("Bot detected")
return True
return False

Expand All @@ -718,7 +727,10 @@ async def event_ready(self):
"""Bot failed to join channel."""
await self.join_channels(
channels=[
self.config.get("twitch").get("channel").get("bot_join_channel")
self.config.get_bot_config()
.get("twitch")
.get("channel")
.get("bot_join_channel")
]
)

Expand All @@ -733,21 +745,32 @@ async def event_ready(self):
f"as bot user: {Fore.LIGHTBLUE_EX}{self.nick}{Fore.LIGHTWHITE_EX} "
f"({Fore.LIGHTBLUE_EX}ID: {self.user_id}{Fore.LIGHTWHITE_EX})!{Style.RESET_ALL}"
)
# uncomment below to say in chat when the bot joins
# await channel.send(f'Logged into channel(s): {channel.name}, as bot user: '
# f'{self.nick} (ID: {self.user_id})')
if (
self.config.get_bot_config()
.get("bot_features")
.get("announce_join")
):
await channel.send(
f"has joined the chat! PowerUpL EntropyWins PowerUpR"
)

# By default, turn on the sound and user commands
from cogs.ascii_cog import AsciiCog
if (
self.config.get_bot_config()
.get("bot_features")
.get("cogs")
.get("ascii_cog")
.get("enable_ascii_cog")
):
from cogs.ascii_cog import AsciiCog

self.add_cog(AsciiCog(self))
self.add_cog(AsciiCog(self))

async def event_message(self, message: twitchio.Message):
"""Messages with echo set to True are messages sent by the bot. ignore them."""
if message.echo:
return
# Print the contents of our message to console...
print(
logger.info(
f"{Fore.RED}[{message.channel.name}]{Fore.BLUE}[{message.author.name}]{Fore.RED}: {Fore.WHITE}"
f"{message.content}{Style.RESET_ALL}"
)
Expand Down Expand Up @@ -786,7 +809,7 @@ async def add_kill_my_shell_redemption_reward(self, broadcaster: PartialUser):
global_cooldown=5 * 60,
token=self.config.BOT_OAUTH_TOKEN,
)
print(
logger.info(
f"{Fore.RED}Added {Fore.MAGENTA}`Kill My Shell`{Fore.RED} channel point redemption.{Style.RESET_ALL}"
)

Expand All @@ -796,20 +819,22 @@ async def add_vip_auto_redemption_reward(self, broadcaster: PartialUser):
token=self.config.BOT_OAUTH_TOKEN, broadcaster_id=broadcaster.id, first=100
)
if len(vips) < int(
self.config.get("twitch").get("channel").get("max_vip_slots")
self.config.get_bot_config()
.get("twitch")
.get("channel")
.get("max_vip_slots")
):
await self._http.create_reward(
broadcaster_id=broadcaster.id,
title="VIP",
cost=80085,
prompt="VIPs have the ability to equip a special chat "
"badge and bypass the chat limit in slow mode!",
prompt="VIPs have the ability to equip a special chat badge and bypass the chat limit in slow mode!",
max_per_user=1,
global_cooldown=5 * 60,
token=self.config.BOT_OAUTH_TOKEN,
)
print(
f"{Fore.RED}Added {Fore.MAGENTA}`VIP`{Fore.RED} channel point redemption.{Style.RESET_ALL}"
logger.info(
f"{Fore.RED}Added {Fore.MAGENTA}'VIP'{Fore.RED} channel point redemption.{Style.RESET_ALL}"
)

async def delete_all_custom_rewards(self, broadcaster: PartialUser):
Expand All @@ -821,7 +846,7 @@ async def delete_all_custom_rewards(self, broadcaster: PartialUser):
only_manageable=True,
token=self.config.BOT_OAUTH_TOKEN,
)
print(
logger.info(
f"{Fore.RED}Got rewards: [{Fore.MAGENTA}{json.dumps(rewards)}{Fore.RED}]{Style.RESET_ALL}"
)
if rewards is not None:
Expand All @@ -834,7 +859,7 @@ async def delete_all_custom_rewards(self, broadcaster: PartialUser):
reward_id=reward["id"],
token=self.config.BOT_OAUTH_TOKEN,
)
print(
logger.info(
f"{Fore.RED}Deleted reward: [{Fore.MAGENTA}id={reward['id']}{Fore.RED}]"
f"[{Fore.MAGENTA}title={reward['title']}{Fore.RED}]{Style.RESET_ALL}"
)
Expand Down Expand Up @@ -879,7 +904,7 @@ async def announce_shoutout(
)

except Exception as error:
print(
logger.error(
f"{Fore.RED}Could not send shoutout announcement to {Fore.MAGENTA}{channel['broadcaster_name']}"
f"{Fore.RED} from channel {Fore.MAGENTA}{broadcaster.name}{Fore.RED}: {error}{Style.RESET_ALL}"
)
Expand All @@ -903,7 +928,7 @@ async def announce_shoutout(

except Exception as error:
"""Eg: shoutout global cooldown "You have to wait 1m 30s before giving another Shoutout."""
print(
logger.error(
f"{Fore.RED}Could not perform a Twitch Shoutout command to {Fore.MAGENTA}{channel['broadcaster_name']}"
f"{Fore.RED} from channel {Fore.MAGENTA}{broadcaster.name}{Fore.RED}: {error}{Style.RESET_ALL}"
)
Expand Down Expand Up @@ -983,35 +1008,51 @@ async def hello(self, ctx: commands.Context):
@commands.command()
async def join(self, ctx: commands.Context):
"""type !join <channel> to join the channel"""
param_username = re.sub(
r"^@", "", str(ctx.message.content).split(maxsplit=1)[1]
)
# Limit to broadcaster
if ctx.author.is_broadcaster or int(ctx.author.id) == 125444292:
await self.join_channels([param_username])
try:
param_username = re.sub(
r"^@", "", str(ctx.message.content).split(maxsplit=1)[1]
)
# Limit to broadcaster
if ctx.author.is_broadcaster or int(ctx.author.id) == 125444292:
await self.join_channels([param_username])
except IndexError:
logger.error("!join command failed. Regex pattern did not match.")

@commands.command()
async def leave(self, ctx: commands.Context):
"""type !leave <channel> to join the channel"""
param_username = re.sub(
r"^@", "", str(ctx.message.content).split(maxsplit=1)[1]
)
# Limit to broadcaster
if (
ctx.author.is_broadcaster
or int(ctx.author.id) == 125444292
and str(
self.config.get("twitch").get("channel").get("bot_join_channel")
).lower()
!= param_username.lower()
): # stay connected to init channel
await self.part_channels([param_username])
# also remove event subs
broadcasters: List[User] = await self.fetch_users(names=[param_username])
await self.delete_event_subscriptions(broadcasters=broadcasters)
print(
f"{Fore.RED}Connected_channels: {Fore.MAGENTA}{self.connected_channels}{Fore.RED}!{Style.RESET_ALL}"
)
try:
param_username = re.sub(
r"^@", "", str(ctx.message.content).split(maxsplit=1)[1]
)
if (
ctx.author.is_broadcaster or int(ctx.author.id) == 125444292
) and str( # TODO: Don't hardcode user
self.config.get_bot_config()
.get("twitch")
.get("channel")
.get("bot_join_channel")
).lower() != param_username.lower():
# stay connected to init channel
await self.part_channels([param_username])
# also remove event subs
if (
self.config.get_bot_config()
.get("bot_features")
.get("enable_esclient")
):
broadcasters: List[User] = await self.fetch_users(
names=[param_username]
)
await self.es_client.delete_event_subscriptions(
broadcasters=broadcasters
)

logger.info(
f"{Fore.RED}Connected_channels: {Fore.MAGENTA}{self.connected_channels}{Fore.RED}!{Style.RESET_ALL}"
)
except IndexError:
logger.error("!leave command failed. Regex pattern did not match.")

@commands.command(aliases=["infosecstreams", "cyber_streams", "streams"])
async def infosec_streams(self, ctx: commands.Context):
Expand Down
25 changes: 20 additions & 5 deletions twitchrce/main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,9 @@ async def refresh_user_token(user: any) -> str:
"dynamodb", region_name=bot_config.get_bot_config().get("aws").get("region_name")
)
user_table = dynamodb.Table("MSecBot_User")
ec2 = boto3.client("ec2", region_name=bot_config.get_bot_config().get("aws").get("region_name"))
ec2 = boto3.client(
"ec2", region_name=bot_config.get_bot_config().get("aws").get("region_name")
)


async def setup_bot() -> CustomBot:
Expand All @@ -119,7 +121,7 @@ async def setup_bot() -> CustomBot:
░ ▒ ░ ░ ▒ ░ ░ ░ ▒ ▒ ░▒░ ░ ░▒ ░ ▒░ ░ ▒ ░ ░ ░
░ ░ ░ ▒ ░ ░ ░ ░ ░░ ░ ░░ ░ ░ ░
░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░
░ ░ {splash['version']}
░ ░ {splash['version']}
{splash['description']}
{splash['project_url']}
Expand Down Expand Up @@ -161,10 +163,16 @@ async def setup_bot() -> CustomBot:
"analytics:read:games analytics:read:extensions"
)
api_gateway_invoke_url = (
bot_config.get_bot_config().get("aws").get("api_gateway").get("api_gateway_invoke_url")
bot_config.get_bot_config()
.get("aws")
.get("api_gateway")
.get("api_gateway_invoke_url")
)
api_gateway_route = (
bot_config.get_bot_config().get("aws").get("api_gateway").get("api_gateway_route")
bot_config.get_bot_config()
.get("aws")
.get("api_gateway")
.get("api_gateway_route")
)
redirect_uri = f"{api_gateway_invoke_url}{api_gateway_route}"
authorization_url = (
Expand All @@ -180,7 +188,14 @@ async def setup_bot() -> CustomBot:
bot_user = None
try:
response = user_table.get_item(
Key={"id": int(bot_config.get_bot_config().get("twitch").get("bot_auth").get("bot_user_id"))}
Key={
"id": int(
bot_config.get_bot_config()
.get("twitch")
.get("bot_auth")
.get("bot_user_id")
)
}
)
bot_user = response.get("Item")

Expand Down

0 comments on commit 6fb10f7

Please sign in to comment.