lapatatedouce/MamieHenriette
1
0
Fork 0
mirror of https://github.com/skylanix/MamieHenriette.git synced 2026-02-06 14:50:34 +01:00
Code Issues Packages Projects Releases Wiki Activity

Bot twitch

Browse Source
This commit is contained in:
Kepka Ludovic
2025-08-21 15:35:58 +02:00
committed by skylanix
parent 206e71758c
commit cf8fd43a4e
1 changed files with 24 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
twitchbot/__init__.py
View File

@@ -6,38 +6,44 @@ from twitchAPI.twitch import Twitch
from twitchAPI.type import AuthScope, ChatEvent
from twitchAPI.chat import Chat, ChatEvent, ChatMessage, EventData
CLIENT_ID= 'TODO'
CLIENT_SECRET='TODO'
from database.helpers import ConfigurationHelper
from webapp import webapp
USER_SCOPE = [AuthScope.CHAT_READ, AuthScope.CHAT_EDIT]
CHANNEL = "#TODO"
ACCESS_TOKEN = 'TODO'
REFRESH_TOKEN = 'TODO'
async def _onReady(ready_event: EventData):
logging.info('Twitch bot ready')
await ready_event.chat.join_room(CHANNEL)
with webapp.app_context():
await ready_event.chat.join_room(ConfigurationHelper().getValue('twitch_channel'))
async def _onMessage(msg: ChatMessage):
print(f'in {msg.room.name}, {msg.user.name} said: {msg.text}')
logging.info(f'in {msg.room.name}, {msg.user.name} said: {msg.text}')
# commande qui repond "bonjour" a "!hello"
async def _helloCommand(msg: ChatMessage):
await msg.reply(f'Bonjour {msg.user.name}')
def _isConfigured() -> bool:
helper = ConfigurationHelper()
return helper.getValue('twitch_client_id') != None and helper.getValue('twitch_client_secret') != None and helper.getValue('twitch_access_token') != None and helper.getValue('twitch_refresh_token') != None and helper.getValue('twitch_channel') != None
class TwitchBot() :
async def _connect(self):
self.twitch = await Twitch(CLIENT_ID, CLIENT_SECRET)
await self.twitch.set_user_authentication(ACCESS_TOKEN, USER_SCOPE, REFRESH_TOKEN)
self.chat = await Chat(self.twitch)
self.chat.register_event(ChatEvent.READY, _onReady)
self.chat.register_event(ChatEvent.MESSAGE, _onMessage)
# chat.register_event(ChatEvent.SUB, on_sub)
self.chat.register_command('hello', _helloCommand)
self.chat.start()
with webapp.app_context():
if _isConfigured() :
helper = ConfigurationHelper()
self.twitch = await Twitch(helper.getValue('twitch_client_id'), helper.getValue('twitch_client_secret'))
await self.twitch.set_user_authentication(helper.getValue('twitch_access_token'), USER_SCOPE, helper.getValue('twitch_refresh_token'))
self.chat = await Chat(self.twitch)
self.chat.register_event(ChatEvent.READY, _onReady)
self.chat.register_event(ChatEvent.MESSAGE, _onMessage)
# chat.register_event(ChatEvent.SUB, on_sub)
self.chat.register_command('hello', _helloCommand)
self.chat.start()
else:
logging.info("Twitch is not configured")
def begin(self):
asyncio.run(self._connect())