EPIC 免费人
-
🚀 优雅地领取 Epic 免费游戏
+
🍷 Gracefully claim weekly free games from Epic Store.
-
+
-
-
-
+
+
@@ -16,20 +15,35 @@
-## 项目简介 👋
+## Introduction 👋
-[Epic AwesomeGamer](https://github.com/QIN2DIM/epic-awesome-gamer) 帮助玩家优雅地领取 Epic 免费游戏。
+[Epic 免费人](https://github.com/QIN2DIM/epic-awesome-gamer) 帮助玩家优雅地领取免费游戏。内置 [hcaptcha-challenger](https://github.com/QIN2DIM/hcaptcha-challenger) AI 模块，直面人机挑战。
-## 快速上手 🛴
+## Guides
-- :gear: [技术文档](https://www.wolai.com/vAiu9mSp6G15xhWeoEPUn2)
-- :small_red_triangle: 注意事项
-- :loudspeaker: 更新日志
-- :world_map: [开源计划](https://github.com/QIN2DIM/epic-awesome-gamer/issues/1)
+[ [`简体中文`](https://echosec.notion.site/Epic-7c74f1e29117420dbac5551e4b031f82?pvs=4) ] [ [`English`](https://echosec.notion.site/Epic-Awesome-Gamer-ba870cdf64c149e69f417448b1eb83c5?pvs=4) ]
-## 联系我们 📧
+## Features
-> 本项目由海南大学机器人与人工智能实验室（`RobAI-Lab`）提供维护。
-- [**Email**](mailto:HainanU_arai@163.com?subject=CampusDailyAutoSign-ISSUE) **||** [**Home**](https://a-rai.github.io/) **||** [**TG**](https://t.me/joinchat/HlB9SQJubb5VmNU5)
+### Task
+
+| Mode | Target | Progress |
+| ---------- | ------------------------------------------------------------ | -------- |
+| epic-games | [Epic Games Store](https://www.epicgames.com/store/free-games) | ✅ |
+| unreal | [Unreal Engine](https://www.unrealengine.com/) | 🚧 |
+| gog | [GOG](https://www.gog.com/) | 🚧 |
+| apg | [Amazon Prime Gaming](https://gaming.amazon.com/) | 🚧 |
+| xbox | [Xbox Live Games with Gold](https://www.xbox.com/en-US/live/gold#gameswithgold) | 🚧 |
+
+### Component
+
+| Demand | Support |
+| :------------------- | :------ |
+| hCaptcha Solver | ✅ |
+| Docker development | ✅ |
+| Persistent context @multi-user | ✅ |
+| Rolling Upgrade | 🚧 |
+| Epicgames DLC | 🚧 |
+| 2FA OTP support | 🚧 |
diff --git a/_token b/_token
new file mode 100644
index 0000000000..38790b853c
--- /dev/null
+++ b/_token
@@ -0,0 +1 @@
+Automated deployment @ 2024-10-23 06:33:15 Asia/Shanghai
diff --git a/requirements.txt b/requirements.txt
index 8c98605794..bdf5340ba5 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,37 +1 @@
-# {{< Scaffolding dependency >}}
-fire>=0.4.0
-
-# {{< Log system >}}
-loguru>=0.5.3
-pyyaml>=6.0
-
-# {{< Timed task >}}
-apscheduler>=3.8.1
-pytz>=2021.3
-
-# {{< Interface /Data cleaning >}}
-requests>=2.27.1
-cloudscraper>=1.2.58
-bs4>=0.0.1
-beautifulsoup4>=4.10.0
-lxml>=4.7.1
-aiohttp>=3.8.1
-
-# {{< Object Detection >}}
-opencv-python>=4.5.5.62
-scikit-image>=0.19.2
-numpy==1.21.5
-
-# {{< Challenger Drive >}}
-gevent>=21.12.0
-selenium>=4.1.0
-webdriver_manager==3.5.2
-undetected_chromedriver==3.1.3
-
-# {{< Message Push >}}
-apprise>=0.9.6
-
-# {{< Conflicting dependencies >}}
-# Manually install according to your needs.
-# Used to skip `tls in tls` authentication.
-# urllib3==1.25.11
+hcaptcha-challenger[playwright]
diff --git a/src/apis/__init__.py b/src/apis/__init__.py
deleted file mode 100644
index 35c4df7bf2..0000000000
--- a/src/apis/__init__.py
+++ /dev/null
@@ -1,5 +0,0 @@
-# -*- coding: utf-8 -*-
-# Time : 2022/1/16 0:25
-# Author : QIN2DIM
-# Github : https://github.com/QIN2DIM
-# Description:
diff --git a/src/apis/scaffold/__init__.py b/src/apis/scaffold/__init__.py
deleted file mode 100644
index 5ba804a652..0000000000
--- a/src/apis/scaffold/__init__.py
+++ /dev/null
@@ -1,29 +0,0 @@
-# -*- coding: utf-8 -*-
-# Time : 2022/1/16 0:25
-# Author : QIN2DIM
-# Github : https://github.com/QIN2DIM
-# Description:
-import os
-import random
-import sys
-
-_THE_PROPHECY = """
-# ====================================================================================================
-# Ithlinne's Prophecy
-# ====================================================================================================
-# Verily I say unto you,the era of the sword and axe is nigh, the era of the wolf's blizzard.
-# The Time of the White Chill and the White Light is nigh, the Time of Madness and the Time of Contempt:
-# Tedd Deireádh, the Time of End. The world will die amidst frost and be reborn with the new sun.
-# It will be reborn of Elder Blood, of Hen Ichaer, of the seed that has been sown.
-# A seed which will not sprout but burst into flame.
-# Ess'tuath esse! Thus it shall be! Watch for the signs!
-# What signs these shall be, I say unto you:
-# first the earth will flow with the blood of claim , the Blood of Epic .
-# ====================================================================================================
-"""
-if random.uniform(0, 1) > 0.711:
- for policy in ["epic", "claim"]:
- if policy in os.getenv("GITHUB_REPOSITORY", "").lower():
- print(f"[EXIT] 仓库名出现非法关键词 `{policy}`")
- sys.exit()
- print(_THE_PROPHECY)
diff --git a/src/apis/scaffold/challenge.py b/src/apis/scaffold/challenge.py
deleted file mode 100644
index ec841f9166..0000000000
--- a/src/apis/scaffold/challenge.py
+++ /dev/null
@@ -1,41 +0,0 @@
-# -*- coding: utf-8 -*-
-# Time : 2022/1/16 0:25
-# Author : QIN2DIM
-# Github : https://github.com/QIN2DIM
-# Description:
-from typing import Optional
-
-from services.bricklayer import Bricklayer
-from services.settings import PATH_USR_COOKIES, logger
-from services.utils import ToolBox
-
-bricklayer = Bricklayer()
-
-
-def run(silence: Optional[bool] = None):
- """刷新上下文身份令牌"""
- logger.info(
- ToolBox.runtime_report(
- motive="STARTUP", action_name="ScaffoldChallenge", message="正在更新身份令牌..."
- )
- )
-
- # [🌀] 激活人机挑战
- if not bricklayer.cookie_manager.refresh_ctx_cookies(silence=silence):
- return
-
- # [🌀] 读取新的身份令牌
- ctx_cookies = bricklayer.cookie_manager.load_ctx_cookies()
-
- # [🌀] 保存身份令牌
- with open(PATH_USR_COOKIES, "w", encoding="utf8") as file:
- file.write(ToolBox.transfer_cookies(ctx_cookies))
-
- logger.success(
- ToolBox.runtime_report(
- motive="GET",
- action_name="ChallengeRunner",
- message="玩家饼干已到货。",
- path=PATH_USR_COOKIES,
- )
- )
diff --git a/src/apis/scaffold/claimer.py b/src/apis/scaffold/claimer.py
deleted file mode 100644
index 8e38655402..0000000000
--- a/src/apis/scaffold/claimer.py
+++ /dev/null
@@ -1,21 +0,0 @@
-# -*- coding: utf-8 -*-
-# Time : 2022/1/16 0:25
-# Author : QIN2DIM
-# Github : https://github.com/QIN2DIM
-# Description:
-from typing import Optional
-
-from services.deploy import ClaimerScheduler
-from services.settings import logger
-
-
-@logger.catch()
-def deploy(platform: Optional[str] = None):
- """在微小容器中部署 `claim` 定时调度任务"""
- ClaimerScheduler(silence=True).deploy_jobs(platform)
-
-
-@logger.catch()
-def run(silence: Optional[bool] = None):
- """运行 `claim` 单步子任务，认领周免游戏"""
- ClaimerScheduler(silence=silence).job_loop_claim()
diff --git a/src/apis/scaffold/get.py b/src/apis/scaffold/get.py
deleted file mode 100644
index 3ee4d1b0a7..0000000000
--- a/src/apis/scaffold/get.py
+++ /dev/null
@@ -1,127 +0,0 @@
-# -*- coding: utf-8 -*-
-# Time : 2022/1/16 0:25
-# Author : QIN2DIM
-# Github : https://github.com/QIN2DIM
-# Description:
-import random
-from typing import Optional
-
-from selenium.common.exceptions import WebDriverException
-
-from services.bricklayer import Bricklayer
-from services.explorer import Explorer
-from services.settings import logger
-from services.utils import CoroutineSpeedup, ToolBox
-
-SILENCE = True
-
-bricklayer = Bricklayer(silence=SILENCE)
-explorer = Explorer(silence=SILENCE)
-
-
-class SpawnBooster(CoroutineSpeedup):
- """协程助推器 并发执行片段代码"""
-
- def __init__(
- self,
- docker,
- ctx_cookies,
- power: Optional[int] = None,
- debug: Optional[bool] = None,
- ):
- super().__init__(docker=docker, power=power)
-
- self.debug = False if debug is None else debug
- self.power = min(4, 4 if power is None else power)
- self.action_name = "SpawnBooster"
-
- self.ctx_cookies = ctx_cookies
-
- if self.docker:
- random.shuffle(self.docker)
-
- def control_driver(self, task, *args, **kwargs):
- url = task
-
- # 运行前置检查
- response = explorer.game_manager.is_my_game(
- ctx_cookies=self.ctx_cookies, page_link=url
- )
-
- # 识别未在库的常驻周免游戏
- if response.get("status") is False:
- logger.debug(
- ToolBox.runtime_report(
- motive="BUILD",
- action_name=self.action_name,
- message="🛒 正在为玩家领取免费游戏",
- progress=f"[{self.progress()}]",
- url=url,
- )
- )
-
- # 启动 Bricklayer 获取免费游戏
- try:
- bricklayer.get_free_game(
- page_link=url, ctx_cookies=self.ctx_cookies, refresh=False
- )
- except WebDriverException as error:
- if self.debug:
- logger.exception(error)
- logger.error(
- ToolBox.runtime_report(
- motive="QUIT",
- action_name="SpawnBooster",
- message="未知错误",
- progress=f"[{self.progress()}]",
- url=url,
- )
- )
-
- def killer(self):
- logger.success(
- ToolBox.runtime_report(
- motive="OVER", action_name=self.action_name, message="✔ 任务队列已清空"
- )
- )
-
-
-def join(trace: bool = False, cache: bool = True):
- """
- 一键搬空免费商店
-
- 需要确保上下文身份令牌有效，可通过 `challenge` 脚手架强制刷新。
- :param cache:
- :param trace:
- :return:
- """
- from gevent import monkey
-
- monkey.patch_all(ssl=False)
-
- logger.info(
- ToolBox.runtime_report(
- motive="STARTUP", action_name="ScaffoldGet", message="🔨 正在为玩家领取免费游戏"
- )
- )
-
- # [🔨] 读取有效的身份令牌
- ctx_cookies = bricklayer.cookie_manager.load_ctx_cookies()
- if not bricklayer.cookie_manager.is_available_cookie(ctx_cookies):
- logger.critical(
- ToolBox.runtime_report(
- motive="SKIP",
- action_name="ScaffoldGet",
- message="身份令牌不存在或失效，手动执行 `challenge` 指令更新身份令牌。",
- )
- )
- return
-
- # [🔨] 缓存免费商城数据
- urls = explorer.game_manager.load_game_objs(only_url=True)
- if not cache or not urls:
- urls = explorer.discovery_free_games(ctx_cookies=ctx_cookies, cover=True)
-
- # [🔨] 启动 Bricklayer 搬空免费商店
- # 启动一轮协程任务，执行效率受限于本地网络带宽
- SpawnBooster(ctx_cookies=ctx_cookies, docker=urls, power=4, debug=trace).speedup()
diff --git a/src/apis/scaffold/install.py b/src/apis/scaffold/install.py
deleted file mode 100644
index 6c6286d99c..0000000000
--- a/src/apis/scaffold/install.py
+++ /dev/null
@@ -1,72 +0,0 @@
-# -*- coding: utf-8 -*-
-# Time : 2022/1/20 16:16
-# Author : QIN2DIM
-# Github : https://github.com/QIN2DIM
-# Description:
-import sys
-import webbrowser
-
-from webdriver_manager.chrome import ChromeDriverManager
-from webdriver_manager.utils import get_browser_version_from_os
-
-from services.settings import DIR_MODEL, logger, PATH_RAINBOW
-from services.utils import SKRecognition
-from services.utils import YOLO
-from services.utils import get_challenge_ctx
-
-
-def _download_model(onnx_prefix: str = None):
- """下载 YOLOv4 目标检测模型"""
- logger.debug("Downloading YOLOv5(ONNX) object detection model...")
-
- YOLO(dir_model=DIR_MODEL, onnx_prefix=onnx_prefix).download_model()
-
-
-def _download_driver():
- """下载浏览器驱动"""
- logger.debug("Downloading ChromeDriver...")
-
- # 自动下载并授权对应版本的 ChromeDriver
- browser_version = get_browser_version_from_os("google-chrome")
- if browser_version != "UNKNOWN":
- return ChromeDriverManager(version="latest").install()
-
- # 环境变量中缺少 `google-chrome` 提示玩家手动安装
- logger.critical("当前环境变量缺少 `google-chrome`，请为你的设备手动安装 Chrome 浏览器。")
- logger.info(
- "Ubuntu: https://linuxize.com/post/how-to-install-google-chrome-web-browser-on-ubuntu-20-04/"
- )
- logger.info(
- "CentOS 7/8: https://linuxize.com/post/how-to-install-google-chrome-web-browser-on-centos-7/"
- )
- if "linux" not in sys.platform:
- webbrowser.open("https://www.google.com/chrome/")
-
- logger.info("安装完毕后重新执行 `install` 脚手架指令。")
-
-
-def _download_rainbow():
- logger.debug("Downloading Reinforcement of Memory | Rainbow Table...")
-
- SKRecognition.sync_rainbow(path_rainbow=PATH_RAINBOW, convert=True)
-
-
-def run(onnx_prefix: str = None):
- """下载项目运行所需的各项依赖"""
- logger.debug("正在下载系统依赖")
- _download_driver()
- _download_model(onnx_prefix=onnx_prefix)
- _download_rainbow()
- logger.success("系统依赖下载完毕")
-
-
-@logger.catch()
-def test():
- """检查挑战者驱动版本是否适配"""
- ctx = get_challenge_ctx(silence=True)
- try:
- ctx.get("https://www.baidu.com")
- finally:
- ctx.quit()
-
- logger.success("驱动适配成功")
diff --git a/src/claim.py b/src/claim.py
new file mode 100644
index 0000000000..bfbc8f9763
--- /dev/null
+++ b/src/claim.py
@@ -0,0 +1,162 @@
+# -*- coding: utf-8 -*-
+# Time : 2023/8/16 5:14
+# Author : QIN2DIM
+# GitHub : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+from dataclasses import dataclass, field
+from typing import List
+
+import importlib_metadata
+from hcaptcha_challenger import install
+from hcaptcha_challenger.agents import Malenia
+from loguru import logger
+from playwright.async_api import BrowserContext, async_playwright
+
+from epic_games import (
+ EpicPlayer,
+ EpicGames,
+ Game,
+ CompletedOrder,
+ get_promotions,
+ get_order_history,
+)
+
+self_supervised = True
+
+
+@dataclass
+class ISurrender:
+ player: EpicPlayer
+
+ promotions: List[Game] = field(default_factory=list)
+ ctx_cookies_is_available: bool = None
+ headless: bool = True
+ locale: str = "en-US"
+
+ _orders = None
+ _namespaces = None
+ _pros = None
+
+ def __post_init__(self):
+ self._orders: List[CompletedOrder] = []
+ self._namespaces: List[str] = []
+ self._pros: List[Game] = []
+
+ @classmethod
+ def from_epic(cls):
+ return cls(player=EpicPlayer.from_account())
+
+ @property
+ def cookies(self):
+ return self.player.cookies
+
+ def create_tasks(self):
+ if not self._orders:
+ self._orders = get_order_history(self.cookies)
+ if not self._namespaces:
+ self._namespaces = [order.namespace for order in self._orders]
+ if not self._pros:
+ self._pros = get_promotions()
+ for pro in self._pros:
+ logger.debug("Put task", title=pro.title, url=pro.url)
+
+ self.promotions = [p for p in self._pros if p.namespace not in self._namespaces]
+
+ async def prelude_with_context(self, context: BrowserContext) -> bool | None:
+ url = "https://www.epicgames.com/account/creator-programs"
+ page = context.pages[0]
+ await page.goto(url, wait_until="networkidle")
+ if not page.url.startswith(url):
+ return
+
+ self.ctx_cookies_is_available = True
+ await context.storage_state(path=self.player.ctx_cookie_path)
+ cookies = self.player.ctx_cookies.reload(self.player.ctx_cookie_path)
+ self.player.cookies = cookies
+
+ self.create_tasks()
+
+ if not self.promotions:
+ logger.success(
+ "Pass claim task",
+ reason="All free games are in my library",
+ stage="context-prelude",
+ )
+ return True
+
+ async def claim_epic_games(self, context: BrowserContext):
+ page = context.pages[0]
+ epic = EpicGames.from_player(self.player, page=page, self_supervised=self_supervised)
+
+ if not self.ctx_cookies_is_available:
+ logger.info("Try to flush cookie", task="claim_epic_games")
+ if await epic.authorize(page):
+ cookies = await epic.flush_token(context)
+ self.player.cookies = cookies
+ else:
+ logger.error("Exit task", reason="Failed to flush token")
+ return
+
+ if not self.promotions:
+ self.create_tasks()
+ if not self.promotions:
+ logger.success(
+ "Pass claim task", reason="All free games are in my library", stage="claim-games"
+ )
+ return
+
+ single_promotions = []
+ bundle_promotions = []
+ for p in self.promotions:
+ if "bundles" in p.url:
+ bundle_promotions.append(p)
+ else:
+ single_promotions.append(p)
+
+ if single_promotions:
+ await epic.claim_weekly_games(page, single_promotions)
+ if bundle_promotions:
+ await epic.claim_bundle_games(page, bundle_promotions)
+
+ @logger.catch
+ async def stash(self):
+ if "linux" in sys.platform and "DISPLAY" not in os.environ:
+ self.headless = True
+
+ logger.info(
+ "run",
+ image="20231121",
+ version=importlib_metadata.version("hcaptcha-challenger"),
+ role="EpicPlayer",
+ headless=self.headless,
+ )
+
+ async with async_playwright() as p:
+ context = await p.firefox.launch_persistent_context(
+ user_data_dir=self.player.browser_context_dir,
+ record_video_dir=self.player.record_dir,
+ record_har_path=self.player.record_har_path,
+ headless=self.headless,
+ locale=self.locale,
+ args=["--hide-crash-restore-bubble"],
+ )
+ await Malenia.apply_stealth(context)
+ if not await self.prelude_with_context(context):
+ install(upgrade=True, clip=True)
+ await self.claim_epic_games(context)
+ await context.close()
+
+
+async def run():
+ agent = ISurrender.from_epic()
+ agent.headless = False
+ await agent.stash()
+
+
+if __name__ == "__main__":
+ asyncio.run(run())
diff --git a/src/config-sample.yaml b/src/config-sample.yaml
deleted file mode 100644
index 9d5e4859ff..0000000000
--- a/src/config-sample.yaml
+++ /dev/null
@@ -1,23 +0,0 @@
-# [√]需要手动修改 [-]不推荐修改 [※]可选项，不影响系统运行
-
-# ===================================================
-# [√] 账号信息
-# ---------------------------------------------------
-# Tips：不建议在公有库上创建工作流运行项目，有仓库禁用风险
-# ===================================================
-EPΙC_EMAΙL: ""
-EPΙC_PASSWΟRD: ""
-# ===================================================
-# [√※] 通信鸽
-# ---------------------------------------------------
-# 一些主流的通信接口，用于反射运行报告
-# Docs https://github.com/caronc/apprise
-# ===================================================
-message_pusher_settings:
- enable: true
- pusher:
- PUSHER_EMAIL: ""
- PUSHER_DINGTALK: ""
- PUSHER_TELEGRAM: ""
- PUSHER_SERVERCHAN: ""
- PUSHER_DISCORD: ""
diff --git a/src/epic_games/__init__.py b/src/epic_games/__init__.py
new file mode 100644
index 0000000000..15cd18d933
--- /dev/null
+++ b/src/epic_games/__init__.py
@@ -0,0 +1,17 @@
+# -*- coding: utf-8 -*-
+# Time : 2022/1/16 0:24
+# Author : QIN2DIM
+# Github : https://github.com/QIN2DIM
+# Description:
+from .agent import EpicGames, Game, CompletedOrder, get_promotions, get_order_history
+
+from .player import EpicPlayer
+
+__all__ = [
+ "EpicGames",
+ "Game",
+ "CompletedOrder",
+ "get_order_history",
+ "get_promotions",
+ "EpicPlayer",
+]
diff --git a/src/epic_games/agent.py b/src/epic_games/agent.py
new file mode 100644
index 0000000000..cbc67d926e
--- /dev/null
+++ b/src/epic_games/agent.py
@@ -0,0 +1,469 @@
+# -*- coding: utf-8 -*-
+# Time : 2023/8/14 23:16
+# Author : QIN2DIM
+# Github : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+import json
+from contextlib import suppress
+from dataclasses import dataclass, field
+from json import JSONDecodeError
+from pathlib import Path
+from typing import List, Dict, Literal
+
+import httpx
+from loguru import logger
+from playwright.async_api import BrowserContext, expect, TimeoutError, Page, FrameLocator, Locator
+from tenacity import *
+
+from epic_games.player import EpicPlayer
+from utils import from_dict_to_model, AgentG
+
+# fmt:off
+URL_CLAIM = "https://store.epicgames.com/en-US/free-games"
+URL_LOGIN = f"https://www.epicgames.com/id/login?lang=en-US&noHostRedirect=true&redirectUrl={URL_CLAIM}"
+URL_PROMOTIONS = "https://store-site-backend-static.ak.epicgames.com/freeGamesPromotions"
+URL_PRODUCT_PAGE = "https://store.epicgames.com/en-US/p/"
+URL_PRODUCT_BUNDLES = "https://store.epicgames.com/en-US/bundles/"
+URL_ORDER_HISTORY = "https://www.epicgames.com/account/v2/payment/ajaxGetOrderHistory"
+URL_CART = "https://store.epicgames.com/en-US/cart"
+URL_CART_SUCCESS = "https://store.epicgames.com/en-US/cart/success"
+# -----
+URL_STORE_EXPLORER = "https://store.epicgames.com/en-US/browse?sortBy=releaseDate&sortDir=DESC&priceTier=tierFree&count=40"
+URL_STORE_EXPLORER_GRAPHQL = (
+ "https://store.epicgames.com/graphql?operationName=searchStoreQuery"
+ '&variables={"category":"games/edition/base","comingSoon":false,"count":80,"freeGame":true,"keywords":"","sortBy":"releaseDate","sortDir":"DESC","start":0,"tag":"","withPrice":true}'
+ '&extensions={"persistedQuery":{"version":1,"sha256Hash":"13a2b6787f1a20d05c75c54c78b1b8ac7c8bf4efc394edf7a5998fdf35d1adb0"}}'
+)
+
+# fmt:on
+
+
+@dataclass
+class CompletedOrder:
+ offerId: str
+ namespace: str
+
+
+@dataclass
+class Game:
+ url: str
+ namespace: str
+ title: str
+ thumbnail: str
+ id: str
+ in_library = None
+
+
+class CommonHandler:
+ @staticmethod
+ async def any_license(page: Page):
+ with suppress(TimeoutError):
+ await page.click("//label[@for='agree']", timeout=2000)
+ accept = page.locator("//button//span[text()='Accept']")
+ if await accept.is_enabled():
+ await accept.click()
+
+ @staticmethod
+ async def move_to_purchase_container(page: Page):
+ wpc = page.frame_locator("//iframe[@class='']")
+ payment_btn = wpc.locator("//div[@class='payment-order-confirm']")
+ with suppress(Exception):
+ await expect(payment_btn).to_be_attached()
+ await page.wait_for_timeout(2000)
+ await payment_btn.click(timeout=6000)
+
+ return wpc, payment_btn
+
+ @staticmethod
+ async def uk_confirm_order(wpc: FrameLocator):
+ # <-- Handle UK confirm-order
+ with suppress(TimeoutError):
+ accept = wpc.locator(
+ "//button[contains(@class, 'payment-confirm__btn payment-btn--primary')]"
+ )
+ if await accept.is_enabled(timeout=5000):
+ await accept.click()
+ return True
+
+ @staticmethod
+ @retry(
+ retry=retry_if_exception_type(TimeoutError),
+ wait=wait_fixed(0.5),
+ stop=stop_after_attempt(15),
+ reraise=True,
+ )
+ async def insert_challenge(
+ solver: AgentG,
+ page: Page,
+ wpc: FrameLocator,
+ payment_btn: Locator,
+ recur_url: str,
+ is_uk: bool,
+ ):
+ response = await solver.execute(window="free")
+ logger.debug("task done", sattus=f"{solver.status.CHALLENGE_SUCCESS}")
+
+ match response:
+ case solver.status.CHALLENGE_BACKCALL | solver.status.CHALLENGE_RETRY:
+ await wpc.locator("//a[@class='talon_close_button']").click()
+ await page.wait_for_timeout(1000)
+ if is_uk:
+ await CommonHandler.uk_confirm_order(wpc)
+ await payment_btn.click(delay=200)
+ case solver.status.CHALLENGE_SUCCESS:
+ await page.wait_for_url(recur_url)
+ return
+
+ @staticmethod
+ async def empty_cart(page: Page, wait_rerender: int = 30) -> bool | None:
+ """
+ URL_CART = "https://store.epicgames.com/en-US/cart"
+ URL_WISHLIST = "https://store.epicgames.com/en-US/wishlist"
+ //span[text()='Your Cart is empty.']
+
+ Args:
+ wait_rerender:
+ page:
+
+ Returns:
+
+ """
+ has_paid_free = False
+
+ try:
+ # Check all items in the shopping cart
+ cards = await page.query_selector_all("//div[@data-testid='offer-card-layout-wrapper']")
+
+ # Move paid games to wishlist games
+ for card in cards:
+ is_free = await card.query_selector("//span[text()='Free']")
+ if not is_free:
+ has_paid_free = True
+ wishlist_btn = await card.query_selector(
+ "//button//span[text()='Move to wishlist']"
+ )
+ await wishlist_btn.click()
+
+ # Wait up to 60 seconds for the page to re-render.
+ # Usually it takes 1~3s for the web page to be re-rendered
+ # - Set threshold for overflow in case of poor Epic network
+ # - It can also prevent extreme situations, such as: the user’s shopping cart has nearly a hundred products
+ if has_paid_free and wait_rerender:
+ wait_rerender -= 1
+ await page.wait_for_timeout(2000)
+ return await CommonHandler.empty_cart(page, wait_rerender)
+ return True
+ except TimeoutError as err:
+ logger.warning("Failed to empty shopping cart", err=err)
+ return False
+
+
+@dataclass
+class EpicGames:
+ player: EpicPlayer
+ """
+ Agent control
+ """
+
+ _solver: AgentG = None
+ """
+ Module for anti-captcha
+ """
+
+ _promotions: List[Game] = field(default_factory=list)
+ """
+ Free promotional items for the week,
+ considered metadata for task sequence of the agent
+ """
+
+ @classmethod
+ def from_player(
+ cls, player: EpicPlayer, *, page: Page, tmp_dir: Path | None = None, **solver_opt
+ ):
+ """尽可能早地实例化，用于部署 captcha 事件监听器"""
+ return cls(
+ player=player, _solver=AgentG.from_page(page=page, tmp_dir=tmp_dir, **solver_opt)
+ )
+
+ @property
+ def handle(self):
+ return CommonHandler
+
+ @property
+ def promotions(self) -> List[Game]:
+ self._promotions = self._promotions or get_promotions()
+ return self._promotions
+
+ async def _login(self, page: Page) -> str | None:
+ async def insert_challenge(stage: Literal["email_exists_prod", "login_prod"]):
+ fall_in_challenge = False
+
+ for _ in range(15):
+ if stage == "login_prod":
+ if not fall_in_challenge:
+ with suppress(TimeoutError):
+ await page.wait_for_url(URL_CART_SUCCESS, timeout=3000)
+ break
+ logger.debug("Attack challenge", stage=stage)
+ elif stage == "email_exists_prod":
+ if not fall_in_challenge:
+ with suppress(TimeoutError):
+ await page.type("#password", "", timeout=3000)
+ break
+ logger.debug("Attack challenge", stage=stage)
+ fall_in_challenge = True
+ result = await self._solver.execute(window=stage)
+ logger.debug("Parse result", stage=stage, result=result)
+ match result:
+ case self._solver.status.CHALLENGE_BACKCALL:
+ await page.click("//a[@class='talon_close_button']")
+ await page.wait_for_timeout(1000)
+ await page.click("#sign-in", delay=200)
+ case self._solver.status.CHALLENGE_RETRY:
+ continue
+ case self._solver.status.CHALLENGE_SUCCESS:
+ if stage == "signin" and not self._solver.qr_queue.empty():
+ continue
+ with suppress(TimeoutError):
+ await page.wait_for_url(URL_CLAIM)
+ break
+ return
+
+ await page.goto(URL_CLAIM, wait_until="domcontentloaded")
+ if "false" == await page.locator("//egs-navigation").get_attribute("isloggedin"):
+ await page.goto(URL_LOGIN, wait_until="domcontentloaded")
+ logger.info("login-with-email", url=page.url)
+
+ # {{< SIGN IN PAGE >}}
+ await page.fill("#email", self.player.email)
+ await page.click("//button[@aria-label='Continue']")
+
+ # {{< INSERT CHALLENGE - email_exists_prod >}}
+ await insert_challenge(stage="email_exists_prod")
+
+ # {{< NESTED PAGE >}}
+ await page.type("#password", self.player.password)
+ await page.click("#sign-in")
+
+ # {{< INSERT CHALLENGE - login_prod >}}
+ await insert_challenge(stage="login_prod")
+
+ logger.success("login", result="Successfully refreshed tokens")
+ await page.goto(URL_CLAIM, wait_until="domcontentloaded")
+ return self._solver.status.CHALLENGE_SUCCESS
+
+ async def authorize(self, page: Page):
+ for i in range(3):
+ try:
+ match await self._login(page):
+ case self._solver.status.CHALLENGE_SUCCESS:
+ return True
+ case _:
+ continue
+ except TimeoutError:
+ logger.warning("执行超时", task="authorize", retry=i)
+ continue
+
+ raise RuntimeError(f"Failed to flush token - agent={self.__class__.__name__}")
+
+ async def flush_token(self, context: BrowserContext) -> Dict[str, str] | None:
+ page = context.pages[0]
+ await page.goto("https://www.epicgames.com/account/personal", wait_until="networkidle")
+ await page.goto(
+ "https://store.epicgames.com/zh-CN/p/orwell-keeping-an-eye-on-you",
+ wait_until="networkidle",
+ )
+ await context.storage_state(path=self.player.ctx_cookie_path)
+ cookies = self.player.ctx_cookies.reload(self.player.ctx_cookie_path)
+ logger.success("flush_token", path=self.player.ctx_cookie_path)
+ return cookies
+
+ @retry(
+ retry=retry_if_exception_type(TimeoutError),
+ wait=wait_fixed(0.5),
+ stop=(stop_after_delay(360) | stop_after_attempt(3)),
+ reraise=True,
+ )
+ async def claim_weekly_games(self, page: Page, promotions: List[Game]):
+ in_cart_nums = 0
+
+ # --> Add promotions to Cart
+ for promotion in promotions:
+ logger.info("claim_weekly_games", action="go to store", url=promotion.url)
+ await page.goto(promotion.url, wait_until="load")
+
+ # <-- Handle pre-page
+ with suppress(TimeoutError):
+ await page.click("//button//span[text()='Continue']", timeout=3000)
+
+ # --> Make sure promotion is not in the library before executing
+ cta_btn = page.locator("//aside//button[@data-testid='add-to-cart-cta-button']")
+ with suppress(TimeoutError):
+ text = await cta_btn.text_content(timeout=10000)
+ if text == "View In Cart":
+ in_cart_nums += 1
+ continue
+ if text == "Add To Cart":
+ await cta_btn.click()
+ await expect(cta_btn).to_have_text("View In Cart")
+ in_cart_nums += 1
+
+ if in_cart_nums == 0:
+ logger.success("Pass claim task", reason="Free games not added to shopping cart")
+ return
+
+ # --> Goto cart page
+ await page.goto(URL_CART, wait_until="domcontentloaded")
+ await self.handle.empty_cart(page)
+ await page.click("//button//span[text()='Check Out']")
+
+ # <-- Handle Any LICENSE
+ await self.handle.any_license(page)
+
+ # --> Move to webPurchaseContainer iframe
+ logger.info("claim_weekly_games", action="move to webPurchaseContainer iframe")
+ wpc, payment_btn = await self.handle.move_to_purchase_container(page)
+ logger.info("claim_weekly_games", action="click payment button")
+
+ # <-- Handle UK confirm-order
+ is_uk = await self.handle.uk_confirm_order(wpc)
+
+ # <-- Insert challenge
+ recur_url = URL_CART_SUCCESS
+ await self.handle.insert_challenge(self._solver, page, wpc, payment_btn, recur_url, is_uk)
+
+ # --> Wait for success
+ await page.wait_for_url(recur_url)
+ logger.success("claim_weekly_games", action="success", url=page.url)
+
+ return True
+
+ @retry(
+ retry=retry_if_exception_type(TimeoutError),
+ wait=wait_fixed(0.5),
+ stop=(stop_after_delay(360) | stop_after_attempt(3)),
+ reraise=True,
+ )
+ async def claim_bundle_games(self, page: Page, promotions: List[Game]):
+ for promotion in promotions:
+ logger.info("claim_bundle_games", action="go to store", url=promotion.url)
+ await page.goto(promotion.url, wait_until="load")
+
+ # <-- Handle pre-page
+ with suppress(TimeoutError):
+ await page.click("//button//span[text()='Continue']", timeout=3000)
+
+ # --> Make sure promotion is not in the library before executing
+ purchase_btn = page.locator("//button[@data-testid='purchase-cta-button']").first
+ with suppress(TimeoutError):
+ text = await purchase_btn.text_content(timeout=10000)
+ if text == "Get":
+ await purchase_btn.click()
+ await page.wait_for_timeout(2000)
+ else:
+ continue
+
+ # <-- Handle Any LICENSE
+ await self.handle.any_license(page)
+
+ # --> Move to webPurchaseContainer iframe
+ logger.info("claim_bundle_games", action="move to webPurchaseContainer iframe")
+ wpc, payment_btn = await self.handle.move_to_purchase_container(page)
+ logger.info("claim_bundle_games", action="click payment button")
+
+ # <-- Handle UK confirm-order
+ is_uk = await self.handle.uk_confirm_order(wpc)
+
+ # <-- Insert challenge
+ recur_url = f"https://store.epicgames.com/en-US/download?ns={promotion.namespace}&id={promotion.id}"
+ await self.handle.insert_challenge(
+ self._solver, page, wpc, payment_btn, recur_url, is_uk
+ )
+
+ # --> Wait for success
+ await page.wait_for_url(recur_url)
+ logger.success("claim_bundle_games", action="success", url=page.url)
+
+ return True
+
+
+def get_promotions() -> List[Game]:
+ """
+ 获取周免游戏数据
+
+ <即将推出> promotion["promotions"]["upcomingPromotionalOffers"]
+ <本周免费> promotion["promotions"]["promotionalOffers"]
+ :return: {"pageLink1": "pageTitle1", "pageLink2": "pageTitle2", ...}
+ """
+
+ def _has_discount_target(prot: dict) -> bool | None:
+ with suppress(KeyError, IndexError, TypeError):
+ offers = prot["promotions"]["promotionalOffers"][0]["promotionalOffers"]
+ for i, offer in enumerate(offers):
+ if offer["discountSetting"]["discountPercentage"] == 0:
+ return True
+
+ _promotions: List[Game] = []
+
+ params = {"local": "zh-CN"}
+ resp = httpx.get(URL_PROMOTIONS, params=params)
+ try:
+ data = resp.json()
+ except JSONDecodeError as err:
+ logger.error("Failed to get promotions", err=err)
+ else:
+ elements = data["data"]["Catalog"]["searchStore"]["elements"]
+ promotions = [e for e in elements if e.get("promotions")]
+ # Get store promotion data and