Implement the bet system. Based on https://github.com/ClementRoyer/TwitchAutoCollect-AutoBet - Use selenium for place the bet/predictions. We can also use directly grapql but we need a transactionID for the POST request but I don't know how to calculate it (or if It's random - but I don't think)

This commit is contained in:
Alessandro Maggio
2021-01-13 17:00:34 +01:00
parent cbac5db4b6
commit cbd482e24c
8 changed files with 381 additions and 4 deletions

View File

@@ -5,6 +5,7 @@ from TwitchChannelPointsMiner.classes.WebSocketsPool import WebSocketsPool
from TwitchChannelPointsMiner.classes.PubsubTopic import PubsubTopic
from TwitchChannelPointsMiner.classes.Streamer import Streamer
from TwitchChannelPointsMiner.classes.Twitch import Twitch
from TwitchChannelPointsMiner.classes.TwitchBrowser import TwitchBrowser, Browser
from TwitchChannelPointsMiner.classes.Exceptions import StreamerDoesNotExistException
logging.basicConfig(
@@ -22,6 +23,7 @@ class TwitchChannelPointsMiner:
raid: bool = True,
):
self.twitch = Twitch(username)
self.twitch_browser = False
self.predictions = predictions
self.raid = raid
self.streamers = []
@@ -48,12 +50,16 @@ class TwitchChannelPointsMiner:
self.twitch.check_streamer_online(streamer)
# logger.info(streamer)
if self.predictions is True: # We need a browser to make predictions / bet
self.twitch_browser = TwitchBrowser(self.twitch.twitch_login.get_auth_token())
self.twitch_browser.init(show=True, browser=Browser.FIREFOX)
self.minute_watcher_thread = threading.Thread(
target=self.twitch.send_minute_watched_events, args=(self.streamers,)
)
self.minute_watcher_thread.start()
self.ws_pool = WebSocketsPool(twitch=self.twitch, streamers=self.streamers)
self.ws_pool = WebSocketsPool(twitch=self.twitch, twitch_browser=self.twitch_browser, streamers=self.streamers)
topics = [
PubsubTopic(
"community-points-user-v1", user_id=self.twitch.twitch_login.get_user_id()

View File

@@ -0,0 +1,63 @@
import logging
from enum import Enum, auto
logger = logging.getLogger(__name__)
class Strategy(Enum):
MOST_VOTED = auto()
HIGH_COTE = auto()
SMART = auto()
class Bet:
def __init__(self, outcomes):
self.outcomes = outcomes
self.__clear_outcomes()
self.total_users = 0
self.total_points = 0
def update_outcomes(self, outcomes):
self.outcomes[0]["total_users"] = int(outcomes[0]["total_users"])
self.outcomes[1]["total_users"] = int(outcomes[1]["total_users"])
self.outcomes[0]["total_points"] = int(outcomes[0]["total_points"])
self.outcomes[1]["total_points"] = int(outcomes[1]["total_points"])
self.total_users = self.outcomes[0]["total_users"] + self.outcomes[1]["total_users"]
self.total_points = self.outcomes[0]["total_points"] + self.outcomes[1]["total_points"]
if (self.total_users > 0 and self.outcomes[0]["total_points"] > 0 and self.outcomes[1]["total_points"] > 0):
self.outcomes[0]["percentage_users"] = round(float((100 * self.outcomes[0]["total_users"]) / self.total_users), 2)
self.outcomes[1]["percentage_users"] = round(float((100 * self.outcomes[1]["total_users"]) / self.total_users), 2)
self.outcomes[0]["cote"] = round(float(self.total_points / self.outcomes[0]["total_points"]), 2)
self.outcomes[1]["cote"] = round(float(self.total_points / self.outcomes[1]["total_points"]), 2)
self.__clear_outcomes()
def __repr__(self):
return f"Bet(Outcome0={self.outcomes[0]}, Outcome1={self.outcomes[1]}, total_users={self.total_users}, total_points={self.total_points})"
def __clear_outcomes(self):
for index in range(0, len(self.outcomes)):
for key in self.outcomes[index]:
if key not in ["total_users", "total_points", "percentage_users", "cote", "title", "color"]:
del self.outcomes[index][key]
def calculate(self, balance: int, strategy: Strategy = Strategy.SMART, percentage=5, percentage_gap=20, max_points=50000):
output = {"choice": "", "amount": 0}
if strategy == Strategy.MOST_VOTED:
output['choice'] = "A" if self.outcomes[0]["total_users"] > self.outcomes[1]["total_users"] else "B"
elif strategy == Strategy.HIGH_COTE:
output['choice'] = "A" if self.outcomes[0]["cote"] > self.outcomes[1]["cote"] else "B"
elif strategy == Strategy.SMART:
difference_percentage = (self.outcomes[0]["percentage_users"] - self.outcomes[1]["percentage_users"]) if self.outcomes[0]["percentage_users"] > self.outcomes[1]["percentage_users"] else (self.outcomes[1]["percentage_users"] - self.outcomes[0]["percentage_users"])
if difference_percentage < percentage_gap:
output['choice'] = "A" if self.outcomes[0]["cote"] > self.outcomes[1]["cote"] else "B"
else:
output['choice'] = "A" if self.outcomes[0]["total_users"] > self.outcomes[1]["total_users"] else "B" # Follow other users
output['amount'] = min(round(balance * (percentage / 100)), max_points)
return output

View File

@@ -0,0 +1,25 @@
from TwitchChannelPointsMiner.classes.Bet import Bet
from TwitchChannelPointsMiner.classes.Streamer import Streamer
class EventPrediction:
def __init__(self, streamer: Streamer, event_id, title, created_at, prediction_window_seconds, status, outcomes):
self.streamer = streamer
self.event_id = event_id
self.title = title
self.created_at = created_at
self.prediction_window_seconds = prediction_window_seconds
self.status = status
self.box_fillable = False
self.bet = Bet(outcomes)
def __repr__(self):
return f"EventPrediction(event_id={self.event_id}, title={self.title})"
def elapsed(self, timestamp):
return round(float((timestamp - self.created_at).total_seconds()), 2)
def closing_bet_after(self, timestamp):
return round(float(self.prediction_window_seconds - self.elapsed(timestamp)), 2)

View File

@@ -20,7 +20,7 @@ class Streamer:
self.raid = None
self.streamer_url = f"https://www.twitch.tv/{self.username}"
self.char_url = f"https://www.twitch.tv/popout/{self.username}/chat?popout="
self.chat_url = f"https://www.twitch.tv/popout/{self.username}/chat?popout="
def __repr__(self):
return f"Streamer(username={self.username}, channel_id={self.channel_id}, channel_points={self.channel_points})"

View File

@@ -114,7 +114,7 @@ class Twitch:
streamer.set_online()
def claim_channel_points_bonus(self, streamer, claim_id):
logger.info(f"🎁 Claiming the bonus for {streamer.username}!")
logger.info(f"🎁 Claiming the bonus for {streamer}!")
json_data = {
"operationName": "ClaimCommunityPoints",
"variables": {

View File

@@ -0,0 +1,244 @@
import time
import logging
import random
from datetime import datetime
from enum import Enum, auto
from PIL import Image, ImageDraw, ImageFont
from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.common.by import By
from selenium.webdriver.common.action_chains import ActionChains
from TwitchChannelPointsMiner.classes.EventPrediction import EventPrediction
TWITCH_URL = 'https://www.twitch.tv/'
cookiePolicyQuery = 'button[data-a-target="consent-banner-accept"]'
streamCoinsMenu = '//div[@data-test-selector="community-points-summary"]//button'
streamBetTitleInBet = '[data-test-selector="predictions-list-item__title"]'
streamBetCustomVote = '[data-test-selector="prediction-checkout-active-footer__input-type-toggle"]'
streamBetVoteInput = "(//input[contains(@class,'tw-block tw-border-bottom-left-radius-medium')])"
streamBetVoteButton = "(//div[@id='channel-points-reward-center-body']//button)"
logger = logging.getLogger(__name__)
class Browser(Enum):
CHROME = auto()
FIREFOX = auto()
class TwitchBrowser:
def __init__(
self,
auth_token: str,
timeout: float = 5.0
):
self.auth_token = auth_token
self.timeout = timeout
self.currently_is_betting = False
self.browser = None
def init(self, show: bool = True, browser: Browser = Browser.FIREFOX):
if browser == Browser.FIREFOX:
self.__init_firefox(show)
elif browser == Browser.CHROME:
self.__init_chrome(show)
if self.browser is not None:
self.browser.set_window_size(250, 900)
self.browser.implicitly_wait(2)
self.__init_twitch()
def __init_twitch(self):
logger.info("⚙️ Init Twitch page - Cookie - LocalStorage items")
cookie = {
"domain": ".twitch.tv",
"hostOnly": False,
"httpOnly": False,
"name": "auth-token",
"path": "/",
"SameSite": "no_restriction",
"secure": True,
"session": False,
"storeId": "0",
"id": 1,
"value": self.auth_token
}
self.browser.get(TWITCH_URL)
self.browser.add_cookie(cookie)
self.__click_when_exist(cookiePolicyQuery, By.CSS_SELECTOR)
time.sleep(1.5)
# Edit value in localStorage for dark theme, point consent etc.
self.browser.execute_script("""
window.localStorage.setItem("volume", 0);
window.localStorage.setItem("channelPointsOnboardingDismissed", true);
window.localStorage.setItem("twilight.theme", 1);
window.localStorage.setItem("mature", true);
window.localStorage.setItem("rebrand-notice-dismissed", true);
""")
time.sleep(1.5)
self.__blank()
def __blank(self):
self.browser.get("about:blank")
# Private method __ - We can instantiate webdriver only with init_browser
def __init_chrome(self, show):
logger.info("⚙️ Init Chrome browser")
options = webdriver.ChromeOptions()
options.add_argument('--mute-audio')
if not show:
options.add_argument('headless')
options.add_argument('--disable-dev-shm-usage')
options.add_argument('--disable-accelerated-2d-canvas')
options.add_argument('--no-first-run')
options.add_argument('--no-zygote')
options.add_argument('--disable-gpu')
options.add_argument('--no-sandbox')
options.add_argument('--disable-setuid-sandbox')
options.add_experimental_option('excludeSwitches', ['enable-logging'])
self.browser = webdriver.Chrome(options=options)
# Private method __ - We can instantiate webdriver only with init_browser
def __init_firefox(self, show):
logger.info("⚙️ Init Firefox browser")
options = webdriver.FirefoxOptions()
if not show:
options.headless = True
fp = webdriver.FirefoxProfile()
fp.set_preference('permissions.default.image', 2)
fp.set_preference("permissions.default.stylesheet", 2)
fp.set_preference('dom.ipc.plugins.enabled.libflashplayer.so', 'false')
fp.set_preference('media.volume_scale', '0.0')
fp.set_preference("browser.startup.homepage", "about:blank")
fp.set_preference("startup.homepage_welcome_url", "about:blank")
fp.set_preference("startup.homepage_welcome_url.additional", "about:blank")
self.browser = webdriver.Firefox(options=options, firefox_profile=fp)
def screenshot(self, fname, write_timestamp=True):
self.browser.save_screenshot(fname)
if write_timestamp is True:
image = Image.open(fname)
draw = ImageDraw.Draw(image)
font = ImageFont.truetype('./Roboto/Roboto-Bold.ttf', size=35)
(x, y) = (15, image.height // 3)
datetime_text = datetime.now().strftime('%d/%m %H:%M:%S.%f')
shadowcolor = 'rgb(0, 0, 0)' # black color
# Thin border
draw.text((x - 1, y), datetime_text, font=font, fill=shadowcolor)
draw.text((x + 1, y), datetime_text, font=font, fill=shadowcolor)
draw.text((x, y - 1), datetime_text, font=font, fill=shadowcolor)
draw.text((x, y + 1), datetime_text, font=font, fill=shadowcolor)
# Thicker border
draw.text((x - 1, y - 1), datetime_text, font=font, fill=shadowcolor)
draw.text((x + 1, y - 1), datetime_text, font=font, fill=shadowcolor)
draw.text((x - 1, y + 1), datetime_text, font=font, fill=shadowcolor)
draw.text((x + 1, y + 1), datetime_text, font=font, fill=shadowcolor)
color = 'rgb(255, 255, 255)' # white color
draw.text((x, y), datetime_text, font=font, fill=color)
image.save(fname, optimize=True, quality=20)
def __click_when_exist(self, selector, by: By = By.CSS_SELECTOR):
try:
element = WebDriverWait(self.browser, self.timeout).until(expected_conditions.element_to_be_clickable((by, selector)))
ActionChains(self.browser).move_to_element(element).click().perform()
return element
except Exception:
logger.error(f"Exception raised with: {selector}", exc_info=True)
finally:
return None
def __send_text(self, selector, text, by: By = By.CSS_SELECTOR):
try:
element = WebDriverWait(self.browser, self.timeout).until(expected_conditions.element_to_be_clickable((By.XPATH, selector)))
ActionChains(self.browser, self.timeout).move_to_element(element).click().send_keys(text).perform()
return element
except Exception:
logger.error(f"Exception raised with: {selector}", exc_info=True)
finally:
return None
def start_bet(self, event: EventPrediction):
if self.currently_is_betting:
logger.info("Sorry, unable to start the bet. The browser it's currently betting another event")
else:
logger.info(f"⚙️ Start betting at {event.streamer.chat_url} for event: {event}")
self.browser.get(event.streamer.chat_url)
time.sleep(random.uniform(4, 6))
self.__open_coins_menu(event)
self.__click_on_bet(event)
self.__enable_custom_bet_value(event)
return self.currently_is_betting
def complete_bet(self, event: EventPrediction):
logger.info(f"⚙️ Going to complete bet for event {event}. Current url page: {self.browser.current_url}")
if event.box_fillable and self.currently_is_betting:
decision = event.bet.calculate(event.streamer.channel_points)
if decision["choice"]:
selector_index = "[1]" if decision["choice"] == "A" else "[2]"
try:
logger.info(f"⚙️ Going to write: {decision['amount']} on input {decision['choice']}")
self.__send_text(streamBetVoteInput + selector_index, decision["amount"], By.XPATH)
if self.do_screenshot:
self.screenshot("./tmp/{}_TEXT_WRITE.png".format(event.event_id))
logger.info("Going to place the bet")
self.__click_when_exist(streamBetVoteButton + selector_index, By.XPATH)
if self.do_screenshot:
self.screenshot("./tmp/{}_CLICK_ON_VOTE.png".format(event.event_id))
time.sleep(random.uniform(15, 25))
event.bet_completed = True
self.browser.get("about:blank")
self.currently_is_betting = False
except Exception:
logger.error("Exception raised", exc_info=True)
self.browser.get("about:blank")
self.currently_is_betting = False
else:
logger.info(f"Sorry, unable to complete the bet. Event box fillable: {event.box_fillable}, the browser is betting: {self.currently_is_betting}")
def __open_coins_menu(self, event: EventPrediction):
logger.info(f"⚙️ Open coins menu for event: {event}")
self.__click_when_exist(streamCoinsMenu, By.XPATH)
time.sleep(random.uniform(0.05, 0.1))
if self.do_screenshot:
self.screenshot("./tmp/{}_OPEN_COINS.png".format(event.event_id))
def __click_on_bet(self, event):
logger.info(f"⚙️ Click on the bet for event: {event}")
self.__click_when_exist(streamBetTitleInBet, By.CSS_SELECTOR)
time.sleep(random.uniform(0.05, 0.1))
if self.do_screenshot:
self.screenshot("./tmp/{}_CLICK_ON_BET.png".format(event.event_id))
def __enable_custom_bet_value(self, event):
logger.info(f"⚙️ Enable input of custom value for event: {event}")
if self.__click_when_exist(streamBetCustomVote, By.CSS_SELECTOR) is not None:
time.sleep(random.uniform(0.05, 0.1))
if self.do_screenshot:
self.screenshot("./tmp/{}_INPUT_ENABLE.png".format(event.event_id))
event.box_fillable = True
self.currently_is_betting = True
else:
logger.info("Something whent wrong unable to continue with betting - Fillable box not avaible")

View File

@@ -3,6 +3,9 @@ import threading
import time
import json
from dateutil import parser
from TwitchChannelPointsMiner.classes.EventPrediction import EventPrediction
from TwitchChannelPointsMiner.classes.Raid import Raid
from TwitchChannelPointsMiner.classes.TwitchWebSocket import TwitchWebSocket
@@ -19,9 +22,10 @@ def get_streamer_index(streamers, channel_id):
# You can't listen for more than 50 topics at once
class WebSocketsPool:
def __init__(self, twitch, streamers):
def __init__(self, twitch, twitch_browser, streamers):
self.ws = None
self.twitch = twitch
self.twitch_browser = twitch_browser
self.streamers = streamers
def submit(self, topic):
@@ -49,7 +53,9 @@ class WebSocketsPool:
self.ws.pending_topics = []
self.ws.twitch = self.twitch
self.ws.twitch_browser = self.twitch_browser
self.ws.streamers = self.streamers
self.ws.events_predictions = {}
self.ws.last_message_time = 0
self.ws.last_message_type = None
@@ -136,6 +142,36 @@ class WebSocketsPool:
raid = Raid(message["raid"]["id"], message["raid"]["target_login"])
ws.twitch.update_raid(ws.streamers[streamer_index], raid)
elif topic == "predictions-channel-v1":
streamer_index = get_streamer_index(ws.streamers, topic_user) # message_data["event"]["channel_id"]
event_id = message_data["event"]["id"]
event_status = message_data["event"]["status"]
current_timestamp = parser.parse(message_data["timestamp"])
if event_id not in ws.events_predictions:
if event_status == "ACTIVE":
event = EventPrediction(
ws.streamers[streamer_index],
event_id,
parser.parse(message_data["event"]["created_at"]),
float(message_data["event"]["prediction_window_seconds"]) - 20,
event_status,
message_data["event"]["outcomes"]
)
if event.closing_bet_after(current_timestamp) > 0:
ws.events_predictions[event_id] = event
if ws.twitch_browser.currently_is_betting is False:
if ws.twitch_browser.start_bet(ws.events_predictions[event_id]):
complete_bet_thread = threading.Timer(event.closing_bet_after(current_timestamp), ws.twitch_browser.complete_bet, (ws.events_predictions[event_id],))
complete_bet_thread.start()
logger.info(f"⏰ A thread should start and place the bet after: {event.closing_bet_after(current_timestamp)}s for the event: {ws.events_predictions[event_id]}")
else:
ws.events_predictions[event_id].status = event_status
ws.events_predictions[event_id].bet.update_outcomes(message_data["event"]["outcomes"])
elif response["type"] == "RESPONSE" and len(response.get("error", "")) > 0:
raise RuntimeError(f"Error while trying to listen for a topic: {response}")

View File

@@ -1,3 +1,6 @@
requests
websocket-client
browser_cookie3
pillow
selenium
python-dateutil