Versão Final do Lives do Cellbit

.env.example

@@ -1,12 +1,14 @@
-WEBHOOK_URL=
+# Twitch
 TWITCH_APP_ID=
 TWITCH_APP_SECRET=
-TWITCH_BEARER=
+TARGET_USERNAME=
+
+# Webhook (URL pública da VPS para receber eventos da Twitch)
+WEBHOOK_URL=
+PORT=8080
+
+# Twitter/X
 TWITTER_CONSUMER_KEY=
 TWITTER_CONSUMER_SECRET=
 TWITTER_ACCESS_TOKEN=
 TWITTER_ACCESS_SECRET=
-
-TARGET_USERNAME=
-
-LOGGING=

.gitignore

@@ -26,7 +26,7 @@ share/python-wheels/
 .installed.cfg
 *.egg
 MANIFEST
-gameImg.jpg
+imageGame.jpg
 
 # PyInstaller
 #  Usually these files are written by a python script from a template
@@ -131,4 +131,5 @@ dmypy.json
 .gitpod.yml
 
 
-test.py
+test.py
+.vscode/

bot.py

@@ -1,151 +1,16 @@
-import os
-import tweepy
-import logging
-import threading
-from twitchAPI import Twitch, EventSub
-from dotenv import load_dotenv
-from functions.functionsBot import compareImages
-from functions.twitchAPI import getStream, dateStream, getVideo
-from asyncio import sleep
-
-# load env variables
-load_dotenv()
-
-# env variables 
-WEBHOOK_URL = os.environ.get('WEBHOOK_URL')
-APP_ID = os.environ.get('TWITCH_APP_ID')
-APP_SECRET = os.environ.get('TWITCH_APP_SECRET')
-ACCESS_TOKEN = os.environ.get('TWITTER_ACCESS_TOKEN')
-ACCESS_SECRET = os.environ.get('TWITTER_ACCESS_SECRET')
-CONSUMER_KEY = os.environ.get('TWITTER_CONSUMER_KEY')
-CONSUMER_SECRET = os.environ.get('TWITTER_CONSUMER_SECRET')
-PORT = os.environ.get('PORT', 8080)
-TARGET_USERNAME = os.environ.get('TARGET_USERNAME')
-LOGGING = os.environ.get('LOGGING')
-
-if LOGGING == 'TRUE':
-    logging.basicConfig(level=logging.INFO)
-
-# global variables
-currentGame = None
-currentTitle = None
-streamId = None
-online = False
-gamesPlayed = list()
-gamesBlacklist = ('Just Chatting', 'Watch Parties', 'Tabletop RPGs')
-forever = threading.Event()
-
-# login in twitch api
-twitch = Twitch(APP_ID, APP_SECRET)
-twitch.authenticate_app([])
-
-# twitter authentication
-auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
-auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET)
-api = tweepy.API(auth)
-
-try:
-    api.verify_credentials()
-    print('Authentication Successful')
-except:
-    print('Authentication Error')
-
-# get the user_id from twitch user
-uid = twitch.get_users(logins=[TARGET_USERNAME])
-user_id = uid['data'][0]['id']
-currentTitle = twitch.get_channel_information(user_id)['data'][0]['title']
-
-# get the informations
-try:
-    stream = twitch.get_streams(user_id=user_id)
-    currentGame = stream['data'][0]['game_name']
-    currentTitle = stream['data'][0]['title']
-    streamId = stream['data'][0]['id']
-    if currentGame not in gamesBlacklist:
-        gamesPlayed.append(currentGame)
-    online = True
-except:
-    online = False
-
-# functions callbacks
-async def stream_online(data: dict):
-    global online, currentGame, streamId
-    stream = twitch.get_streams(user_id=user_id)
-    title = stream['data'][0]['title']
-    currentGame = stream['data'][0]['game_name']
-    streamId = stream['data'][0]['id']
-    api.update_status(f'Cellbit entrou ao vivo!\n\nTítulo: {title}\ntwitch.tv/cellbit')
-    online = True
-
-async def stream_offline(data: dict):
-    global online, gamesPlayed
-    api.update_status('Cellbit encerrou a live!')
-    online = False
-
-    if len(gamesPlayed) > 0:
-        tweetId = api.user_timeline(screen_name='livesdocellbit')[0].id
-
-        date = dateStream()
-        status = f"[{date['day']}/{date['month']}/{date['year']}] Games Jogados:\n\n"
-        for game in gamesPlayed:
-            status += f'• {game}\n'
-        if getVideo['title'] != currentTitle:
-            status += f'\nVOD: https://twitchtracker.com/cellbit/streams/{streamId}'
-        else:
-            status += f'\nVOD: {getVideo()["link"]}'
-        api.update_status(status, in_reply_to_status_id = tweetId)
-        gamesPlayed = list()
-
-async def channel_update(data: dict):
-    global currentGame, currentTitle, gamesPlayed
-
-    game = data['event']['category_name']
-    title = data['event']['title']
-
-    if game != currentGame and online == True:
-        # stream = twitch.get_streams(user_id=user_id)
-        timeVod = getStream()
-        h, m, s = timeVod['vodHours'], timeVod['vodMinutes'], timeVod['vodSeconds']
-        try:
-            import urllib.request
-            imageUrl = twitch.get_games(names=game)['data'][0]['box_art_url'].replace('{width}', '600').replace('{height}', '800')
-            urllib.request.urlretrieve(imageUrl, 'gameImg.jpg')
-            status = f'Cellbit está jogando: {game}\nTempo no VOD: {h}h{m}m{s}s\n\ntwitch.tv/cellbit'
-            if compareImages() or game == 'Just Chatting':
-                api.update_status(status)
-            else: 
-                api.update_status_with_media(status, 'gameImg.jpg')
-        except:
-            api.update_status(status)
-        finally:
-            if game not in gamesPlayed and game not in gamesBlacklist:
-                gamesPlayed.append(game)
-            currentGame = game
-
-        # sleep(1)
-        #  infoVideo = getVideo()        
-    if title != currentTitle and online == False:
-        api.update_status(f'[TÍTULO] {title}')
-        currentTitle = title
-
-# subscribe to EventSub
-hook = EventSub(WEBHOOK_URL, APP_ID, PORT, twitch)
-hook.unsubscribe_all()
-hook.start()
-
-print('Iniciando webhooks...\n')
-
-hook.listen_channel_update(user_id, channel_update)
-print('[OK] CHANNEL UPDATE - WEBHOOK')
-hook.listen_stream_offline(user_id, stream_offline)
-print('[OK] STREAM OFFLINE - WEBHOOK')
-hook.listen_stream_online(user_id, stream_online)
-print('[OK] STREAM ONLINE - WEBHOOK')
-print('\n')
-
-
-try:
-    print('Rodando!')
-    forever.wait()
-finally:
-    hook.stop()
+import asyncio
+from functions.botFunctions import printEvent
+from functions.twitchFunctions import connectEventSub, connectTwitch, verifyStreamIsOnline
+
+async def main():
+    await connectTwitch()
+    hook = await connectEventSub()
+    await verifyStreamIsOnline()
+
+    printEvent(True, 'on_ready')
+    try:
+        await asyncio.Event().wait()
+    finally:
+        await hook.stop()
+
+asyncio.run(main())

config.py

@@ -0,0 +1,31 @@
+from dataclasses import dataclass
+import os
+
+@dataclass(frozen=True)
+class BotConfig:
+    twitch_app_id: str
+    twitch_app_secret: str
+    webhook_url: str
+    twitter_consumer_key: str
+    twitter_consumer_secret: str
+    twitter_access_token: str
+    twitter_access_token_secret: str
+    streamer_username: str
+
+    @staticmethod
+    def load_from_env() -> 'BotConfig':
+        def get(name: str, required: bool = True) -> str:
+            value = os.getenv(name)
+            if required and not value:
+                raise ValueError(f"Environment variable '{name}' is required but not set.")
+            return value or ""
+        return BotConfig(
+            twitch_app_id=get('TWITCH_APP_ID'),
+            twitch_app_secret=get('TWITCH_APP_SECRET'),
+            webhook_url=get('WEBHOOK_URL'),
+            twitter_consumer_key=get('TWITTER_CONSUMER_KEY'),
+            twitter_consumer_secret=get('TWITTER_CONSUMER_SECRET'),
+            twitter_access_token=get('TWITTER_ACCESS_TOKEN'),
+            twitter_access_token_secret=get('TWITTER_ACCESS_TOKEN_SECRET'),
+            streamer_username=get('STREAMER_USERNAME')
+        )

core/stream_state.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+
+@dataclass
+class StreamState:
+    """
+    Represents the current state of the stream, including whether it's online, the current game being played, and the stream title.
+    """
+
+    online: bool = False
+    current_game: str = ""
+    stream_title: str = ""
+
+    def reset_for_offline(self):
+        """
+        Resets the stream state to default values when the stream goes offline.
+        """
+        self.online = False
+
+    def update_from_update(self, update: dict):
+        self.online = update.get('online', self.online)
+        self.current_game = update.get('current_game', self.current_game)
+        self.stream_title = update.get('stream_title', self.stream_title)
+

functions/botFunctions.py

@@ -0,0 +1,63 @@
+from utils import variables
+
+def printEvent(sucessOrfail: bool, event: str, info: str='') -> None:
+    colors = {
+        True: '\33[92m',
+        False: '\33[91m'
+    }
+
+    events = {
+        # Bot
+
+        'info_stream': '[BOT] Informações da Stream',
+        'on_ready': '[BOT] ONLINE',
+
+        # Twitch Authentication
+        'twitch_authenticated': '[TWITCH AUTHENTICATION] OK',
+
+        # Twitter Authenticaton
+        'twitter_authenticated': '[TWITTER AUTHENTICATION] OK',
+
+        # Twitter Posts 
+        'twitter_post': '[TWITTER] POST SUCESSO',
+        'twitter_post_image': '[TWITTER] POST IMAGE SUCESSO',
+        'twitter_post_reply': '[TWITTER] POST REPLY SUCESSO',
+
+        # Twitch Subscribe Event Subs
+        'event_sub': '[EVENTSUB] OK',
+
+        # Twitch Event Sub
+        'live_on': '[TWITCH EVENTSUB] - Live Online!',
+        'live_off': '[TWITCH EVENTSUB] - Live Offline',
+        'game_changed': '[TWITCH EVENTSUB] - Game trocado - ',
+        'title': '[TWITCH EVENTSUB] - Título trocado - '
+
+    }
+
+    return print(colors[sucessOrfail], events[event])
+
+def downloadImageGame(image_url):
+    from urllib.request import urlretrieve
+    try:
+        urlretrieve(image_url, 'imageGame.jpg')
+    except:
+        return False
+
+def compareImages():
+    from base64 import b64encode
+    with open('404.jpg', 'rb') as not_found:
+        image_not_found = b64encode(not_found.read())
+    with open ('imageGame.jpg', 'rb') as game_image:
+        image_game = b64encode(game_image.read())
+        if image_not_found == image_game:
+            return True
+        return False
+
+def gamesPlayed(games_list):
+    games = ''
+    for game in games_list: 
+        games = f'• {game}\n'
+    return games
+
+def linkTwitchTracker(stream_id):
+    return f'https://twitchtracker.com/{variables.streamer_nickname}/streams/{str(stream_id)}'