diff --git a/.env.example b/.env.example index 57b01d9..675612c 100644 --- a/.env.example +++ b/.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= \ No newline at end of file diff --git a/.gitignore b/.gitignore index ce44efb..088a26e 100644 --- a/.gitignore +++ b/.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 \ No newline at end of file +test.py +.vscode/ \ No newline at end of file diff --git a/bot.py b/bot.py index 2514929..9701742 100644 --- a/bot.py +++ b/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()) diff --git a/config.py b/config.py new file mode 100644 index 0000000..8203ad9 --- /dev/null +++ b/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') + ) \ No newline at end of file diff --git a/core/stream_state.py b/core/stream_state.py new file mode 100644 index 0000000..bcc3560 --- /dev/null +++ b/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) + \ No newline at end of file diff --git a/functions/botFunctions.py b/functions/botFunctions.py new file mode 100644 index 0000000..9526e1c --- /dev/null +++ b/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)}' diff --git a/functions/functionsBot.py b/functions/functionsBot.py deleted file mode 100644 index ee7bba1..0000000 --- a/functions/functionsBot.py +++ /dev/null @@ -1,11 +0,0 @@ -from base64 import b64encode - -def compareImages(): - with open('./404.jpg', 'rb') as notFound: - imageNotFound = b64encode(notFound.read()) - with open('gameImg.jpg', 'rb') as gameImage: - img = b64encode(gameImage.read()) - if imageNotFound == img: - return True - else: - return False \ No newline at end of file diff --git a/functions/twitchAPI.py b/functions/twitchAPI.py deleted file mode 100644 index f1e70a9..0000000 --- a/functions/twitchAPI.py +++ /dev/null @@ -1,140 +0,0 @@ -import requests -import json -import os -from dotenv import load_dotenv -load_dotenv() - -bearer = os.environ.get('TWITCH_BEARER') -client_id = os.environ.get('TWITCH_APP_ID') - -headers = { - 'Authorization': f'Bearer {bearer}', - 'Client-Id': f'{client_id}', -} - -params = { - 'user_login': 'Cellbit', -} - -def getStream(): - - import datetime - import dateutil.parser - - response = requests.get( - 'https://api.twitch.tv/helix/streams', params=params, headers=headers) - responseText = response.text - responseJson = json.loads(responseText) - - global started_at, game, vodId - - started_at = responseJson['data'][0]['started_at'] - game = responseJson['data'][0]['game_name'] - vodId = responseJson['data'][0]['id'] - title = responseJson['data'][0]['title'] - - parsedDate = dateutil.parser.isoparse(started_at) - parsedDate = str(parsedDate).split("+", 1) - parsedDate = parsedDate[0] - parsedDate = datetime.datetime.strptime(parsedDate, "%Y-%m-%d %H:%M:%S") - - now = datetime.datetime.now() - now = str(now).split(".", 1) - now = now[0] - now = datetime.datetime.strptime(now, "%Y-%m-%d %H:%M:%S") - - diff = now - (parsedDate - datetime.timedelta(hours=0)) - - seconds = diff.total_seconds() - - if seconds > 86400: - tempo_no_vod = str(datetime.timedelta(seconds=seconds-86400)) - tempo_no_vod = datetime.datetime.strptime(tempo_no_vod, "%H:%M:%S") - tempo_no_vod_hour = tempo_no_vod.hour + 24 - tempo_no_vod_minutes = tempo_no_vod.minute - tempo_no_vod_seconds = tempo_no_vod.second - - else: - tempo_no_vod = str(datetime.timedelta(seconds=seconds+86400, days=-1)) - tempo_no_vod = datetime.datetime.strptime(tempo_no_vod, "%H:%M:%S") - tempo_no_vod_hour = tempo_no_vod.hour - tempo_no_vod_minutes = tempo_no_vod.minute - tempo_no_vod_seconds = tempo_no_vod.second - - return { - 'vodHours': tempo_no_vod_hour, - 'vodMinutes': tempo_no_vod_minutes, - 'vodSeconds': tempo_no_vod_seconds, - 'game': game, - 'vodId': vodId, - 'title': title, - 'started_at': started_at - } - -def isOnline(): - try: - getStream() - return True - except: - return False - -def getImageGame(game): - import urllib.request - - response = requests.get( - f'https://api.twitch.tv/helix/games?name={game}', headers=headers) - responseText = response.text - responseJson = json.loads(responseText) - - imageUrl = responseJson['data'][0]['box_art_url'].replace('{width}', '800').replace('{height}', '600') - - urllib.request.urlretrieve(imageUrl, 'gameImg.jpg') - - -def getInfoUser(): - response = requests.get( - f'https://api.twitch.tv/helix/users?login=Cellbit', params=params, headers=headers) - responseText = response.text - responseJson = json.loads(responseText) - return { - 'userId': responseJson['data'][0]['id'] - } - -def getVideo(): - - params = { - 'user_id': '28579002', - } - - response = requests.get( - f'https://api.twitch.tv/helix/videos?', headers=headers, params=params) - responseText = response.text - responseJson = json.loads(responseText) - - for videos in responseJson['data']: - if videos['type'] == 'archive': - link = videos['url'] - started_at = videos['created_at'] - title = videos['title'] - return { - 'link': link, - 'started_at': started_at, - 'title': title, - } - -def dateStream(): - import datetime - import dateutil.parser - - started_at = getVideo()['started_at'] - - parsedDate = dateutil.parser.isoparse(started_at) - parsedDate = str(parsedDate).split("+", 1) - parsedDate = parsedDate[0] - parsedDate = datetime.datetime.strptime(parsedDate, "%Y-%m-%d %H:%M:%S") - - return { - 'day': parsedDate.day, - 'month': parsedDate.month, - 'year': parsedDate.year - } diff --git a/functions/twitchFunctions.py b/functions/twitchFunctions.py new file mode 100644 index 0000000..c6b906a --- /dev/null +++ b/functions/twitchFunctions.py @@ -0,0 +1,116 @@ +from twitchAPI.twitch import Twitch +from twitchAPI.eventsub.webhook import EventSubWebhook +from functions.botFunctions import compareImages, downloadImageGame, gamesPlayed, linkTwitchTracker, printEvent +from functions.twitterFunctions import postTweet, postTweetWithImage +from utils import variables +from random import choice + +twitch = None + +async def connectTwitch(): + global twitch + twitch = await Twitch(variables.app_key, variables.app_secret) + printEvent(True, 'twitch_authenticated') + +async def connectEventSub(): + user_id = None + async for user in twitch.get_users(logins=[variables.streamer_nickname]): + user_id = user.id + + hook = EventSubWebhook(variables.webhook_url, variables.port, twitch) + await hook.unsubscribe_all() + hook.start() + + await hook.listen_stream_online(user_id, stream_online) + await hook.listen_stream_offline(user_id, stream_offline) + await hook.listen_channel_update(user_id, channel_update) + printEvent(True, 'event_sub') + return hook + +async def verifyStreamIsOnline(): + try: + user_id = None + async for user in twitch.get_users(logins=[variables.streamer_nickname]): + user_id = user.id + + found = False + async for stream in twitch.get_streams(user_id=user_id): + variables.title_stream = stream.title + variables.category_name = stream.game_name + variables.stream_id = stream.id + found = True + break + + if found: + if variables.category_name not in variables.games_played and \ + variables.category_name not in variables.games_blacklist: + variables.games_played.append(variables.category_name) + variables.online = True + printEvent(True, 'info_stream') + else: + variables.online = False + + except Exception: + variables.online = False + +async def stream_online(data): + status = f'{variables.streamer_nickname} entrou ao vivo! {variables.title_stream}\n\ntwitch.tv/{variables.streamer_nickname}' + try: + postTweet(status) + printEvent(True, 'live_on') + except: + try: + postTweet(f'A stream provavelmente caiu, mas tá de volta -> twitch.tv/{variables.streamer_nickname}') + printEvent(True, 'live_on') + except: + emoji = ('🌹', '✨', '🍎') + postTweet(f'A stream provavelmente caiu, mas tá de volta -> twitch.tv/{variables.streamer_nickname} ({choice(emoji)}') + printEvent(True, 'live_on') + + variables.online = True + +async def stream_offline(data): + from asyncio import sleep + emoji = ('🌹', '✨', '🍎') + status = f'{variables.streamer_nickname} encerrou a live!' + + try: + postTweet(status) + await sleep(1) + printEvent(True, 'live_off') + except: + postTweet(f'{status} ({choice(emoji)})') + printEvent(True, 'live_off') + + variables.online = False + +async def channel_update(data): + if variables.title_stream != data.event.title and variables.online is True: + variables.title_stream = data.event.title + return + + if variables.title_stream != data.event.title and variables.online is False: + variables.title_stream = data.event.title + postTweet(f'[TÍTULO] {variables.title_stream}') + printEvent(True, 'title') + + if variables.category_name != data.event.category_name and variables.online is True: + variables.category_name = data.event.category_name + status = f'{variables.streamer_nickname} está jogando: {variables.category_name}\ntwitch.tv/{variables.streamer_nickname}' + + box_art_url = None + async for game in twitch.get_games(names=[variables.category_name]): + box_art_url = game.box_art_url.replace('{width}', '600').replace('{height}', '800') + break + + if box_art_url: + downloadImageGame(box_art_url) + + if variables.category_name not in variables.games_blacklist: + variables.games_played.append(variables.category_name) + + if compareImages() is False and variables.category_name not in variables.games_blacklist: + postTweetWithImage(status, 'imageGame.jpg') + else: + postTweet(status) + printEvent(True, 'game_changed') diff --git a/functions/twitterFunctions.py b/functions/twitterFunctions.py new file mode 100644 index 0000000..46cd5ca --- /dev/null +++ b/functions/twitterFunctions.py @@ -0,0 +1,38 @@ +import time +import tweepy +from utils import variables +from functions.botFunctions import printEvent + + +client = tweepy.Client( + consumer_key=variables.consumer_key, + consumer_secret=variables.consumer_secret, + access_token=variables.twitter_access_token, + access_token_secret=variables.twitter_access_secret, +) + +def _post(text: str, retries: int = 3, delay: int = 15) -> None: + for attempt in range(retries): + try: + client.create_tweet(text=text) + return + except tweepy.BadRequest: + printEvent(True, 'twitter_duplicate_tweet') + return + except tweepy.TooManyRequests: + printEvent(True, 'twitter_rate_limit') + time.sleep(60) + except tweepy.TwitterServerError: + if attempt < retries - 1: + printEvent(True, f'twitter_server_error_retry_{attempt + 1}') + time.sleep(delay) + else: + raise + +def postTweet(status: str) -> None: + _post(status) + printEvent(True, 'twitter_post') + +def postTweetWithImage(status: str, image: str) -> None: + _post(status) + printEvent(True, 'twitter_post') diff --git a/requirements.txt b/requirements.txt index 7746c06..8d78d2d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,9 @@ -pycurl==7.45.1 -python-dateutil==2.8.2 -python-dotenv==0.20.0 -requests==2.27.1 +tweepy==4.14.0 +twitchAPI>=4.5.0 +aiohttp==3.10.5 +python-dotenv==1.0.1 +requests==2.32.3 +websockets==12.0 +python-dateutil==2.9.0 +certifi==2024.8.30 requests-oauthlib==1.3.1 -tweepy==4.10.0 -twitchAPI==2.5.5 -urllib3==1.26.9 \ No newline at end of file diff --git a/tests/unit/test_stream_state.py b/tests/unit/test_stream_state.py new file mode 100644 index 0000000..9216c39 --- /dev/null +++ b/tests/unit/test_stream_state.py @@ -0,0 +1,34 @@ +from core.stream_state import StreamState + + +def test_default_state_values(): + state = StreamState() + + assert state.online is False + assert state.current_game == '' + assert state.stream_title == '' + + +def test_update_from_update_changes_state(): + state = StreamState() + update = { + 'online': True, + 'current_game': 'Umineko When They Cry', + 'stream_title': 'The Golden Witch is Real' + } + + state.update_from_update(update) + + assert state.online is True + assert state.current_game == 'Umineko When They Cry' + assert state.stream_title == 'The Golden Witch is Real' + + +def test_reset_for_offline_sets_online_false(): + state = StreamState(online=True, current_game='Umineko When They Cry', stream_title='The Golden Witch is Real') + + state.reset_for_offline() + + assert state.online is False + assert state.current_game == 'Umineko When They Cry' + assert state.stream_title == 'The Golden Witch is Real' diff --git a/utils/variables.py b/utils/variables.py new file mode 100644 index 0000000..4681f92 --- /dev/null +++ b/utils/variables.py @@ -0,0 +1,24 @@ +import os +from dotenv import load_dotenv +load_dotenv() + +# Enviroment Variables +app_key = os.environ.get('TWITCH_APP_ID') +app_secret = os.environ.get('TWITCH_APP_SECRET') +streamer_nickname = os.environ.get('TARGET_USERNAME', '').capitalize() +webhook_url = os.environ.get('WEBHOOK_URL') +port = int(os.environ.get('PORT', 8080)) +consumer_key = os.environ.get('TWITTER_CONSUMER_KEY') +consumer_secret = os.environ.get('TWITTER_CONSUMER_SECRET') +twitter_access_token = os.environ.get('TWITTER_ACCESS_TOKEN') +twitter_access_secret = os.environ.get('TWITTER_ACCESS_SECRET') + +# Variables Bot +games_played = list() +games_blacklist = ('Just Chatting', 'Tabletop RPGs', 'Watch Parties') + +# Stream Variables +online: bool = False +title_stream = str +category_name = str +stream_id = int