kurbezz/discord-bot
1
0
Fork 0
mirror of https://github.com/kurbezz/discord-bot.git synced 2025-12-06 07:05:36 +01:00
Code Issues Packages Projects Releases Wiki Activity

Update

Browse Source
This commit is contained in:
Bulat Kurbanov
2024-10-04 18:48:32 +02:00
parent 83a407a84c
commit 055e9579da
9 changed files with 130 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1,2 +1,3 @@
venv venv
.DS_Store

29
configs/hafmc.toml Normal file
View File

@@ -0,0 +1,29 @@
[twitch]
id = 59900845
name = "hafmc"
[notifications]
start_stream = '''
HafMC сейчас стримит {title} ({category})!
Присоединяйся: https://twitch.tv/hafmc
'''
change_category = '''
HafMC начал играть в {category}!
Присоединяйся: https://twitch.tv/hafmc
'''
[integrations]
[integrations.discord]
guild_id = 1198051900906549329
notifications_channel_id = 1198296540964475002
[integrations.discord.games_list]
channel_id = 1201810638800691210
message_id = 1239664178038313012
[integrations.telegram]
notifications_channel_id = -1001939021131

18
configs/ssstano.toml Normal file
View File

@@ -0,0 +1,18 @@
[twitch]
id = 188615689
name = "ssstano"
[notifications]
start_stream = '''
ssstano ебашит LIVE стрим для everyone
https://www.twitch.tv/ssstano
https://www.twitch.tv/ssstano
https://www.twitch.tv/ssstano
'''
[integrations]
[integrations.telegram]
notifications_channel_id = -1002152372995

1
docker/build.dockerfile
View File

@@ -15,6 +15,7 @@ FROM python:3.12-slim AS runtime
RUN apt update && apt install -y --no-install-recommends netcat-traditional wkhtmltopdf && apt clean RUN apt update && apt install -y --no-install-recommends netcat-traditional wkhtmltopdf && apt clean
COPY ./src/ /app COPY ./src/ /app
COPY ./configs/ /app/configs
ENV PATH="/opt/venv/bin:$PATH" ENV PATH="/opt/venv/bin:$PATH"
ENV VENV_PATH=/opt/venv ENV VENV_PATH=/opt/venv

48
src/config.py
View File

@@ -1,29 +1,38 @@
import json import tomllib
from pydantic import BaseModel, field_validator from pydantic import BaseModel, field_validator
from pydantic_settings import BaseSettings from pydantic_settings import BaseSettings
from pathlib import Path
class TwitchConfig(BaseModel): class TwitchConfig(BaseModel):
CHANNEL_ID: str id: int
CHANNEL_NAME: str name: str
class NotificationsConfig(BaseModel):
start_stream: str
change_category: str | None = None
class GamesListConfig(BaseModel):
channel_id: int
message_id: int
class DiscordConfig(BaseModel): class DiscordConfig(BaseModel):
GUILD_ID: int guild_id: int
CHANNEL_ID: int notifications_channel_id: int
games_list: GamesListConfig | None = None
GAME_LIST_CHANNEL_ID: int class TelegramConfig(BaseModel):
GAME_LIST_MESSAGE_ID: int notifications_channel_id: int
class IntegrationsConfig(BaseModel):
discord: DiscordConfig | None = None
telegram: TelegramConfig | None = None
class StreamerConfig(BaseModel): class StreamerConfig(BaseModel):
TWITCH: TwitchConfig twitch: TwitchConfig
DISCORD: DiscordConfig | None = None notifications: NotificationsConfig
TELEGRAM_CHANNEL_ID: int | None = None integrations: IntegrationsConfig
START_STREAM_MESSAGE: str | None = None
CHANGE_CATEGORY_MESSAGE: str | None = None
class Config(BaseSettings): class Config(BaseSettings):
@@ -45,13 +54,16 @@ class Config(BaseSettings):
SECRETS_FILE_PATH: str SECRETS_FILE_PATH: str
@field_validator("STREAMERS", mode="before") @field_validator("STREAMERS", mode="before")
def check_streamers(cls, value): def check_streamers(cls, value):
if isinstance(value, str): config_dir = Path("/app/configs")
return json.loads(value) streamers = []
for toml_file in config_dir.glob("*.toml"):
return value if toml_file.is_file():
with open(toml_file, "rb") as f:
streamer_config = tomllib.load(f)
streamers.append(StreamerConfig(**streamer_config))
return streamers if streamers else value
config = Config() # type: ignore config = Config() # type: ignore

17
src/services/discord.py
View File

@@ -17,13 +17,16 @@ def get_game_list_channel_to_message_map() -> dict[int, int]:
result = {} result = {}
for streamer in config.STREAMERS: for streamer in config.STREAMERS:
if streamer.DISCORD is None: if (integration := streamer.integrations.discord) is None:
continue continue
if streamer.DISCORD.GAME_LIST_CHANNEL_ID is None or streamer.DISCORD.GAME_LIST_MESSAGE_ID is None: if (games_list := integration.games_list) is None:
continue continue
result[streamer.DISCORD.GAME_LIST_CHANNEL_ID] = streamer.DISCORD.GAME_LIST_MESSAGE_ID if games_list.channel_id is None or games_list.message_id is None:
continue
result[games_list.channel_id] = games_list.message_id
return result return result
@@ -39,14 +42,14 @@ class DiscordClient(discord.Client):
async def setup_hook(self): async def setup_hook(self):
for streamer in config.STREAMERS: for streamer in config.STREAMERS:
if streamer.DISCORD is None: if (integration := streamer.integrations.discord) is None:
continue continue
if streamer.DISCORD.GAME_LIST_CHANNEL_ID is None or streamer.DISCORD.GAME_LIST_MESSAGE_ID is None: if integration.games_list is None:
continue continue
self.tree.copy_global_to(guild=Object(id=streamer.DISCORD.GUILD_ID)) self.tree.copy_global_to(guild=Object(id=integration.guild_id))
await self.tree.sync(guild=Object(id=streamer.DISCORD.GUILD_ID)) await self.tree.sync(guild=Object(id=integration.guild_id))
async def on_ready(self): async def on_ready(self):
await self.change_presence( await self.change_presence(

24
src/services/notification.py
View File

@@ -33,14 +33,18 @@ async def notify_discord(msg: str, channel_id: str):
async def notify(msg: str, streamer_config: StreamerConfig): async def notify(msg: str, streamer_config: StreamerConfig):
if streamer_config.DISCORD is not None: integrations = streamer_config.integrations
try:
await notify_discord(msg, str(streamer_config.DISCORD.CHANNEL_ID))
except Exception as e:
logger.error("Failed to notify discord", exc_info=e)
if streamer_config.TELEGRAM_CHANNEL_ID is not None: if (discord := integrations.discord) is not None:
try: if discord.notifications_channel_id is not None:
await notify_telegram(msg, str(streamer_config.TELEGRAM_CHANNEL_ID)) try:
except Exception as e: await notify_discord(msg, str(discord.notifications_channel_id))
logger.error("Failed to notify telegram", exc_info=e) except Exception as e:
logger.error("Failed to notify discord", exc_info=e)
if (telegram := integrations.telegram) is not None:
if telegram.notifications_channel_id is not None:
try:
await notify_telegram(msg, str(telegram.notifications_channel_id))
except Exception as e:
logger.error("Failed to notify telegram", exc_info=e)

10
src/services/scheduler_sync/synchronizer.py
View File

@@ -82,7 +82,7 @@ async def edit_events(
async def syncronize(twitch: TwitchConfig, discord_guild_id: int): async def syncronize(twitch: TwitchConfig, discord_guild_id: int):
twitch_events = await get_twitch_events(twitch.CHANNEL_ID) twitch_events = await get_twitch_events(str(twitch.id))
discord_events = await get_discord_events(discord_guild_id) discord_events = await get_discord_events(discord_guild_id)
twitch_events_with_id = [(event.uid, event) for event in twitch_events] twitch_events_with_id = [(event.uid, event) for event in twitch_events]
@@ -91,9 +91,9 @@ async def syncronize(twitch: TwitchConfig, discord_guild_id: int):
for event in discord_events for event in discord_events
] ]
await add_events(discord_guild_id, twitch.CHANNEL_NAME, twitch_events_with_id, discord_events_with_id) await add_events(discord_guild_id, twitch.name, twitch_events_with_id, discord_events_with_id)
await remove_events(discord_guild_id, twitch_events_with_id, discord_events_with_id) await remove_events(discord_guild_id, twitch_events_with_id, discord_events_with_id)
await edit_events(discord_guild_id, twitch.CHANNEL_NAME, twitch_events_with_id, discord_events_with_id) await edit_events(discord_guild_id, twitch.name, twitch_events_with_id, discord_events_with_id)
async def start_synchronizer(): async def start_synchronizer():
@@ -102,10 +102,10 @@ async def start_synchronizer():
while True: while True:
try: try:
for streamer in config.STREAMERS: for streamer in config.STREAMERS:
if streamer.DISCORD is None: if (integration := streamer.integrations.discord) is None:
continue continue
await syncronize(streamer.TWITCH, streamer.DISCORD.GUILD_ID) await syncronize(streamer.twitch, integration.guild_id)
except Exception as e: except Exception as e:
logging.error(e) logging.error(e)

42
src/services/twitch.py
View File

@@ -60,7 +60,7 @@ class TwitchService:
def __init__(self, twitch: Twitch): def __init__(self, twitch: Twitch):
self.twitch = twitch self.twitch = twitch
self.state: dict[str, State | None] = {} self.state: dict[int, State | None] = {}
@classmethod @classmethod
async def authorize(cls): async def authorize(cls):
@@ -78,31 +78,31 @@ class TwitchService:
return twitch return twitch
def get_streamer_config(self, streamer_id: str) -> StreamerConfig: def get_streamer_config(self, streamer_id: int) -> StreamerConfig:
for streamer in config.STREAMERS: for streamer in config.STREAMERS:
if streamer.TWITCH.CHANNEL_ID == streamer_id: if streamer.twitch.id == streamer_id:
return streamer return streamer
raise ValueError(f"Streamer with id {streamer_id} not found") raise ValueError(f"Streamer with id {streamer_id} not found")
async def notify_online(self, streamer_id: str): async def notify_online(self, streamer_id: int):
current_state = self.state.get(streamer_id) current_state = self.state.get(streamer_id)
if current_state is None: if current_state is None:
raise RuntimeError("State is None") raise RuntimeError("State is None")
streamer = self.get_streamer_config(streamer_id) streamer = self.get_streamer_config(streamer_id)
if streamer.START_STREAM_MESSAGE is None: if streamer.notifications.start_stream is None:
return return
msg = streamer.START_STREAM_MESSAGE.replace("\\n", "\n").format( msg = streamer.notifications.start_stream.format(
title=current_state.title, title=current_state.title,
category=current_state.category category=current_state.category
) )
await notify(msg, streamer) await notify(msg, streamer)
async def notify_change_category(self, streamer_id: str): async def notify_change_category(self, streamer_id: int):
current_state = self.state.get(streamer_id) current_state = self.state.get(streamer_id)
if current_state is None: if current_state is None:
@@ -113,20 +113,21 @@ class TwitchService:
streamer = self.get_streamer_config(streamer_id) streamer = self.get_streamer_config(streamer_id)
if streamer.CHANGE_CATEGORY_MESSAGE is None: if streamer.notifications.change_category is None:
return return
msg = streamer.CHANGE_CATEGORY_MESSAGE.replace("\\n", "\n").format( msg = streamer.notifications.change_category.format(
title=current_state.title,
category=current_state.category category=current_state.category
) )
await notify(msg, streamer) await notify(msg, streamer)
async def get_current_stream(self, streamer_id: str, retry_count: int = 5, delay: int = 5): async def get_current_stream(self, streamer_id: int, retry_count: int = 5, delay: int = 5):
remain_retry = retry_count remain_retry = retry_count
while remain_retry > 0: while remain_retry > 0:
stream = await first(self.twitch.get_streams(user_id=[streamer_id])) stream = await first(self.twitch.get_streams(user_id=[str(streamer_id)]))
if stream is not None: if stream is not None:
return stream return stream
@@ -137,7 +138,7 @@ class TwitchService:
return None return None
async def on_channel_update(self, event: ChannelUpdateEvent): async def on_channel_update(self, event: ChannelUpdateEvent):
brodcaster_id = event.event.broadcaster_user_id brodcaster_id = int(event.event.broadcaster_user_id)
stream = await self.get_current_stream(brodcaster_id) stream = await self.get_current_stream(brodcaster_id)
if stream is None: if stream is None:
@@ -158,7 +159,7 @@ class TwitchService:
if changed: if changed:
await self.notify_change_category(brodcaster_id) await self.notify_change_category(brodcaster_id)
async def _on_stream_online(self, streamer_id: str): async def _on_stream_online(self, streamer_id: int):
current_stream = await self.get_current_stream(streamer_id) current_stream = await self.get_current_stream(streamer_id)
if current_stream is None: if current_stream is None:
return return
@@ -180,7 +181,7 @@ class TwitchService:
await self.notify_online(streamer_id) await self.notify_online(streamer_id)
async def on_stream_online(self, event: StreamOnlineEvent): async def on_stream_online(self, event: StreamOnlineEvent):
await self._on_stream_online(event.event.broadcaster_user_id) await self._on_stream_online(int(event.event.broadcaster_user_id))
async def run(self): async def run(self):
eventsub = EventSubWebhook( eventsub = EventSubWebhook(
@@ -191,15 +192,16 @@ class TwitchService:
) )
for streamer in config.STREAMERS: for streamer in config.STREAMERS:
current_stream = await self.get_current_stream(streamer.TWITCH.CHANNEL_ID) current_stream = await self.get_current_stream(streamer.twitch.id)
if current_stream: if current_stream:
self.state[streamer.TWITCH.CHANNEL_ID] = State( self.state[streamer.twitch.id] = State(
title=current_stream.title, title=current_stream.title,
category=current_stream.game_name, category=current_stream.game_name,
last_live_at=datetime.now() last_live_at=datetime.now()
) )
else: else:
self.state[streamer.TWITCH.CHANNEL_ID] = None self.state[streamer.twitch.id] = None
try: try:
await eventsub.unsubscribe_all() await eventsub.unsubscribe_all()
@@ -209,8 +211,8 @@ class TwitchService:
logger.info("Subscribe to events...") logger.info("Subscribe to events...")
for streamer in config.STREAMERS: for streamer in config.STREAMERS:
await eventsub.listen_channel_update_v2(streamer.TWITCH.CHANNEL_ID, self.on_channel_update) await eventsub.listen_channel_update_v2(str(streamer.twitch.id), self.on_channel_update)
await eventsub.listen_stream_online(streamer.TWITCH.CHANNEL_ID, self.on_stream_online) await eventsub.listen_stream_online(str(streamer.twitch.id), self.on_stream_online)
logger.info("Twitch service started") logger.info("Twitch service started")
@@ -218,7 +220,7 @@ class TwitchService:
await sleep(self.UPDATE_DELAY) await sleep(self.UPDATE_DELAY)
for streamer in config.STREAMERS: for streamer in config.STREAMERS:
await self._on_stream_online(streamer.TWITCH.CHANNEL_ID) await self._on_stream_online(streamer.twitch.id)
finally: finally:
await eventsub.stop() await eventsub.stop()
await self.twitch.close() await self.twitch.close()