-
-
Notifications
You must be signed in to change notification settings - Fork 146
/
twitch.py
1728 lines (1646 loc) · 77 KB
/
twitch.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
from __future__ import annotations
import re
import sys
import json
import asyncio
import logging
from time import time
from itertools import chain
from functools import partial
from collections import abc, deque, OrderedDict
from datetime import datetime, timedelta, timezone
from contextlib import suppress, asynccontextmanager
from typing import Any, Literal, Final, NoReturn, overload, cast, TYPE_CHECKING
if sys.platform == "win32":
from subprocess import CREATE_NO_WINDOW
import aiohttp
from yarl import URL
try:
from seleniumwire.request import Request
from selenium.common.exceptions import WebDriverException
from seleniumwire.undetected_chromedriver import Chrome, ChromeOptions
except ModuleNotFoundError:
# the dependencies weren't installed, but they're not used either, so skip them
pass
except ImportError as exc:
if "_brotli" in exc.msg:
raise ImportError(
"You need to install Visual C++ Redist (x86 and x64): "
"https://support.microsoft.com/en-gb/help/2977003/"
"the-latest-supported-visual-c-downloads"
) from exc
raise
from translate import _
from gui import GUIManager
from channel import Channel
from websocket import WebsocketPool
from inventory import DropsCampaign
from exceptions import (
MinerException,
CaptchaRequired,
ExitRequest,
LoginException,
ReloadRequest,
RequestInvalid,
)
from utils import (
CHARS_HEX_LOWER,
chunk,
timestamp,
create_nonce,
task_wrapper,
first_to_complete,
OrderedSet,
AwaitableValue,
ExponentialBackoff,
)
from constants import (
CALL,
BASE_URL,
COOKIES_PATH,
GQL_OPERATIONS,
MAX_CHANNELS,
WATCH_INTERVAL,
State,
ClientType,
WebsocketTopic,
)
if TYPE_CHECKING:
from utils import Game
from gui import LoginForm
from channel import Stream
from settings import Settings
from inventory import TimedDrop
from constants import JsonType, GQLOperation
logger = logging.getLogger("TwitchDrops")
gql_logger = logging.getLogger("TwitchDrops.gql")
class SkipExtraJsonDecoder(json.JSONDecoder):
def decode(self, s: str, *args):
# skip whitespace check
obj, end = self.raw_decode(s)
return obj
CLIENT_URL, CLIENT_ID, USER_AGENT = ClientType.MOBILE_WEB
SAFE_LOADS = lambda s: json.loads(s, cls=SkipExtraJsonDecoder)
class _AuthState:
def __init__(self, twitch: Twitch):
self._twitch: Twitch = twitch
self._lock = asyncio.Lock()
self._logged_in = asyncio.Event()
self.user_id: int
self.device_id: str
self.session_id: str
self.access_token: str
self.client_version: str
self.integrity_token: str
self.integrity_expires: datetime
@property
def integrity_expired(self) -> bool:
return (
not hasattr(self, "integrity_expires")
or datetime.now(timezone.utc) >= self.integrity_expires
)
def _hasattrs(self, *attrs: str) -> bool:
return all(hasattr(self, attr) for attr in attrs)
def _delattrs(self, *attrs: str) -> None:
for attr in attrs:
if hasattr(self, attr):
delattr(self, attr)
def clear(self) -> None:
self._delattrs(
"user_id",
"device_id",
"session_id",
"access_token",
"client_version",
"integrity_token",
"integrity_expires",
)
self._logged_in.clear()
@staticmethod
def interceptor(request: Request) -> None:
if (
request.method == "POST"
and request.url == "https://passport.twitch.tv/protected_login"
):
body = request.body.decode("utf-8")
data = json.loads(body)
data["client_id"] = CLIENT_ID
request.body = json.dumps(data).encode("utf-8")
del request.headers["Content-Length"]
request.headers["Content-Length"] = str(len(request.body))
async def _chrome_login(self) -> None:
gui_print = self._twitch.gui.print
login_form: LoginForm = self._twitch.gui.login
coro_unless_closed = self._twitch.gui.coro_unless_closed
# open the chrome browser on the Twitch's login page
# use a separate executor to void blocking the event loop
loop = asyncio.get_running_loop()
driver: Chrome | None = None
while True:
gui_print(_("login", "chrome", "startup"))
try:
version_main = None
for attempt in range(2):
options = ChromeOptions()
options.add_argument("--log-level=3")
options.add_argument("--disable-web-security")
options.add_argument("--allow-running-insecure-content")
options.add_argument("--lang=en")
options.add_argument("--disable-gpu")
options.set_capability("pageLoadStrategy", "eager")
try:
wire_options: dict[str, Any] = {"proxy": {}}
if self._twitch.settings.proxy:
wire_options["proxy"]["http"] = str(self._twitch.settings.proxy)
driver_coro = loop.run_in_executor(
None,
lambda: Chrome(
options=options,
no_sandbox=True,
suppress_welcome=True,
version_main=version_main,
seleniumwire_options=wire_options,
service_creationflags=CREATE_NO_WINDOW,
)
)
driver = await coro_unless_closed(driver_coro)
break
except WebDriverException as exc:
message = exc.msg
if (
message is not None
and (
match := re.search(
(
r'Chrome version ([\d]+)\n'
r'Current browser version is ((\d+)\.[\d.]+)'
),
message,
)
) is not None
):
if not attempt:
version_main = int(match.group(3))
continue
else:
raise MinerException(
"Your Chrome browser is out of date\n"
f"Required version: {match.group(1)}\n"
f"Current version: {match.group(2)}"
) from None
raise MinerException(
"An error occured while boostrapping the Chrome browser"
) from exc
assert driver is not None
driver.request_interceptor = self.interceptor
# driver.set_page_load_timeout(30)
# page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv")
# await coro_unless_closed(page_coro)
page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv/login")
await coro_unless_closed(page_coro)
# auto login
# if login_data.username and login_data.password:
# driver.find_element("id", "login-username").send_keys(login_data.username)
# driver.find_element("id", "password-input").send_keys(login_data.password)
# driver.find_element(
# "css selector", '[data-a-target="passport-login-button"]'
# ).click()
# token submit button css selectors
# Button: "screen="two_factor" target="submit_button"
# Input: <input type="text" autocomplete="one-time-code" data-a-target="tw-input"
# inputmode="numeric" pattern="[0-9]*" value="">
# wait for the user to navigate away from the URL, indicating successful login
# alternatively, they can press on the login button again
async def url_waiter(driver=driver):
while driver.current_url != "https://www.twitch.tv/?no-reload=true":
await asyncio.sleep(0.5)
gui_print(_("login", "chrome", "login_to_complete"))
await first_to_complete([
url_waiter(),
coro_unless_closed(login_form.wait_for_login_press()),
])
# cookies = [
# {
# "domain": ".twitch.tv",
# "expiry": 1700000000,
# "httpOnly": False,
# "name": "auth-token",
# "path": "/",
# "sameSite": "None",
# "secure": True,
# "value": "..."
# },
# ...,
# ]
cookies = driver.get_cookies()
for cookie in cookies:
if "twitch.tv" in cookie["domain"] and cookie["name"] == "auth-token":
self.access_token = cookie["value"]
break
else:
gui_print(_("login", "chrome", "no_token"))
except WebDriverException:
gui_print(_("login", "chrome", "closed_window"))
finally:
if driver is not None:
driver.quit()
driver = None
await coro_unless_closed(login_form.wait_for_login_press())
async def _oauth_login(self) -> str:
login_form: LoginForm = self._twitch.gui.login
headers = {
"Accept": "application/json",
"Accept-Encoding": "gzip",
"Accept-Language": "en-US",
"Cache-Control": "no-cache",
"Client-Id": CLIENT_ID,
"Host": "id.twitch.tv",
"Origin": str(CLIENT_URL),
"Pragma": "no-cache",
"Referer": str(CLIENT_URL),
"User-Agent": USER_AGENT,
"X-Device-Id": self.device_id,
}
payload = {
"client_id": CLIENT_ID,
"scopes": (
"channel_read chat:read user_blocks_edit "
"user_blocks_read user_follows_edit user_read"
),
}
while True:
try:
async with self._twitch.request(
"POST", "https://id.twitch.tv/oauth2/device", headers=headers, data=payload
) as response:
# {
# "device_code": "40 chars [A-Za-z0-9]",
# "expires_in": 1800,
# "interval": 5,
# "user_code": "8 chars [A-Z]",
# "verification_uri": "https://www.twitch.tv/activate"
# }
now = datetime.now(timezone.utc)
response_json: JsonType = await response.json()
device_code: str = response_json["device_code"]
user_code: str = response_json["user_code"]
interval: int = response_json["interval"]
expires_at = now + timedelta(seconds=response_json["expires_in"])
# Print the code to the user, open them the activate page so they can type it in
await login_form.ask_enter_code(user_code)
payload = {
"client_id": CLIENT_ID,
"device_code": device_code,
"grant_type": "urn:ietf:params:oauth:grant-type:device_code",
}
while True:
# sleep first, not like the user is gonna enter the code *that* fast
await asyncio.sleep(interval)
async with self._twitch.request(
"POST",
"https://id.twitch.tv/oauth2/token",
headers=headers,
data=payload,
invalidate_after=expires_at,
) as response:
# 200 means success, 400 means the user haven't entered the code yet
if response.status != 200:
continue
response_json = await response.json()
# {
# "access_token": "40 chars [A-Za-z0-9]",
# "refresh_token": "40 chars [A-Za-z0-9]",
# "scope": [...],
# "token_type": "bearer"
# }
self.access_token = cast(str, response_json["access_token"])
return self.access_token
except RequestInvalid:
# the device_code has expired, request a new code
continue
async def _login(self) -> str:
logger.info("Login flow started")
gui_print = self._twitch.gui.print
login_form: LoginForm = self._twitch.gui.login
token_kind: str = ''
use_chrome: bool = False
payload: JsonType = {
# username and password are added later
# "username": str,
# "password": str,
"client_id": CLIENT_ID, # client ID to-be associated with the access token
"undelete_user": False, # purpose unknown
"remember_me": True, # persist the session via the cookie
# "authy_token": str, # 2FA token
# "twitchguard_code": str, # email code
# "captcha": str, # self-fed captcha
# 'force_twitchguard': False, # force email code confirmation
}
while True:
login_data = await login_form.ask_login()
payload["username"] = login_data.username
payload["password"] = login_data.password
# reinstate the 2FA token, if present
payload.pop("authy_token", None)
payload.pop("twitchguard_code", None)
if login_data.token:
# if there's no token kind set yet, and the user has entered a token,
# we can immediately assume it's an authenticator token and not an email one
if not token_kind:
token_kind = "authy"
if token_kind == "authy":
payload["authy_token"] = login_data.token
elif token_kind == "email":
payload["twitchguard_code"] = login_data.token
# use fancy headers to mimic the twitch android app
headers = {
"Accept": "application/vnd.twitchtv.v3+json",
"Accept-Encoding": "gzip",
"Accept-Language": "en-US",
"Client-Id": CLIENT_ID,
"Content-Type": "application/json; charset=UTF-8",
"Host": "passport.twitch.tv",
"User-Agent": USER_AGENT,
"X-Device-Id": self.device_id,
# "X-Device-Id": ''.join(random.choices('0123456789abcdef', k=32)),
}
async with self._twitch.request(
"POST", "https://passport.twitch.tv/login", headers=headers, json=payload
) as response:
login_response: JsonType = await response.json(loads=SAFE_LOADS)
# Feed this back in to avoid running into CAPTCHA if possible
if "captcha_proof" in login_response:
payload["captcha"] = {"proof": login_response["captcha_proof"]}
# Error handling
if "error_code" in login_response:
error_code: int = login_response["error_code"]
logger.info(f"Login error code: {error_code}")
if error_code == 1000:
logger.info("1000: CAPTCHA is required")
use_chrome = True
break
elif error_code in (2004, 3001):
logger.info("3001: Login failed due to incorrect username or password")
gui_print(_("login", "incorrect_login_pass"))
if error_code == 2004:
# invalid username
login_form.clear(login=True)
login_form.clear(password=True)
continue
elif error_code in (
3012, # Invalid authy token
3023, # Invalid email code
):
logger.info("3012/23: Login failed due to incorrect 2FA code")
if error_code == 3023:
token_kind = "email"
gui_print(_("login", "incorrect_email_code"))
else:
token_kind = "authy"
gui_print(_("login", "incorrect_twofa_code"))
login_form.clear(token=True)
continue
elif error_code in (
3011, # Authy token needed
3022, # Email code needed
):
# 2FA handling
logger.info("3011/22: 2FA token required")
# user didn't provide a token, so ask them for it
if error_code == 3022:
token_kind = "email"
gui_print(_("login", "email_code_required"))
else:
token_kind = "authy"
gui_print(_("login", "twofa_code_required"))
continue
elif error_code >= 5000:
# Special errors, usually from Twitch telling the user to "go away"
# We print the code out to inform the user, and just use chrome flow instead
# {
# "error_code":5023,
# "error":"Please update your app to continue",
# "error_description":"client is not supported for this feature"
# }
# {
# "error_code":5027,
# "error":"Please update your app to continue",
# "error_description":"client blocked from this operation"
# }
gui_print(_("login", "error_code").format(error_code=error_code))
logger.info(str(login_response))
use_chrome = True
break
else:
ext_msg = str(login_response)
logger.info(ext_msg)
raise LoginException(ext_msg)
# Success handling
if "access_token" in login_response:
self.access_token = cast(str, login_response["access_token"])
logger.info("Access token granted")
login_form.clear()
break
if use_chrome:
# await self._chrome_login()
raise CaptchaRequired()
if hasattr(self, "access_token"):
return self.access_token
raise MinerException("Login flow finished without setting the access token")
def headers(
self, *, user_agent: str = '', gql: bool = False, integrity: bool = False
) -> JsonType:
headers = {
"Accept": "*/*",
"Accept-Encoding": "gzip",
"Accept-Language": "en-US",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
"Client-Id": CLIENT_ID,
}
if user_agent:
headers["User-Agent"] = user_agent
if hasattr(self, "session_id"):
headers["Client-Session-Id"] = self.session_id
# if hasattr(self, "client_version"):
# headers["Client-Version"] = self.client_version
if hasattr(self, "device_id"):
headers["X-Device-Id"] = self.device_id
if gql:
headers["Origin"] = str(BASE_URL)
headers["Referer"] = str(BASE_URL)
headers["Authorization"] = f"OAuth {self.access_token}"
if integrity:
headers["Client-Integrity"] = self.integrity_token
return headers
async def validate(self):
async with self._lock:
await self._validate()
async def _validate(self):
if not hasattr(self, "session_id"):
self.session_id = create_nonce(CHARS_HEX_LOWER, 16)
if not self._hasattrs("device_id", "access_token", "user_id"):
session = await self._twitch.get_session()
jar = cast(aiohttp.CookieJar, session.cookie_jar)
if not self._hasattrs("device_id"):
async with self._twitch.request(
"GET", CLIENT_URL, headers=self.headers()
) as response:
page_html = await response.text("utf8")
assert page_html is not None
# match = re.search(r'twilightBuildID="([-a-z0-9]+)"', page_html)
# if match is None:
# raise MinerException("Unable to extract client_version")
# self.client_version = match.group(1)
# doing the request ends up setting the "unique_id" value in the cookie
cookie = jar.filter_cookies(CLIENT_URL)
self.device_id = cookie["unique_id"].value
if not self._hasattrs("access_token", "user_id"):
# looks like we're missing something
login_form: LoginForm = self._twitch.gui.login
logger.info("Checking login")
login_form.update(_("gui", "login", "logging_in"), None)
for attempt in range(2):
cookie = jar.filter_cookies(CLIENT_URL)
if "auth-token" not in cookie:
self.access_token = await self._oauth_login()
cookie["auth-token"] = self.access_token
elif not hasattr(self, "access_token"):
logger.info("Restoring session from cookie")
self.access_token = cookie["auth-token"].value
# validate the auth token, by obtaining user_id
async with self._twitch.request(
"GET",
"https://id.twitch.tv/oauth2/validate",
headers={"Authorization": f"OAuth {self.access_token}"}
) as response:
status = response.status
if status == 401:
# the access token we have is invalid - clear the cookie and reauth
logger.info("Restored session is invalid")
assert CLIENT_URL.host is not None
jar.clear_domain(CLIENT_URL.host)
continue
elif status == 200:
validate_response = await response.json()
break
else:
raise RuntimeError("Login verification failure")
if validate_response["client_id"] != CLIENT_ID:
raise MinerException("You're using an old cookie file, please generate a new one.")
self.user_id = int(validate_response["user_id"])
cookie["persistent"] = str(self.user_id)
logger.info(f"Login successful, user ID: {self.user_id}")
login_form.update(_("gui", "login", "logged_in"), self.user_id)
# update our cookie and save it
jar.update_cookies(cookie, CLIENT_URL)
jar.save(COOKIES_PATH)
# if not self._hasattrs("integrity_token") or self.integrity_expired:
# async with self._twitch.request(
# "POST",
# "https://gql.twitch.tv/integrity",
# headers=self.gql_headers(integrity=False)
# ) as response:
# self._last_request = datetime.now(timezone.utc)
# response_json: JsonType = await response.json()
# self.integrity_token = cast(str, response_json["token"])
# now = datetime.now(timezone.utc)
# expiration = datetime.fromtimestamp(response_json["expiration"] / 1000, timezone.utc)
# self.integrity_expires = ((expiration - now) * 0.9) + now
# # verify the integrity token's contents for the "is_bad_bot" flag
# stripped_token: str = self.integrity_token.split('.')[2] + "=="
# messy_json: str = urlsafe_b64decode(stripped_token.encode()).decode(errors="ignore")
# match = re.search(r'(.+)(?<="}).+$', messy_json)
# if match is None:
# raise MinerException("Unable to parse the integrity token")
# decoded_header: JsonType = json.loads(match.group(1))
# if decoded_header.get("is_bad_bot", "false") != "false":
# self._twitch.print(
# "Twitch has detected this miner as a \"Bad Bot\". "
# "You're proceeding at your own risk!"
# )
# await asyncio.sleep(8)
self._logged_in.set()
def invalidate(self, *, auth: bool = False, integrity: bool = False):
if auth:
self._delattrs("access_token")
if integrity:
self._delattrs("client_version")
self.integrity_expires = datetime.now(timezone.utc)
class Twitch:
def __init__(self, settings: Settings):
self.settings: Settings = settings
# State management
self._state: State = State.IDLE
self._state_change = asyncio.Event()
self.wanted_games: dict[Game, int] = {}
self.inventory: list[DropsCampaign] = []
self._drops: dict[str, TimedDrop] = {}
self._mnt_triggers: deque[datetime] = deque()
# Session and auth
self._session: aiohttp.ClientSession | None = None
self._auth_state: _AuthState = _AuthState(self)
# GUI
self.gui = GUIManager(self)
# Storing and watching channels
self.channels: OrderedDict[int, Channel] = OrderedDict()
self.watching_channel: AwaitableValue[Channel] = AwaitableValue()
self._watching_task: asyncio.Task[None] | None = None
self._watching_restart = asyncio.Event()
self._drop_update: asyncio.Future[bool] | None = None
# Websocket
self.websocket = WebsocketPool(self)
# Maintenance task
self._mnt_task: asyncio.Task[None] | None = None
async def get_session(self) -> aiohttp.ClientSession:
if (session := self._session) is not None:
if session.closed:
raise RuntimeError("Session is closed")
return session
# load in cookies
cookie_jar = aiohttp.CookieJar()
try:
if COOKIES_PATH.exists():
cookie_jar.load(COOKIES_PATH)
except Exception:
# if loading in the cookies file ends up in an error, just ignore it
# clear the jar, just in case
cookie_jar.clear()
# create timeouts
# connection quality mulitiplier determines the magnitude of timeouts
connection_quality = self.settings.connection_quality
if connection_quality < 1:
connection_quality = self.settings.connection_quality = 1
elif connection_quality > 6:
connection_quality = self.settings.connection_quality = 6
timeout = aiohttp.ClientTimeout(
sock_connect=5*connection_quality,
total=10*connection_quality,
)
# create session, limited to 50 connections at maximum
connector = aiohttp.TCPConnector(limit=50)
self._session = aiohttp.ClientSession(
timeout=timeout,
connector=connector,
cookie_jar=cookie_jar,
headers={"User-Agent": USER_AGENT},
)
return self._session
async def shutdown(self) -> None:
start_time = time()
self.stop_watching()
if self._watching_task is not None:
self._watching_task.cancel()
self._watching_task = None
if self._mnt_task is not None:
self._mnt_task.cancel()
self._mnt_task = None
# stop websocket, close session and save cookies
await self.websocket.stop(clear_topics=True)
if self._session is not None:
cookie_jar = cast(aiohttp.CookieJar, self._session.cookie_jar)
cookie_jar.save(COOKIES_PATH)
await self._session.close()
self._session = None
self._drop_update = None
self._drops.clear()
self.channels.clear()
self.inventory.clear()
self._auth_state.clear()
self.wanted_games.clear()
self._mnt_triggers.clear()
# wait at least half a second + whatever it takes to complete the closing
# this allows aiohttp to safely close the session
await asyncio.sleep(start_time + 0.5 - time())
def wait_until_login(self) -> abc.Coroutine[Any, Any, Literal[True]]:
return self._auth_state._logged_in.wait()
def change_state(self, state: State) -> None:
if self._state is not State.EXIT:
# prevent state changing once we switch to exit state
self._state = state
self._state_change.set()
def state_change(self, state: State) -> abc.Callable[[], None]:
# this is identical to change_state, but defers the call
# perfect for GUI usage
return partial(self.change_state, state)
def close(self):
"""
Called when the application is requested to close by the user,
usually by the console or application window being closed.
"""
self.change_state(State.EXIT)
def prevent_close(self):
"""
Called when the application window has to be prevented from closing, even after the user
closes it with X. Usually used solely to display tracebacks from the closing sequence.
"""
self.gui.prevent_close()
def print(self, message: str):
"""
Can be used to print messages within the GUI.
"""
self.gui.print(message)
def save(self, *, force: bool = False) -> None:
"""
Saves the application state.
"""
self.gui.save(force=force)
self.settings.save(force=force)
def get_priority(self, channel: Channel) -> int:
"""
Return a priority number for a given channel.
Higher number, higher priority.
Priority 0 is given to channels streaming a game not on the priority list.
Priority -1 is given to OFFLINE channels, or channels streaming no particular games.
"""
if (game := channel.game) is None:
# None when OFFLINE or no game set
return -1
elif game not in self.wanted_games:
return 0
return self.wanted_games[game]
@staticmethod
def _viewers_key(channel: Channel) -> int:
if (viewers := channel.viewers) is not None:
return viewers
return -1
async def run(self):
while True:
try:
await self._run()
break
except ReloadRequest:
await self.shutdown()
except ExitRequest:
break
except aiohttp.ContentTypeError as exc:
raise MinerException(_("login", "unexpected_content")) from exc
async def _run(self):
"""
Main method that runs the whole client.
Here, we manage several things, specifically:
• Fetching the drops inventory to make sure that everything we can claim, is claimed
• Selecting a stream to watch, and watching it
• Changing the stream that's being watched if necessary
"""
self.gui.start()
auth_state = await self.get_auth()
await self.websocket.start()
# NOTE: watch task is explicitly restarted on each new run
if self._watching_task is not None:
self._watching_task.cancel()
self._watching_task = asyncio.create_task(self._watch_loop())
# Add default topics
self.websocket.add_topics([
WebsocketTopic("User", "Drops", auth_state.user_id, self.process_drops),
WebsocketTopic("User", "CommunityPoints", auth_state.user_id, self.process_points),
WebsocketTopic(
"User", "Notifications", auth_state.user_id, self.process_notifications
),
])
full_cleanup: bool = False
channels: Final[OrderedDict[int, Channel]] = self.channels
self.change_state(State.INVENTORY_FETCH)
while True:
if self._state is State.IDLE:
self.gui.status.update(_("gui", "status", "idle"))
self.stop_watching()
# clear the flag and wait until it's set again
self._state_change.clear()
elif self._state is State.INVENTORY_FETCH:
# ensure the websocket is running
await self.websocket.start()
await self.fetch_inventory()
self.gui.set_games(set(campaign.game for campaign in self.inventory))
# Save state on every inventory fetch
self.save()
self.change_state(State.GAMES_UPDATE)
elif self._state is State.GAMES_UPDATE:
# claim drops from expired and active campaigns
for campaign in self.inventory:
if not campaign.upcoming:
for drop in campaign.drops:
if drop.can_claim:
await drop.claim()
# figure out which games we want
self.wanted_games.clear()
priorities = self.gui.settings.priorities()
exclude = self.settings.exclude
priority = self.settings.priority
priority_only = self.settings.priority_only
next_hour = datetime.now(timezone.utc) + timedelta(hours=1)
for campaign in self.inventory:
game = campaign.game
if (
game not in self.wanted_games # isn't already there
and game.name not in exclude # and isn't excluded
# and isn't excluded by priority_only
and (not priority_only or game.name in priority)
# and can be progressed within the next hour
and campaign.can_earn_within(next_hour)
):
# non-excluded games with no priority are placed last, below priority ones
self.wanted_games[game] = priorities.get(game.name, 0)
full_cleanup = True
self.restart_watching()
self.change_state(State.CHANNELS_CLEANUP)
elif self._state is State.CHANNELS_CLEANUP:
self.gui.status.update(_("gui", "status", "cleanup"))
if not self.wanted_games or full_cleanup:
# no games selected or we're doing full cleanup: remove everything
to_remove_channels: list[Channel] = list(channels.values())
else:
# remove all channels that:
to_remove_channels = [
channel
for channel in channels.values()
if (
not channel.acl_based # aren't ACL-based
and (
channel.offline # and are offline
# or online but aren't streaming the game we want anymore
or (channel.game is None or channel.game not in self.wanted_games)
)
)
]
full_cleanup = False
if to_remove_channels:
to_remove_topics: list[str] = []
for channel in to_remove_channels:
to_remove_topics.append(
WebsocketTopic.as_str("Channel", "StreamState", channel.id)
)
to_remove_topics.append(
WebsocketTopic.as_str("Channel", "StreamUpdate", channel.id)
)
self.websocket.remove_topics(to_remove_topics)
for channel in to_remove_channels:
del channels[channel.id]
channel.remove()
del to_remove_channels, to_remove_topics
if self.wanted_games:
self.change_state(State.CHANNELS_FETCH)
else:
# with no games available, we switch to IDLE after cleanup
self.print(_("status", "no_campaign"))
self.change_state(State.IDLE)
elif self._state is State.CHANNELS_FETCH:
self.gui.status.update(_("gui", "status", "gathering"))
# start with all current channels, clear the memory and GUI
new_channels: OrderedSet[Channel] = OrderedSet(channels.values())
channels.clear()
self.gui.channels.clear()
# gather and add ACL channels from campaigns
# NOTE: we consider only campaigns that can be progressed
# NOTE: we use another set so that we can set them online separately
no_acl: set[Game] = set()
acl_channels: OrderedSet[Channel] = OrderedSet()
next_hour = datetime.now(timezone.utc) + timedelta(hours=1)
for campaign in self.inventory:
if (
campaign.game in self.wanted_games
and campaign.can_earn_within(next_hour)
):
if campaign.allowed_channels:
acl_channels.update(campaign.allowed_channels)
else:
no_acl.add(campaign.game)
# remove all ACL channels that already exist from the other set
acl_channels.difference_update(new_channels)
# use the other set to set them online if possible
if acl_channels:
await asyncio.gather(
*(channel.update_stream(trigger_events=False) for channel in acl_channels)
)
# finally, add them as new channels
new_channels.update(acl_channels)
for game in no_acl:
# for every campaign without an ACL, for it's game,
# add a list of live channels with drops enabled
new_channels.update(await self.get_live_streams(game))
# sort them descending by viewers, by priority and by game priority
# NOTE: We can drop OrderedSet now because there's no more channels being added
ordered_channels: list[Channel] = sorted(
new_channels, key=self._viewers_key, reverse=True
)
ordered_channels.sort(key=lambda ch: ch.acl_based, reverse=True)
ordered_channels.sort(key=self.get_priority, reverse=True)
# ensure that we won't end up with more channels than we can handle
# NOTE: we trim from the end because that's where the non-priority,
# offline (or online but low viewers) channels end up
to_remove_channels = ordered_channels[MAX_CHANNELS:]
ordered_channels = ordered_channels[:MAX_CHANNELS]
if to_remove_channels:
# tracked channels and gui were cleared earlier, so no need to do it here
# just make sure to unsubscribe from their topics
to_remove_topics = []
for channel in to_remove_channels:
to_remove_topics.append(
WebsocketTopic.as_str("Channel", "StreamState", channel.id)
)
to_remove_topics.append(
WebsocketTopic.as_str("Channel", "StreamUpdate", channel.id)
)
self.websocket.remove_topics(to_remove_topics)
del to_remove_channels, to_remove_topics
# set our new channel list
for channel in ordered_channels:
channels[channel.id] = channel
channel.display(add=True)
# subscribe to these channel's state updates
to_add_topics: list[WebsocketTopic] = []
for channel_id in channels:
to_add_topics.append(
WebsocketTopic(
"Channel", "StreamState", channel_id, self.process_stream_state
)
)
to_add_topics.append(
WebsocketTopic(
"Channel", "StreamUpdate", channel_id, self.process_stream_update
)
)
self.websocket.add_topics(to_add_topics)
# relink watching channel after cleanup,
# or stop watching it if it no longer qualifies
# NOTE: this replaces 'self.watching_channel's internal value with the new object
watching_channel = self.watching_channel.get_with_default(None)
if watching_channel is not None:
new_watching: Channel | None = channels.get(watching_channel.id)
if new_watching is not None and self.can_watch(new_watching):
self.watch(new_watching, update_status=False)
else:
# we've removed a channel we were watching
self.stop_watching()
del new_watching
# pre-display the active drop with a substracted minute
for channel in channels.values():
# check if there's any channels we can watch first
if self.can_watch(channel):
if (active_drop := self.get_active_drop(channel)) is not None:
active_drop.display(countdown=False, subone=True)
del active_drop
break
self.change_state(State.CHANNEL_SWITCH)
del (
no_acl,
acl_channels,
new_channels,
to_add_topics,
ordered_channels,
watching_channel,
)
elif self._state is State.CHANNEL_SWITCH:
self.gui.status.update(_("gui", "status", "switching"))
# Change into the selected channel, stay in the watching channel,
# or select a new channel that meets the required conditions
new_watching = None
selected_channel = self.gui.channels.get_selection()
if selected_channel is not None and self.can_watch(selected_channel):
# selected channel is checked first, and set as long as we can watch it
new_watching = selected_channel
else:
# other channels additionally need to have a good reason
# for a switch (including the watching one)