majjo/Twitch-Drops-Miner
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #36 from Tkd-Alex/dedicated-settings-refactory

Browse Source 
Dedicated settings for each streamer
This commit is contained in:
Alessandro Maggio
2021-02-01 22:43:21 +01:00
committed by GitHub
parent 2d1997b686 4bda13c887
commit e13cd4b46b
22 changed files with 308 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
README.md
View File

@@ -158,16 +158,13 @@ For the bet system the script use Selenium. Could be usefull understand how to M
import logging
from TwitchChannelPointsMiner import TwitchChannelPointsMiner
from TwitchChannelPointsMiner.classes.Logger import LoggerSettings
from TwitchChannelPointsMiner.classes.Bet import Strategy, BetSettings
from TwitchChannelPointsMiner.logger import LoggerSettings
from TwitchChannelPointsMiner.classes.entities.Bet import Strategy, BetSettings
from TwitchChannelPointsMiner.classes.entities.Streamer import Streamer, StreamerSettings
from TwitchChannelPointsMiner.classes.TwitchBrowser import Browser, BrowserSettings
twitch_miner = TwitchChannelPointsMiner(
username="your-twitch-username",
make_predictions=True, # If you want to Bet / Make prediction | The browser will never start
follow_raid=True, # Follow raid to obtain more points
watch_streak=True, # If a streamer go online change the priotiry of streamers array and catch the watch screak. Issue #11
drops_events=True, # If you want to auto claim game drops from Twitch inventory Issue #21
claim_drops_startup=False, # If you want to auto claim all drops from Twitch inventory on startup
logger_settings=LoggerSettings(
save=True, # If you want to save logs in file (suggested)
@@ -181,17 +178,43 @@ twitch_miner = TwitchChannelPointsMiner(
show=False, # Show the browser during bet else headless mode
do_screenshot=False, # Do screenshot during the bet
),
bet_settings=BetSettings(
strategy=Strategy.SMART, # Choose you strategy!
percentage=5, # Place the x% of your channel points
percentage_gap=20, # Gap difference between outcomesA and outcomesB (for SMART stragegy)
max_points=50000, # If the x percentage of your channel points is gt bet_max_points set this value
streamer_settings=StreamerSettings(
make_predictions=True, # If you want to Bet / Make prediction
follow_raid=True, # Follow raid to obtain more points
claim_drops=True, # We can't filter rewards base on stream. Set to False for skip viewing counter increase and you will never obtain a drop reward from this script. Issue #21
watch_streak=True, # If a streamer go online change the priotiry of streamers array and catch the watch screak. Issue #11
bet=BetSettings(
strategy=Strategy.SMART, # Choose you strategy!
percentage=5, # Place the x% of your channel points
percentage_gap=20, # Gap difference between outcomesA and outcomesB (for SMART stragegy)
max_points=50000, # If the x percentage of your channel points is gt bet_max_points set this value
)
)
)
# You can customize the settings for each streamer. If not settings was provided the script will use the streamer_settings from TwitchChannelPointsMiner.
# If no streamer_settings provided in TwitchChannelPointsMiner the script will use default settings.
# The streamers array can be a String -> username or Streamer instance.
# The settings priority are: settings in mine function, settings in TwitchChannelPointsMiner instance, default settings.
# For example if in the mine function you don't provide any value for 'make_prediction' but you have set it on TwitchChannelPointsMiner instance the script will take the value from here.
# If you haven't set any value even in the instance the default one will be used
twitch_miner.mine(
["streamer1", "streamer2"], # Array of streamers (order = priority)
followers=False # Automatic download the list of your followers
[
Streamer("streamer-username01", settings=StreamerSettings(make_predictions=True , follow_raid=False , claim_drops=True , watch_streak=True , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , percentage_gap=20 , max_points=234 ) )),
Streamer("streamer-username02", settings=StreamerSettings(make_predictions=False , follow_raid=True , claim_drops=False , bet=BetSettings(strategy=Strategy.PERCENTAGE , percentage=5 , percentage_gap=20 , max_points=1234 ) )),
Streamer("streamer-username03", settings=StreamerSettings(make_predictions=True , follow_raid=False , watch_streak=True , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , percentage_gap=30 , max_points=50000 ) )),
Streamer("streamer-username04", settings=StreamerSettings(make_predictions=False , follow_raid=True , watch_streak=True )),
Streamer("streamer-username05", settings=StreamerSettings(make_predictions=True , follow_raid=True , claim_drops=True , watch_streak=True , bet=BetSettings(strategy=Strategy.ODDS , percentage=7 , percentage_gap=20 , max_points=90 ) )),
Streamer("streamer-username06"),
Streamer("streamer-username07"),
Streamer("streamer-username08"),
"streamer-username09",
"streamer-username10",
"streamer-username11"
], # Array of streamers (order = priority)
followers=False # Automatic download the list of your followers (unable to set custom settings for you followers list)
)
```
You can also use all the default values except for your username obv. Short version:
@@ -218,7 +241,7 @@ If the browser are currently betting or wait for more data It's impossible to in
- **PERCENTAGE**: Select the option with the highest percentage based on odds (It's the same that show Twitch) - Should be the same of select LOWEST_ODDS
- **SMART**: If the majority in percent chose an option then follow the other users, otherwise choose the option with the highest odds
![Screenshot](./assets/prediction_screen.png)
![Screenshot](./assets/prediction.png)
Here a concrete example:

188
TwitchChannelPointsMiner/TwitchChannelPointsMiner.py
View File

@@ -11,18 +11,26 @@ import uuid
from collections import OrderedDict
from datetime import datetime
from TwitchChannelPointsMiner.classes.Bet import BetSettings
from TwitchChannelPointsMiner.classes.entities.PubsubTopic import PubsubTopic
from TwitchChannelPointsMiner.classes.entities.Streamer import (
Streamer,
StreamerSettings,
)
from TwitchChannelPointsMiner.classes.Exceptions import StreamerDoesNotExistException
from TwitchChannelPointsMiner.classes.Logger import LoggerSettings, configure_loggers
from TwitchChannelPointsMiner.classes.PubsubTopic import PubsubTopic
from TwitchChannelPointsMiner.classes.Streamer import Streamer
from TwitchChannelPointsMiner.classes.Settings import Settings
from TwitchChannelPointsMiner.classes.Twitch import Twitch
from TwitchChannelPointsMiner.classes.TwitchBrowser import (
BrowserSettings,
TwitchBrowser,
)
from TwitchChannelPointsMiner.classes.WebSocketsPool import WebSocketsPool
from TwitchChannelPointsMiner.utils import _millify, get_user_agent
from TwitchChannelPointsMiner.logger import LoggerSettings, configure_loggers
from TwitchChannelPointsMiner.utils import (
_millify,
at_least_one_value_in_settings_is,
get_user_agent,
set_default_settings,
)
# Suppress warning for urllib3.connectionpool (selenium close connection)
# Suppress also the selenium logger please
@@ -36,45 +44,44 @@ class TwitchChannelPointsMiner:
def __init__(
self,
username: str,
make_predictions: bool = True,
follow_raid: bool = True,
watch_streak: bool = False,
claim_drops_startup: bool = False,
drops_events: bool = False,
# Settings for logging and selenium as you can see.
# This settings will be global shared trought Settings class
logger_settings: LoggerSettings = LoggerSettings(),
browser_settings: BrowserSettings = BrowserSettings(),
bet_settings: BetSettings = BetSettings(),
# Default values for all streamers
streamer_settings: StreamerSettings = StreamerSettings(),
):
self.username = username
self.browser_settings = browser_settings
self.bet_settings = bet_settings
self.twitch = Twitch(
self.username, get_user_agent(self.browser_settings.browser)
)
# Set as globally config
Settings.logger = logger_settings
Settings.browser = browser_settings
# Init as default all the missing values
streamer_settings.default()
streamer_settings.bet.default()
Settings.streamer_settings = streamer_settings
user_agent = get_user_agent(browser_settings.browser)
self.twitch = Twitch(self.username, user_agent)
self.twitch_browser = None
self.follow_raid = follow_raid
self.watch_streak = watch_streak
self.drops_events = drops_events
self.claim_drops_startup = claim_drops_startup
self.streamers = []
self.events_predictions = {}
self.minute_watcher_thread = None
self.ws_pool = None
self.make_predictions = make_predictions
self.session_id = str(uuid.uuid4())
self.running = False
self.start_datetime = None
self.original_streamers = []
self.logger_settings = logger_settings
self.logs_file = configure_loggers(self.username, self.logger_settings)
self.logs_file = configure_loggers(self.username, logger_settings)
signal.signal(signal.SIGINT, self.end)
signal.signal(signal.SIGSEGV, self.end)
signal.signal(signal.SIGTERM, self.end)
for sign in [signal.SIGINT, signal.SIGSEGV, signal.SIGTERM]:
signal.signal(sign, self.end)
def mine(self, streamers: list = [], followers=False):
self.run(streamers, followers)
@@ -94,57 +101,86 @@ class TwitchChannelPointsMiner:
if self.claim_drops_startup is True:
self.twitch.claim_all_drops_from_inventory()
# Clear streamers array
# Remove duplicate 3. Preserving Order: Use OrderedDict (askpython .com)
streamers = [streamer_name.lower().strip() for streamer_name in streamers]
streamers = list(OrderedDict.fromkeys(streamers))
streamers_name: list = []
streamers_dict: dict = {}
for streamer in streamers:
username = (
streamer.username
if isinstance(streamer, Streamer)
else streamer.lower().strip()
)
streamers_name.append(username)
streamers_dict[username] = streamer
if followers is True:
# Append at the end with lowest priority
followers_array = self.twitch.get_followers()
logger.info(
f"Loading {len(followers_array)} followers from your profile!",
f"Load {len(followers_array)} followers from your profile!",
extra={"emoji": ":clipboard:"},
)
streamers += [fw for fw in followers_array if fw not in streamers]
for username in followers_array:
if username not in streamers_dict:
streamers_dict[username] = username.lower().strip()
else:
followers_array = []
streamers_name = list(
OrderedDict.fromkeys(streamers_name + followers_array)
)
logger.info(
f"Loading data for {len(streamers)} streamers. Please wait...",
f"Loading data for {len(streamers_name)} streamers. Please wait...",
extra={"emoji": ":nerd_face:"},
)
for streamer_username in streamers:
for username in streamers_name:
time.sleep(random.uniform(0.3, 0.7))
streamer_username.lower().strip()
try:
channel_id = self.twitch.get_channel_id(streamer_username)
streamer = Streamer(
streamer_username,
channel_id,
less_printing=self.logger_settings.less,
if isinstance(streamers_dict[username], Streamer) is True:
streamer = streamers_dict[username]
else:
streamer = Streamer(username)
streamer.channel_id = self.twitch.get_channel_id(username)
streamer.settings = set_default_settings(
streamer.settings, Settings.streamer_settings
)
streamer.settings.bet = set_default_settings(
streamer.settings.bet, Settings.streamer_settings.bet
)
self.streamers.append(streamer)
except StreamerDoesNotExistException:
logger.info(
f"Streamer {streamer_username} does not exist",
f"Streamer {username} does not exist",
extra={"emoji": ":cry:"},
)
# Populate the streamers with default values.
# 1. Load channel points and auto-claim bonus
# 2. Check if streamers is online
# 3. Check if the user is a Streamer. In thi case you can't do prediction
for streamer in self.streamers:
time.sleep(random.uniform(0.3, 0.7))
self.twitch.load_channel_points_context(
streamer, less_printing=self.logger_settings.less
)
self.twitch.load_channel_points_context(streamer)
self.twitch.check_streamer_online(streamer)
self.twitch.viewer_is_mod(streamer)
if streamer.viewer_is_mod is True:
streamer.settings.make_predictions = False
self.original_streamers = copy.deepcopy(self.streamers)
if (
self.make_predictions is True
): # We need a browser to make predictions / bet
# If we have at least one streamer with settings = make_predictions True
make_predictions = at_least_one_value_in_settings_is(
self.streamers, "make_predictions", True
)
# We need a browser to make predictions / bet
if make_predictions is True:
self.twitch_browser = TwitchBrowser(
self.twitch.twitch_login.get_auth_token(),
self.session_id,
settings=self.browser_settings,
settings=Settings.browser,
)
self.twitch_browser.init()
@@ -152,37 +188,44 @@ class TwitchChannelPointsMiner:
target=self.twitch.send_minute_watched_events,
args=(
self.streamers,
self.watch_streak,
at_least_one_value_in_settings_is(
self.streamers, "watch_streak", True
),
),
)
# self.minute_watcher_thread.daemon = True
self.minute_watcher_thread.start()
self.ws_pool = WebSocketsPool(
twitch=self.twitch,
twitch_browser=self.twitch_browser,
browser=self.twitch_browser,
streamers=self.streamers,
bet_settings=self.bet_settings,
events_predictions=self.events_predictions,
less_printing=self.logger_settings.less,
)
topics = [
# Subscribe to community-points-user. Get update for points spent or gains
self.ws_pool.submit(
PubsubTopic(
"community-points-user-v1",
user_id=self.twitch.twitch_login.get_user_id(),
)
]
)
if self.drops_events is True:
topics.append(
# If we have at least one streamer with settings = claim_drops True
# Going to subscribe to user-drop-events. Get update for drop-progress
claim_drops = at_least_one_value_in_settings_is(
self.streamers, "claim_drops", True
)
if claim_drops is True:
self.ws_pool.submit(
PubsubTopic(
"user-drop-events",
user_id=self.twitch.twitch_login.get_user_id(),
),
)
)
if self.make_predictions is True:
topics.append(
# Going to subscribe to predictions-user-v1. Get update when we place a new prediction (confirm)
if make_predictions is True:
self.ws_pool.submit(
PubsubTopic(
"predictions-user-v1",
user_id=self.twitch.twitch_login.get_user_id(),
@@ -190,19 +233,18 @@ class TwitchChannelPointsMiner:
)
for streamer in self.streamers:
topics.append(PubsubTopic("video-playback-by-id", streamer=streamer))
self.ws_pool.submit(
PubsubTopic("video-playback-by-id", streamer=streamer)
)
if self.follow_raid is True:
topics.append(PubsubTopic("raid", streamer=streamer))
if streamer.settings.follow_raid is True:
self.ws_pool.submit(PubsubTopic("raid", streamer=streamer))
if self.make_predictions is True:
topics.append(
if streamer.settings.make_predictions is True:
self.ws_pool.submit(
PubsubTopic("predictions-channel-v1", streamer=streamer)
)
for topic in topics:
self.ws_pool.submit(topic)
while self.running:
time.sleep(random.uniform(20, 60))
# Do an external control for WebSocket. Check if the thread is running
@@ -243,11 +285,12 @@ class TwitchChannelPointsMiner:
)
if self.make_predictions:
print("")
logger.info(f"{self.bet_settings}", extra={"emoji": ":bar_chart:"})
for event_id in self.events_predictions:
if self.events_predictions[event_id].bet_confirmed is True:
self.events_predictions[event_id].set_less_printing(False)
logger.info(
f"{self.events_predictions[event_id].streamer.bet.settings}",
extra={"emoji": ":bar_chart:"},
)
logger.info(
f"{self.events_predictions[event_id].print_recap()}",
extra={"emoji": ":bar_chart:"},
@@ -255,9 +298,8 @@ class TwitchChannelPointsMiner:
print("")
for streamer_index in range(0, len(self.streamers)):
self.streamers[streamer_index].set_less_printing(False)
logger.info(
f"{self.streamers[streamer_index]}, Total points gained (after farming - before farming): {_millify(self.streamers[streamer_index].channel_points - self.original_streamers[streamer_index].channel_points)}",
f"{repr(self.streamers[streamer_index])}, Total Points Gained (after farming - before farming): {_millify(self.streamers[streamer_index].channel_points - self.original_streamers[streamer_index].channel_points)}",
extra={"emoji": ":robot:"},
)
if self.streamers[streamer_index].history != {}:

2
TwitchChannelPointsMiner/__init__.py
View File

@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
__version__ = "2.4.1"
__version__ = "2.5.0"
from .TwitchChannelPointsMiner import TwitchChannelPointsMiner
__all__ = [

3
TwitchChannelPointsMiner/classes/Settings.py Normal file
View File

@@ -0,0 +1,3 @@
# Empty object shared between class
class Settings(object):
pass

21
TwitchChannelPointsMiner/classes/Twitch.py
View File

@@ -19,6 +19,7 @@ from TwitchChannelPointsMiner.classes.Exceptions import (
StreamerIsOfflineException,
TimeBasedDropNotFound,
)
from TwitchChannelPointsMiner.classes.Settings import Settings
from TwitchChannelPointsMiner.classes.TwitchLogin import TwitchLogin
from TwitchChannelPointsMiner.constants.twitch import API, CLIENT_ID, GQLOperations
@@ -60,7 +61,10 @@ class Twitch:
"user_id": self.twitch_login.get_user_id(),
}
if streamer.stream.game_name() is not None:
if (
streamer.stream.game_name() is not None
and streamer.settings.claim_drops is True
):
event_properties["game"] = streamer.stream.game_name()
streamer.stream.payload = [
@@ -131,8 +135,8 @@ class Twitch:
except StreamerIsOfflineException:
streamer.set_offline()
def claim_bonus(self, streamer, claim_id, less_printing=False):
if less_printing is False:
def claim_bonus(self, streamer, claim_id):
if Settings.logger.less is False:
logger.info(
f"Claiming the bonus for {streamer}!", extra={"emoji": ":gift:"}
)
@@ -178,7 +182,7 @@ class Twitch:
return response["data"]["currentUser"]["inventory"]
# Load the amount of current points for a channel, check if a bonus is available
def load_channel_points_context(self, streamer, less_printing=False):
def load_channel_points_context(self, streamer):
json_data = copy.deepcopy(GQLOperations.ChannelPointsContext)
json_data["variables"] = {"channelLogin": streamer.username}
@@ -190,11 +194,7 @@ class Twitch:
streamer.channel_points = community_points["balance"]
if community_points["availableClaim"] is not None:
self.claim_bonus(
streamer,
community_points["availableClaim"]["id"],
less_printing=less_printing,
)
self.claim_bonus(streamer, community_points["availableClaim"]["id"])
def make_predictions(self, event):
decision = event.bet.calculate(event.streamer.channel_points)
@@ -232,7 +232,8 @@ class Twitch:
if watch_streak is True:
for index in streamers_index:
if (
streamers[index].stream.watch_streak_missing is True
streamers[index].settings.watch_streak is True
and streamers[index].stream.watch_streak_missing is True
and (
streamers[index].offline_at == 0
or ((time.time() - streamers[index].offline_at) // 60) > 30

2
TwitchChannelPointsMiner/classes/TwitchBrowser.py
View File

@@ -13,7 +13,7 @@ from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions
from selenium.webdriver.support.ui import WebDriverWait
from TwitchChannelPointsMiner.classes.EventPrediction import EventPrediction
from TwitchChannelPointsMiner.classes.entities.EventPrediction import EventPrediction
from TwitchChannelPointsMiner.constants.browser import Javascript, Selectors
from TwitchChannelPointsMiner.constants.twitch import URL
from TwitchChannelPointsMiner.utils import _millify, bet_condition, get_user_agent

4
TwitchChannelPointsMiner/classes/TwitchWebSocket.py
View File

@@ -39,11 +39,9 @@ class TwitchWebSocket(WebSocketApp):
self.pending_topics = []
self.twitch = parent_pool.twitch
self.twitch_browser = parent_pool.twitch_browser
self.browser = parent_pool.browser
self.streamers = parent_pool.streamers
self.bet_settings = parent_pool.bet_settings
self.events_predictions = parent_pool.events_predictions
self.less_printing = parent_pool.less_printing
self.last_message_timestamp = None
self.last_message_type_channel = None

30
TwitchChannelPointsMiner/classes/WebSocketsPool.py
View File

@@ -6,10 +6,10 @@ import time
from dateutil import parser
from TwitchChannelPointsMiner.classes.EventPrediction import EventPrediction
from TwitchChannelPointsMiner.classes.entities.EventPrediction import EventPrediction
from TwitchChannelPointsMiner.classes.entities.Message import Message
from TwitchChannelPointsMiner.classes.entities.Raid import Raid
from TwitchChannelPointsMiner.classes.Exceptions import TimeBasedDropNotFound
from TwitchChannelPointsMiner.classes.Message import Message
from TwitchChannelPointsMiner.classes.Raid import Raid
from TwitchChannelPointsMiner.classes.TwitchWebSocket import TwitchWebSocket
from TwitchChannelPointsMiner.constants.twitch import WEBSOCKET
from TwitchChannelPointsMiner.utils import (
@@ -23,23 +23,12 @@ logger = logging.getLogger(__name__)
class WebSocketsPool:
def __init__(
self,
twitch,
twitch_browser,
streamers,
bet_settings,
events_predictions,
less_printing: bool = False,
):
def __init__(self, twitch, browser, streamers, events_predictions):
self.ws = None
self.twitch = twitch
self.twitch_browser = twitch_browser
self.browser = browser
self.streamers = streamers
self.events_predictions = events_predictions
self.bet_settings = bet_settings
self.less_printing = less_printing
"""
API Limits
@@ -155,7 +144,6 @@ class WebSocketsPool:
ws.twitch.claim_bonus(
ws.streamers[streamer_index],
message.data["claim"]["id"],
less_printing=ws.less_printing,
)
elif message.topic == "video-playback-by-id":
@@ -206,14 +194,12 @@ class WebSocketsPool:
prediction_window_seconds,
event_status,
event_dict["outcomes"],
bet_settings=ws.bet_settings,
less_printing=ws.less_printing,
)
if (
ws.streamers[streamer_index].is_online
and event.closing_bet_after(current_tmsp) > 0
and bet_condition(
ws.twitch_browser,
ws.browser,
event,
logger,
)
@@ -223,7 +209,7 @@ class WebSocketsPool:
(
start_bet_status,
execution_time,
) = ws.twitch_browser.start_bet(
) = ws.browser.start_bet(
ws.events_predictions[event_id]
)
if start_bet_status is True:
@@ -235,7 +221,7 @@ class WebSocketsPool:
place_bet_thread = threading.Timer(
start_after,
ws.twitch_browser.place_bet,
ws.browser.place_bet,
(ws.events_predictions[event_id],),
)
place_bet_thread.daemon = True

24
TwitchChannelPointsMiner/classes/Bet.py → TwitchChannelPointsMiner/classes/entities/Bet.py
View File

@@ -2,7 +2,9 @@ import copy
import logging
from enum import Enum, auto
from TwitchChannelPointsMiner.utils import _millify, float_round
from millify import millify
from TwitchChannelPointsMiner.utils import float_round
logger = logging.getLogger(__name__)
@@ -17,18 +19,24 @@ class Strategy(Enum):
class BetSettings:
def __init__(
self,
strategy: Strategy = Strategy.SMART,
percentage: int = 5,
percentage_gap: int = 2,
max_points: int = 50000,
strategy: Strategy = None,
percentage: int = None,
percentage_gap: int = None,
max_points: int = None,
):
self.strategy = strategy
self.percentage = percentage
self.percentage_gap = percentage_gap
self.max_points = max_points
def default(self):
self.strategy = self.strategy if not None else Strategy.SMART
self.percentage = self.percentage if not None else 5
self.percentage_gap = self.percentage_gap if not None else 2
self.max_points = self.max_points if not None else 50000
def __repr__(self):
return f"BetSettings(Strategy={self.strategy}, Percentage={self.percentage}, PercentageGap={self.percentage_gap}, MaxPoints={self.max_points}"
return f"BetSettings(Strategy={self.strategy}, Percentage={self.percentage}, PercentageGap={self.percentage_gap}, MaxPoints={self.max_points})"
class Bet:
@@ -72,11 +80,11 @@ class Bet:
self.__clear_outcomes()
def __repr__(self):
return f"Bet(TotalUsers={_millify(self.total_users)}, TotalPoints={_millify(self.total_points)}), Decision={self.decision})\n\t\tOutcome0({self.get_outcome(0)})\n\t\tOutcome1({self.get_outcome(1)})"
return f"Bet(TotalUsers={millify(self.total_users)}, TotalPoints={millify(self.total_points)}), Decision={self.decision})\n\t\tOutcome0({self.get_outcome(0)})\n\t\tOutcome1({self.get_outcome(1)})"
def get_outcome(self, index):
outcome = self.outcomes[index]
return f"{outcome['title']} ({outcome['color']}), Points: {_millify(outcome['total_points'])}, Users: {_millify(outcome['total_users'])} ({outcome['percentage_users']}%), Odds: {outcome['odds']} ({outcome['odds_percentage']}%)"
return f"{outcome['title']} ({outcome['color']}), Points: {millify(outcome['total_points'])}, Users: {millify(outcome['total_users'])} ({outcome['percentage_users']}%), Odds: {outcome['odds']} ({outcome['odds_percentage']}%)"
def __clear_outcomes(self):
for index in range(0, len(self.outcomes)):

25
TwitchChannelPointsMiner/classes/EventPrediction.py → TwitchChannelPointsMiner/classes/entities/EventPrediction.py
View File

@@ -1,5 +1,6 @@
from TwitchChannelPointsMiner.classes.Bet import Bet, BetSettings
from TwitchChannelPointsMiner.classes.Streamer import Streamer
from TwitchChannelPointsMiner.classes.entities.Bet import Bet
from TwitchChannelPointsMiner.classes.entities.Streamer import Streamer
from TwitchChannelPointsMiner.classes.Settings import Settings
from TwitchChannelPointsMiner.utils import float_round
@@ -13,8 +14,6 @@ class EventPrediction:
prediction_window_seconds,
status,
outcomes,
bet_settings: BetSettings,
less_printing: bool = False,
):
self.streamer = streamer
@@ -28,22 +27,16 @@ class EventPrediction:
self.box_fillable = False
self.bet_confirmed = False
self.bet_placed = False
self.bet = Bet(outcomes, bet_settings)
self.less_printing = less_printing
self.bet = Bet(outcomes, streamer.settings.bet)
def __repr__(self):
return (
f"EventPrediction: {self.title}"
if self.less_printing is True
else f"EventPrediction(event_id={self.event_id}, title={self.title})"
)
return f"EventPrediction(event_id={self.event_id}, title={self.title})"
def __str__(self):
return (
f"EventPrediction: {self.title}"
if self.less_printing is True
else f"EventPrediction(event_id={self.event_id}, title={self.title})"
if Settings.logger.less
else self.__repr__()
)
def elapsed(self, timestamp):
@@ -54,7 +47,3 @@ class EventPrediction:
def print_recap(self) -> str:
return f"{self}\n\t\t{self.streamer}\n\t\t{self.bet}\n\t\tResult: {self.final_result}"
def set_less_printing(self, value):
self.less_printing = value
self.streamer.less_printing = value

0
TwitchChannelPointsMiner/classes/Message.py → TwitchChannelPointsMiner/classes/entities/Message.py
View File

0
TwitchChannelPointsMiner/classes/PubsubTopic.py → TwitchChannelPointsMiner/classes/entities/PubsubTopic.py
View File

0
TwitchChannelPointsMiner/classes/Raid.py → TwitchChannelPointsMiner/classes/entities/Raid.py
View File

17
TwitchChannelPointsMiner/classes/Stream.py → TwitchChannelPointsMiner/classes/entities/Stream.py
View File

@@ -3,13 +3,14 @@ import logging
import time
from base64 import b64encode
from TwitchChannelPointsMiner.classes.Settings import Settings
from TwitchChannelPointsMiner.constants.twitch import DROP_ID
logger = logging.getLogger(__name__)
class Stream:
def __init__(self, less_printing: bool = False):
def __init__(self):
self.broadcast_id = None
self.title = None
@@ -24,8 +25,6 @@ class Stream:
self.init_watch_streak()
self.less_printing = less_printing
def encode_payload(self) -> dict:
json_event = json.dumps(self.payload, separators=(",", ":"))
return {"data": (b64encode(json_event.encode("utf-8"))).decode("utf-8")}
@@ -45,18 +44,10 @@ class Stream:
logger.debug(f"Update: {self}")
def __repr__(self):
return (
f"{self.title}"
if self.less_printing is True
else f"Stream(title={self.title}, game={self.__str_game()}, tags={self.__str_tags()})"
)
return f"Stream(title={self.title}, game={self.__str_game()}, tags={self.__str_tags()})"
def __str__(self):
return (
f"{self.title}"
if self.less_printing is True
else f"Stream(title={self.title}, game={self.__str_game()}, tags={self.__str_tags()})"
)
return f"{self.title}" if Settings.logger.less else self.__repr__()
def __str_tags(self):
return (

56
TwitchChannelPointsMiner/classes/Streamer.py → TwitchChannelPointsMiner/classes/entities/Streamer.py
View File

@@ -1,17 +1,46 @@
import logging
import time
from TwitchChannelPointsMiner.classes.Stream import Stream
from TwitchChannelPointsMiner.classes.entities.Bet import BetSettings
from TwitchChannelPointsMiner.classes.entities.Stream import Stream
from TwitchChannelPointsMiner.classes.Settings import Settings
from TwitchChannelPointsMiner.constants.twitch import URL
from TwitchChannelPointsMiner.utils import _millify
logger = logging.getLogger(__name__)
class Streamer:
def __init__(self, username, channel_id, less_printing: bool = False):
self.username = username
self.channel_id = channel_id
class StreamerSettings(object):
def __init__(
self,
make_predictions: bool = None,
follow_raid: bool = None,
claim_drops: bool = None,
watch_streak: bool = None,
bet: BetSettings = None,
):
self.make_predictions = make_predictions
self.follow_raid = follow_raid
self.claim_drops = claim_drops
self.watch_streak = watch_streak
self.bet = bet
def default(self):
for name in ["make_predictions", "follow_raid", "claim_drops", "watch_streak"]:
if getattr(self, name) is None:
setattr(self, name, True)
if self.bet is None:
self.bet = BetSettings()
def __repr__(self):
return f"BetSettings(MakePredictions={self.make_predictions}, FollowRaid={self.follow_raid}, ClaimDrops={self.claim_drops}, WatchStreak={self.watch_streak}, Bet={self.bet})"
class Streamer(object):
def __init__(self, username, settings=None):
self.username = username.lower().strip()
self.channel_id = 0
self.settings = settings
self.is_online = False
self.stream_up = 0
self.online_at = 0
@@ -20,7 +49,7 @@ class Streamer:
self.minute_watched_requests = None
self.viewer_is_mod = False
self.stream = Stream(less_printing=less_printing)
self.stream = Stream()
self.raid = None
self.history = {}
@@ -28,20 +57,14 @@ class Streamer:
self.streamer_url = f"{URL}/{self.username}"
self.chat_url = f"{URL}/popout/{self.username}/chat?popout="
self.less_printing = less_printing
def __repr__(self):
return (
f"{self.username} ({_millify(self.channel_points)} points)"
if self.less_printing is True
else f"Streamer(username={self.username}, channel_id={self.channel_id}, channel_points={_millify(self.channel_points)})"
)
return f"Streamer(username={self.username}, channel_id={self.channel_id}, channel_points={_millify(self.channel_points)})"
def __str__(self):
return (
f"{self.username} ({_millify(self.channel_points)} points)"
if self.less_printing is True
else f"Streamer(username={self.username}, channel_id={self.channel_id}, channel_points={_millify(self.channel_points)})"
if Settings.logger.less
else self.__repr__()
)
def set_offline(self):
@@ -76,8 +99,5 @@ class Streamer:
if reason_code == "WATCH_STREAK":
self.stream.watch_streak_missing = False
def set_less_printing(self, value):
self.less_printing = value
def stream_up_elapsed(self):
return self.stream_up == 0 or ((time.time() - self.stream_up) > 120)

0
TwitchChannelPointsMiner/classes/Logger.py → TwitchChannelPointsMiner/logger.py
View File

29
TwitchChannelPointsMiner/utils.py
View File

@@ -119,3 +119,32 @@ def remove_emoji(string: str) -> str:
flags=re.UNICODE,
)
return emoji_pattern.sub(r"", string)
def at_least_one_value_in_settings_is(array, attr_name, condition=True):
return [
itme for itme in array if getattr(itme.settings, attr_name) == condition
] != []
def copy_values_if_none(settings, defaults):
values = [
name
for name in dir(settings)
if name.startswith("__") is False and callable(getattr(settings, name)) is False
]
for value in values:
if getattr(settings, value) is None:
setattr(settings, value, getattr(defaults, value))
return settings
def set_default_settings(settings, defaults):
# If no settings was provided use the default settings ...
if settings is None:
settings = defaults
else:
# If settings was provided but maybe are only partial set
# Get the default values from Settings.streamer_settings
settings = copy_values_if_none(settings, defaults)
return settings

BIN
assets/Roboto-Bold.ttf
View File

Binary file not shown.

BIN
assets/main_screen.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 307 KiB

BIN
assets/prediction.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 30 KiB

BIN
assets/prediction_screen.png
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 24 KiB

51
example.py
View File

@@ -2,16 +2,13 @@
import logging
from TwitchChannelPointsMiner import TwitchChannelPointsMiner
from TwitchChannelPointsMiner.classes.Logger import LoggerSettings
from TwitchChannelPointsMiner.classes.Bet import Strategy, BetSettings
from TwitchChannelPointsMiner.logger import LoggerSettings
from TwitchChannelPointsMiner.classes.entities.Bet import Strategy, BetSettings
from TwitchChannelPointsMiner.classes.entities.Streamer import Streamer, StreamerSettings
from TwitchChannelPointsMiner.classes.TwitchBrowser import Browser, BrowserSettings
twitch_miner = TwitchChannelPointsMiner(
username="your-twitch-username",
make_predictions=True, # If you want to Bet / Make prediction | The browser will never start
follow_raid=True, # Follow raid to obtain more points
watch_streak=True, # If a streamer go online change the priotiry of streamers array and catch the watch screak. Issue #11
drops_events=True, # If you want to auto claim game drops from Twitch inventory Issue #21
claim_drops_startup=False, # If you want to auto claim all drops from Twitch inventory on startup
logger_settings=LoggerSettings(
save=True, # If you want to save logs in file (suggested)
@@ -25,15 +22,41 @@ twitch_miner = TwitchChannelPointsMiner(
show=False, # Show the browser during bet else headless mode
do_screenshot=False, # Do screenshot during the bet
),
bet_settings=BetSettings(
strategy=Strategy.SMART, # Choose you strategy!
percentage=5, # Place the x% of your channel points
percentage_gap=20, # Gap difference between outcomesA and outcomesB (for SMART stragegy)
max_points=50000, # If the x percentage of your channel points is gt bet_max_points set this value
streamer_settings=StreamerSettings(
make_predictions=True, # If you want to Bet / Make prediction
follow_raid=True, # Follow raid to obtain more points
claim_drops=True, # We can't filter rewards base on stream. Set to False for skip viewing counter increase and you will never obtain a drop reward from this script. Issue #21
watch_streak=True, # If a streamer go online change the priotiry of streamers array and catch the watch screak. Issue #11
bet=BetSettings(
strategy=Strategy.SMART, # Choose you strategy!
percentage=5, # Place the x% of your channel points
percentage_gap=20, # Gap difference between outcomesA and outcomesB (for SMART stragegy)
max_points=50000, # If the x percentage of your channel points is gt bet_max_points set this value
)
)
)
# You can customize the settings for each streamer. If not settings was provided the script will use the streamer_settings from TwitchChannelPointsMiner.
# If no streamer_settings provided in TwitchChannelPointsMiner the script will use default settings.
# The streamers array can be a String -> username or Streamer instance.
# The settings priority are: settings in mine function, settings in TwitchChannelPointsMiner instance, default settings.
# For example if in the mine function you don't provide any value for 'make_prediction' but you have set it on TwitchChannelPointsMiner instance the script will take the value from here.
# If you haven't set any value even in the instance the default one will be used
twitch_miner.mine(
["streamer1", "streamer2"], # Array of streamers (order = priority)
followers=False # Automatic download the list of your followers
)
[
Streamer("streamer-username01", settings=StreamerSettings(make_predictions=True , follow_raid=False , claim_drops=True , watch_streak=True , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , percentage_gap=20 , max_points=234 ) )),
Streamer("streamer-username02", settings=StreamerSettings(make_predictions=False , follow_raid=True , claim_drops=False , bet=BetSettings(strategy=Strategy.PERCENTAGE , percentage=5 , percentage_gap=20 , max_points=1234 ) )),
Streamer("streamer-username03", settings=StreamerSettings(make_predictions=True , follow_raid=False , watch_streak=True , bet=BetSettings(strategy=Strategy.SMART , percentage=5 , percentage_gap=30 , max_points=50000 ) )),
Streamer("streamer-username04", settings=StreamerSettings(make_predictions=False , follow_raid=True , watch_streak=True )),
Streamer("streamer-username05", settings=StreamerSettings(make_predictions=True , follow_raid=True , claim_drops=True , watch_streak=True , bet=BetSettings(strategy=Strategy.ODDS , percentage=7 , percentage_gap=20 , max_points=90 ) )),
Streamer("streamer-username06"),
Streamer("streamer-username07"),
Streamer("streamer-username08"),
"streamer-username09",
"streamer-username10",
"streamer-username11"
], # Array of streamers (order = priority)
followers=False # Automatic download the list of your followers (unable to set custom settings for you followers list)
)