extras
/
epic-awesome-gamer
mirror of https://github.com/QIN2DIM/epic-awesome-gamer
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Feat(challenge): merge YOLOv8 motion-flow (#204) 

* Feat(challenge): merge YOLOv8 motion-flow

* Update claim.py

* Update epic_games.py
This commit is contained in:
QIN2DIM 2023-08-30 18:48:07 +08:00 committed by GitHub
parent 09719138c0
commit 338eb55eec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 78 additions and 285 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -134,7 +134,7 @@ tests/
.github/workflows/sync_repo.yaml
docker-compose.yaml
archivist/
src/config.json
src/*.json
user_data_dir/
logs/
database/

2
requirements.txt
View File

@ -4,7 +4,7 @@ beautifulsoup4>=4.10.0
lxml>=4.9.3
# [extra_requires.solver]
hcaptcha-challenger[playwright]>=0.6.3-post4
hcaptcha-challenger[playwright]==0.7.0
# [extra_requires.notify]
apprise==1.1.0

14
src/claim.py
View File

@ -12,11 +12,12 @@ from playwright.sync_api import BrowserContext
from services.agents.epic_games import EpicPlayer, EpicGames
from services.agents.epic_games import get_promotions, get_order_history
solver.install(upgrade=True)
solver.install(upgrade=True, flush_yolo=True)
player = EpicPlayer.from_account()
epic = EpicGames.from_player(player)
promotions = []
ctx_cookies_is_available = None
@ -47,10 +48,13 @@ def prelude():
def claim_epic_games(context: BrowserContext):
global promotions
page = context.pages[0]
epic = EpicGames.from_player(player, page=page)
# Authorize
if not ctx_cookies_is_available:
logger.info("claim_epic_games", action="Try to flush cookie")
if epic.authorize(context):
if epic.authorize(page):
epic.flush_token(context)
else:
logger.error(
@ -69,7 +73,7 @@ def claim_epic_games(context: BrowserContext):
return
# Execute
epic.claim_weekly_games(context, promotions)
epic.claim_weekly_games(page, promotions)
@logger.catch
@ -78,7 +82,7 @@ def run():
# Cookie is unavailable or need to process promotions
agent = player.build_agent()
agent.execute(sequence=[claim_epic_games], headless=True)
agent.execute(sequence=[claim_epic_games], headless=False)
if __name__ == "__main__":

3
src/services/agents/__init__.py
View File

@ -15,6 +15,3 @@ XLangAI Agents https://github.com/xlang-ai/
| xbox | [Xbox Live Games with Gold](https://www.xbox.com/en-US/live/gold#gameswithgold) |
"""
from services.agents.epic_games import EpicGamesAgent
__all__ = ["EpicGamesAgent"]

147
src/services/agents/epic_games.py
View File

@ -9,16 +9,15 @@ import json
from contextlib import suppress
from dataclasses import dataclass, field
from json import JSONDecodeError
from pathlib import Path
from typing import List, Dict
import httpx
from hcaptcha_challenger.agents.exceptions import ChallengePassed
from hcaptcha_challenger.agents.skeleton import Status
from hcaptcha_challenger.agents.playwright.control import AgentT
from loguru import logger
from playwright.sync_api import BrowserContext, expect, TimeoutError
from playwright.sync_api import Page
from services.agents.hcaptcha_solver import is_fall_in_captcha, Radagon
from services.models import EpicPlayer
from utils import from_dict_to_model
@ -64,7 +63,7 @@ class EpicGames:
Agent control
"""
_radagon: Radagon = None
_solver: AgentT = None
"""
Module for anti-captcha
"""
@ -76,8 +75,13 @@ class EpicGames:
"""
@classmethod
def from_player(cls, player: EpicPlayer):
return cls(player=player, _radagon=Radagon.from_modelhub())
def from_player(
cls, player: EpicPlayer, *, page: Page, tmp_dir: Path | None = None, **solver_opt
):
"""尽可能早地实例化，用于部署 captcha 事件监听器"""
return cls(
player=player, _solver=AgentT.from_page(page=page, tmp_dir=tmp_dir, **solver_opt)
)
@property
def promotions(self) -> List[Game]:
@ -93,47 +97,39 @@ class EpicGames:
logger.info("login-with-epic", url=page.url)
page.fill("#email", self.player.email)
page.type("#password", self.player.password)
page.click("#sign-in")
for _ in range(8):
page.click("#sign-in")
try:
result = self._radagon.anti_hcaptcha(page, window="login", recur_url=URL_CLAIM)
if result in [self._radagon.status.CHALLENGE_BACKCALL]:
page.click("//a[@class='talon_close_button']")
page.wait_for_timeout(1000)
continue
fall_in_challenge = False
for _ in range(15):
if not fall_in_challenge:
with suppress(TimeoutError):
page.wait_for_url(URL_CLAIM, timeout=10000)
break
if not self._solver.qr:
return
fall_in_challenge = True
result = self._solver(window="login", recur_url=URL_CLAIM)
if result in [
self._solver.status.CHALLENGE_BACKCALL,
self._solver.status.CHALLENGE_RETRY,
]:
page.click("//a[@class='talon_close_button']")
page.wait_for_timeout(1000)
page.click("#sign-in", delay=200)
continue
if result == self._solver.status.CHALLENGE_SUCCESS:
page.wait_for_url(URL_CLAIM)
break
except ChallengePassed:
pass
page.wait_for_url(URL_CLAIM)
return self._solver.status.AUTH_SUCCESS
return self._radagon.status.AUTH_SUCCESS
def authorize(self, context: BrowserContext) -> bool | None:
page = context.pages[0]
beta = -1
while beta < 8:
beta += 1
def authorize(self, page: Page):
for _ in range(3):
result = self._login(page)
# Assert if you are fall in the hcaptcha challenge
if result not in [self._radagon.status.AUTH_SUCCESS]:
result = is_fall_in_captcha(page)
# Pass Challenge
if result == self._radagon.status.AUTH_SUCCESS:
return True
# Exciting moment :>
if result == self._radagon.status.AUTH_CHALLENGE:
resp = self._radagon.anti_hcaptcha(page, window="login")
if resp == Status.CHALLENGE_SUCCESS:
return True
if resp == Status.CHALLENGE_REFRESH:
beta -= 0.5
elif resp == Status.CHALLENGE_BACKCALL:
beta -= 0.75
elif resp == Status.CHALLENGE_CRASH:
beta += 0.5
if result not in [self._solver.status.CHALLENGE_SUCCESS]:
continue
return True
logger.critical("Failed to flush token", agent=self.__class__.__name__)
def flush_token(self, context: BrowserContext):
@ -145,16 +141,15 @@ class EpicGames:
)
context.storage_state(path=self.player.ctx_cookie_path)
self.player.ctx_cookies.reload(self.player.ctx_cookie_path)
logger.success("flush_token", path=self.player.ctx_cookie_path)
def claim_weekly_games(self, context: BrowserContext, promotions: List[Game]):
def claim_weekly_games(self, page: Page, promotions: List[Game]):
"""
:param context:
:param page:
:param promotions: 未在库的 promotions
:return:
"""
page = context.new_page()
# --> Add promotions to Cart
for promotion in promotions:
logger.info("claim_weekly_games", action="go to store", url=promotion.url)
@ -187,26 +182,37 @@ class EpicGames:
# --> Move to webPurchaseContainer iframe
logger.info("claim_weekly_games", action="move to webPurchaseContainer iframe")
wpc = page.frame_locator("//iframe[@class='']")
locator = wpc.locator("//div[@class='payment-order-confirm']")
payment_btn = wpc.locator("//div[@class='payment-order-confirm']")
with suppress(Exception):
expect(locator).to_be_attached()
expect(payment_btn).to_be_attached()
page.wait_for_timeout(2000)
payment_btn.click()
logger.info("claim_weekly_games", action="click payment button")
# <-- Insert challenge
for _ in range(8):
locator.click()
logger.info("claim_weekly_games", action="click payment button")
try:
result = self._radagon.anti_hcaptcha(
page, window="free", recur_url=URL_CART_SUCCESS
)
if result in [self._radagon.status.CHALLENGE_BACKCALL]:
page.click("//a[@class='talon_close_button']")
page.wait_for_timeout(1000)
continue
fall_in_challenge = False
for _ in range(15):
if not fall_in_challenge:
with suppress(TimeoutError):
page.wait_for_url(URL_CART_SUCCESS, timeout=10000)
break
logger.debug("claim_weekly_games", action="handle challenge")
fall_in_challenge = True
result = self._solver(window="free", recur_url=URL_CART_SUCCESS)
logger.debug("claim_weekly_games", action="challenge", result=result)
if result in [
self._solver.status.CHALLENGE_BACKCALL,
self._solver.status.CHALLENGE_RETRY,
]:
wpc.locator("//a[@class='talon_close_button']").click()
page.wait_for_timeout(1000)
payment_btn.click(delay=200)
continue
if result == self._solver.status.CHALLENGE_SUCCESS:
page.wait_for_url(URL_CART_SUCCESS)
break
except ChallengePassed:
pass
# --> Wait for success
page.wait_for_url(URL_CART_SUCCESS)
@ -284,22 +290,3 @@ def get_order_history(
logger.warning(err)
return completed_orders
@dataclass
class EpicGamesAgent:
player: EpicPlayer
methods: EpicGames
@classmethod
def build(cls):
player = EpicPlayer.from_account()
epic = EpicGames.from_player(player)
return cls(player=player, methods=epic)
def claim_weekly_games(self, context: BrowserContext):
orders = get_order_history(self.player.cookies)
namespaces = {order.namespace for order in orders}
promotions = [p for p in get_promotions() if p.namespace not in namespaces]
if promotions:
self.methods.claim_weekly_games(context, promotions)

195
src/services/agents/hcaptcha_solver.py
View File

@ -1,195 +0,0 @@
# -*- coding: utf-8 -*-
# Time : 2023/8/14 23:15
# Author : QIN2DIM
# Github : https://github.com/QIN2DIM
# Description:
from __future__ import annotations
from contextlib import suppress
from dataclasses import dataclass
from typing import Tuple
from hcaptcha_challenger.agents.exceptions import AuthMFA, AuthUnknownException, LoginException
from hcaptcha_challenger.agents.exceptions import ChallengePassed
from hcaptcha_challenger.agents.playwright import PlaywrightAgent
from hcaptcha_challenger.agents.skeleton import Status
from loguru import logger
from playwright.sync_api import Error as NinjaError
from playwright.sync_api import Page, FrameLocator
from playwright.sync_api import TimeoutError as NinjaTimeout
@dataclass
class Radagon(PlaywrightAgent):
"""人机对抗模组"""
def is_success(
self,
page: Page,
frame_challenge: FrameLocator = None,
window=None,
init=True,
hook_url=None,
*args,
**kwargs,
) -> Tuple[str, str]:
"""
判断挑战是否成功的复杂逻辑
:param hook_url:
:param frame_challenge:
:param init:
:param window:
:param page: 挑战者驱动上下文
:return:
"""
def is_continue_clickable():
""" "
False >> dom elements hidden
True >> it's clickable
"""
try:
prompts_obj = frame_challenge.locator("//div[@class='error-text']")
prompts_obj.first.wait_for(timeout=2000)
logger.debug("Checkout - status=再试一次")
return True
except NinjaTimeout:
task_image = frame_challenge.locator("//div[@class='task-image']")
task_image.first.wait_for(state="detached", timeout=3000)
return False
except NinjaError:
return False
def is_init_clickable():
with suppress(NinjaError):
return frame_challenge.locator("//div[@class='task-image']").first.is_visible()
# 首轮测试后判断短时间内页内是否存在可点击的拼图元素
# hcaptcha 最多两轮验证，一般情况下，账号信息有误仅会执行一轮，然后返回登录窗格提示密码错误
# 其次是被识别为自动化控制，这种情况也是仅执行一轮，回到登录窗格提示“返回数据错误”
if init and is_init_clickable():
return self.status.CHALLENGE_CONTINUE, "继续挑战"
if is_continue_clickable():
return self.status.CHALLENGE_CONTINUE, "继续挑战"
flag = page.url
if window == "free":
try:
page.locator(self.HOOK_PURCHASE).wait_for(state="detached")
return self.status.CHALLENGE_SUCCESS, "退火成功"
except NinjaTimeout:
return self.status.CHALLENGE_RETRY, "決策中斷"
if window == "login":
for _ in range(3):
if hook_url:
with suppress(NinjaTimeout):
page.wait_for_url(hook_url, timeout=3000)
return self.status.CHALLENGE_SUCCESS, "退火成功"
else:
page.wait_for_timeout(2000)
if page.url != flag:
if "id/login/mfa" not in page.url:
return self.status.CHALLENGE_SUCCESS, "退火成功"
raise AuthMFA("人机挑战已退出 - error=遭遇意外的 MFA 多重认证")
mui_typography = page.locator("//h6")
with suppress(NinjaTimeout):
mui_typography.first.wait_for(timeout=1000, state="attached")
if mui_typography.count() > 1:
with suppress(AttributeError):
error_text = mui_typography.nth(1).text_content().strip()
if "错误回复" in error_text:
self.critical_threshold += 1
return self.status.CHALLENGE_RETRY, "登入页面错误回复"
if "there was a socket open error" in error_text:
return self.status.CHALLENGE_RETRY, "there was a socket open error"
if self.critical_threshold > 3:
logger.debug(f"認證失敗 - {error_text=}")
_unknown = AuthUnknownException(msg=error_text)
_unknown.report(error_text)
raise _unknown
def anti_hcaptcha(
self, page: Page, window: str = "login", recur_url=None, *args, **kwargs
) -> bool | str:
"""
Handle hcaptcha challenge
:param recur_url:
:param window: [login free]
:param page:
:return:
"""
if window == "login":
frame_challenge = page.frame_locator(self.HOOK_CHALLENGE)
else:
frame_purchase = page.frame_locator(self.HOOK_PURCHASE)
frame_challenge = frame_purchase.frame_locator(self.HOOK_CHALLENGE)
try:
# [👻] 人机挑战！
for i in range(2):
page.wait_for_timeout(2000)
# [👻] 获取挑战标签
self.get_label(frame_challenge)
# [👻] 編排定位器索引
self.mark_samples(frame_challenge)
# [👻] 拉取挑戰圖片
self.download_images()
# [👻] 滤除无法处理的挑战类别
if "please click on the" in self._label.lower():
return self.status.CHALLENGE_BACKCALL
if not self._label_alias.get(self._label):
return self.status.CHALLENGE_BACKCALL
# [👻] 注册解决方案
# 根据挑战类型自动匹配不同的模型
model = self.match_solution()
# [👻] 識別|點擊|提交
self.challenge(frame_challenge, model=model)
# [👻] 輪詢控制臺響應
with suppress(TypeError):
result, message = self.is_success(
page, frame_challenge, window=window, init=not i, hook_url=recur_url
)
logger.debug("获取响应", desc=f"{message}({result})")
if result in [
self.status.CHALLENGE_SUCCESS,
self.status.CHALLENGE_CRASH,
self.status.CHALLENGE_RETRY,
]:
return result
page.wait_for_timeout(2000)
# from::mark_samples url = re.split(r'[(")]', image_style)[2]
except IndexError:
return self.anti_hcaptcha(page, window, recur_url)
except ChallengePassed:
return self.status.CHALLENGE_SUCCESS
except Exception as err:
logger.exception(err)
def is_fall_in_captcha(page: Page) -> str | None:
"""判断在登录时是否遇到人机挑战"""
logger.info("正在检测隐藏在登录界面的人机挑战...")
flag = page.url
for _ in range(15):
# 控制台信息
mui_typography = page.locator("//h6")
with suppress(NinjaTimeout):
mui_typography.first.wait_for(timeout=2000, state="attached")
if mui_typography.count() > 1:
error_text = mui_typography.nth(1).text_content().strip()
logger.error(f"認證異常", err=error_text)
if "账号或密码" in error_text:
raise LoginException(error_text)
return Status.AUTH_ERROR
# 頁面重定向|跳過挑戰
if page.url != flag:
logger.info("🥤 跳过人机挑战")
return Status.AUTH_SUCCESS
# 多因素判斷
page.wait_for_timeout(2000)
with suppress(NinjaError):
if page.locator(Radagon.HOOK_CHALLENGE).is_visible():
return Status.AUTH_CHALLENGE