feat(add): 大幅度提升 claim 执行速度 #42

src/services/bricklayer/bricklayer.py

@@ -5,7 +5,7 @@
 # Description:
 import os
 from hashlib import sha256
-from typing import List, Optional, Dict
+from typing import List, Optional, Dict, Union
 
 import cloudscraper
 import yaml
@@ -340,25 +340,28 @@ def get_free_game(
             if _ctx_session is None:
                 ctx.quit()
 
-    def get_free_dlc_details(self, _ctx_session) -> Optional[List[Dict[str, str]]]:
+    def get_free_dlc_details(
+        self, ctx_url: str, ctx_cookies: List[dict]
+    ) -> Optional[List[Dict[str, Union[str, bool]]]]:
         """
 
-        :param _ctx_session: {"url":访问链接 "name":DLC名称 }
+        :param ctx_cookies:
+        :param ctx_url: 游戏本体链接
         :return:
         """
-        urls = self._get_free_dlc_details(_ctx_session)
-        if not urls:
+        dlc_details = self._get_free_dlc_details(ctx_url, ctx_cookies)
+        if not dlc_details:
             return []
-        return urls
+        return dlc_details
 
-    def get_free_dlc(self, dlc_page_link: str, ctx_cookies: List[dict], _ctx_session):
+    def get_free_resources(self, page_link: str, ctx_cookies: List[dict], ctx_session):
         """
 
         :param ctx_cookies:
-        :param dlc_page_link:
-        :param _ctx_session:
+        :param page_link:
+        :param ctx_session:
         :return:
         """
-        return self._get_free_dlc(
-            page_link=dlc_page_link, ctx_cookies=ctx_cookies, ctx=_ctx_session
+        return self._get_free_resources(
+            page_link=page_link, ctx_cookies=ctx_cookies, ctx=ctx_session
         )

src/services/bricklayer/core.py

@@ -9,9 +9,10 @@
 import sys
 import time
 import urllib.request
-from typing import List, Optional, NoReturn, Dict
+from typing import List, Optional, NoReturn, Dict, Union
 
 import cloudscraper
+from lxml import etree
 from requests.exceptions import RequestException
 from selenium.common.exceptions import (
     TimeoutException,
@@ -714,6 +715,7 @@ class AwesomeFreeMan:
     """白嫖人的基础设施"""
 
     # 操作对象参数
+    URL_MASTER_HOST = "https://store.epicgames.com"
     URL_LOGIN_GAMES = "https://www.epicgames.com/id/login/epic?lang=zh-CN"
     URL_LOGIN_UNREAL = "https://www.unrealengine.com/id/login/epic?lang=zh-CN"
     URL_ACCOUNT_PERSONAL = "https://www.epicgames.com/account/personal"
@@ -965,70 +967,79 @@ def _get_free_game(
 
         return self.result
 
-    @staticmethod
-    def _get_free_dlc_details(ctx: Chrome) -> Optional[List[Dict[str, str]]]:
-        # 检测当前商品是否有附加内容
-        try:
-            dlc_tag = ctx.find_element(
-                By.XPATH,
-                "//li[@data-component='PDPTertiaryNavigation']//a[contains(@href,'dlc')]",
-            )
-        except NoSuchElementException:
-            return
+    def _get_free_dlc_details(
+        self, ctx_url: str, ctx_cookies: List[dict]
+    ) -> Optional[List[Dict[str, Union[str, bool]]]]:
+        """
+        1. 检测一个游戏实体是否存在免费附加内容
+        2. 将可领取的免费附加内容编织成任务对象并返回
+        3. 一个游戏实体可能存在多个可领取的免费DLC
+        :param ctx_url: 游戏本体商城链接
+        :param ctx_cookies:
+        :return: [{"url": url of dlc, "name": name of dlc, "dlc": True}, ... ]
+        """
 
-        # 检测当前商品是否有免费的DLC
-        dlc_page = f"{dlc_tag.get_attribute('href')}?sortBy=relevancy&sortDir=DESC&priceTier=tierFree&count=40&start=0"
-        ctx.get(dlc_page)
-        try:
-            ctx.find_element(By.XPATH, "//span[text()='未找到结果']")
+        def handle_html(url_):
+            headers = {
+                "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/100.0.4896.75 Safari/537.36 Edg/100.0.1185.36",
+                "cookie": ToolBox.transfer_cookies(ctx_cookies),
+            }
+            scraper = cloudscraper.create_scraper()
+            response_ = scraper.get(url_, headers=headers, allow_redirects=False)
+            tree_ = etree.HTML(response_.content)
+
+            return tree_, response_
+
+        # [🚀] 检测当前商品是否有附加内容
+        tree, response = handle_html(ctx_url)
+        dlc_tag = tree.xpath(
+            "//li[@data-component='PDPTertiaryNavigation']//a[contains(@href,'dlc')]"
+        )
+        if not dlc_tag:
             return
-        except NoSuchElementException:
-            pass
 
-        # 返回当前商品所有免费DLC链接
-        try:
-            time.sleep(3)
-            WebDriverWait(ctx, 70).until(
-                EC.presence_of_element_located(
-                    (By.XPATH, "//div[@data-component='DiscoverCard']//a")
-                )
-            )
-            dlc_tags = ctx.find_elements(
-                By.XPATH, "//div[@data-component='DiscoverCard']//a"
-            )
-        # 超时/元素不存在或被修改/New Case
-        except WebDriverException:
+        # [🚀] 检测当前商品是否有免费的DLC
+        dlc_page = (
+            f"{self.URL_MASTER_HOST}{dlc_tag[0].attrib.get('href')}?"
+            f"sortBy=relevancy&sortDir=DESC&priceTier=tierFree&count=40&start=0"
+        )
+        dlc_tree, response = handle_html(dlc_page)
+        if dlc_tree.xpath("//span[text()='未找到结果']"):
             return
-        else:
-            dlc_details = []
-            for tag in dlc_tags:
-                # 获取 DLC 名称
-                aria_label = tag.get_attribute("aria-label")
-                try:
-                    name = aria_label.split(",")[0]
-                except (IndexError, AttributeError):
-                    name = ctx.current_url.split("/")[-1]
 
-                # 部分地区账号会被重定向至附加内容的默认页面
-                # 此页面未触发筛选器，混杂着付费/免费的附加内容
-                is_free = True
-                try:
-                    # 重新判断当前游戏的状态，清洗付费游戏
-                    if "tierFree" not in ctx.current_url:
-                        is_free = aria_label.split(",")[-1].strip() == "0"
-                # 当出现意外的标签时将此实例视为免费游戏送入任务队列
-                # 下层驱动中有更加明确的游戏状态用以剔除杂质
-                except (IndexError, AttributeError):
-                    pass
-
-                # 编织缓存
-                if is_free:
-                    dlc_detail = {"url": tag.get_attribute("href"), "name": name}
-                    dlc_details.append(dlc_detail)
-
-            return dlc_details
-
-    def _get_free_dlc(self, page_link: str, ctx_cookies: List[dict], ctx: Chrome):
+        # [🚀] 返回当前商品所有免费DLC链接
+        dlc_tags: list = dlc_tree.xpath("//div[@data-component='DiscoverCard']//a")
+        dlc_details = {}
+        for tag in dlc_tags:
+            # [📝] 获取 DLC 名称
+            aria_label = tag.attrib.get("aria-label")
+            try:
+                name = aria_label.split(",")[0]
+            except (IndexError, AttributeError):
+                name = response.url.split("/")[-1]
+
+            # 部分地区账号会被重定向至附加内容的默认页面
+            # 此页面未触发筛选器，混杂着付费/免费的附加内容
+            is_free = True
+            try:
+                # 重新判断当前游戏的状态，清洗付费游戏
+                if "tierFree" not in response.url or response.status_code == 302:
+                    is_free = aria_label.split(",")[-1].strip() == "0"
+            # 当出现意外的标签时将此实例视为免费游戏送入任务队列
+            # 下层驱动中有更加明确的游戏状态用以剔除杂质
+            except (IndexError, AttributeError):
+                pass
+
+            # 编织缓存
+            if is_free:
+                url = f"{self.URL_MASTER_HOST}{tag.attrib.get('href')}"
+                dlc_detail = {"url": url, "name": name, "dlc": True}
+                dlc_details.update({url: dlc_detail})
+
+        return list(dlc_details.values())
+
+    def _get_free_resources(self, page_link: str, ctx_cookies: List[dict], ctx: Chrome):
         return self._get_free_game(page_link=page_link, api_cookies=ctx_cookies, ctx=ctx)
 
     def _unreal_activate_payment(

src/services/deploy.py

@@ -5,7 +5,7 @@
 # Description:
 import random
 from datetime import datetime, timedelta
-from typing import Optional, List
+from typing import Optional, List, Dict, Any
 
 import apprise
 import pytz
@@ -239,70 +239,69 @@ def _push(self, inline_docker: list, pusher_settings: Optional[dict] = None):
             )
         )
 
-    def claim_free_game(
-        self,
-        challenger,
-        ctx_cookies: List[dict],
-        game_objs: dict,
-        urls: Optional[List[str]] = None,
-    ):
-        """认领周免游戏"""
-        if not urls:
-            self.logger.debug(
+    def promotions_filter(
+        self, promotions: Optional[Dict[str, Any]], ctx_cookies: List[dict]
+    ) -> Optional[List[Dict[str, Any]]]:
+        """
+        促销实体过滤器
+
+        1. 判断游戏本体是否在库
+        2. 判断是否存在免费附加内容
+        3. 识别并弹出已在库资源
+        4. 返回待认领的实体资源
+        :param promotions:
+        :param ctx_cookies:
+        :return:
+        """
+
+        def in_library(page_link: str, name: str):
+            response = self.explorer.game_manager.is_my_game(
+                ctx_cookies=ctx_cookies, page_link=page_link
+            )
+            # 资源待认领
+            if not response["status"] and response["assert"] != "AssertObjectNotFound":
+                self.logger.debug(
+                    ToolBox.runtime_report(
+                        motive="STARTUP",
+                        action_name="ScaffoldClaim",
+                        message="🍜 正在为玩家领取周免游戏",
+                        game=f"『{name}』",
+                    )
+                )
+                return False
+            self.logger.info(
                 ToolBox.runtime_report(
-                    motive="SKIP",
+                    motive="GET",
                     action_name=self.action_name,
-                    message="🛴 当前玩家暂无待认领的周免游戏。",
+                    message="🛴 资源已在库",
+                    game=f"『{name}』",
                 )
             )
-            return
-
-        # 优先处理常规情况 urls.__len__() == 1
-        for url in urls:
-            self.logger.debug(
-                ToolBox.runtime_report(
-                    motive="STARTUP",
-                    action_name="ScaffoldClaim",
-                    message="🍜 正在为玩家领取周免游戏",
-                    game=f"『{game_objs[url]}』",
-                )
+            return True
+
+        if not isinstance(promotions, dict) or not promotions["urls"]:
+            return promotions
+
+        # 过滤资源实体
+        pending_objs = []
+        for url in promotions["urls"]:
+            # 标记已在库游戏本体
+            job_name = promotions[url]
+            pending_objs.append(
+                {"url": url, "name": job_name, "in_library": in_library(url, job_name)}
             )
 
-            # 更新任务队列
-            challenger.switch_to.new_window("tab")
-            self.task_queue.put({"game": challenger.current_window_handle})
-
-            # 反复生产挑战者领取周免游戏
-            self.bricklayer.get_free_game(
-                page_link=url, ctx_cookies=ctx_cookies, _ctx_session=challenger
+            # 识别免费附加内容
+            dlc_details = self.bricklayer.get_free_dlc_details(
+                ctx_url=url, ctx_cookies=ctx_cookies
             )
 
-            # 编制运行缓存 用于生成业务报告
-            _runtime = {"status": self.bricklayer.result, "name": game_objs[url]}
-            self.message_queue.put_nowait(_runtime)
-
-    def claim_free_dlc(self, challenger, ctx_cookies):
-        """认领周免游戏的免费附加内容"""
-        while not self.task_queue.empty():
-            context = self.task_queue.get()
+            # 标记已在库的免费附加内容
+            for dlc in dlc_details:
+                dlc.update({"in_library": in_library(dlc["url"], dlc["name"])})
+                pending_objs.append(dlc)
 
-            # {"game": WebDriver Window}
-            if isinstance(context, dict) and context.get("game"):
-                challenger.switch_to.window(context["game"])
-                dlc_details = self.bricklayer.get_free_dlc_details(
-                    _ctx_session=challenger
-                )
-                for dlc in dlc_details:
-                    self.task_queue.put(dlc)
-            # {"url": link of dlc , "name": alia-label of dlc}
-            elif isinstance(context, dict) and context.get("url"):
-                result = self.bricklayer.get_free_dlc(
-                    dlc_page_link=context["url"],
-                    ctx_cookies=ctx_cookies,
-                    _ctx_session=challenger,
-                )
-                _runtime = {"status": result, "name": context["name"], "dlc": True}
-                self.message_queue.put_nowait(_runtime)
+        return pending_objs
 
     def just_do_it(self):
         """单步子任务 认领周免游戏"""
@@ -314,21 +313,27 @@ def just_do_it(self):
             ctx_cookies = self.bricklayer.cookie_manager.load_ctx_cookies()
 
             # 扫描商城促销活动，返回“0折”商品的名称与商城链接
-            limited_free_game_objs = self.explorer.get_the_absolute_free_game(
-                ctx_cookies, _ctx_session=self.challenger
-            )
+            promotions = self.explorer.get_promotions(ctx_cookies)
 
-            # 释放 Claimer 认领周免游戏
-            urls = limited_free_game_objs["urls"]
-            self.claim_free_game(
-                challenger=self.challenger,
-                ctx_cookies=ctx_cookies,
-                game_objs=limited_free_game_objs,
-                urls=urls,
-            )
+            # 资源聚合过滤 从顶级接口剔除已在库资源
+            game_objs = self.promotions_filter(promotions, ctx_cookies)
 
-            # 释放 Claimer 认领游戏DLC
-            self.claim_free_dlc(challenger=self.challenger, ctx_cookies=ctx_cookies)
+            # 启动任务队列
+            for game in game_objs:
+                if game["in_library"]:
+                    result = self.bricklayer.assert_.GAME_OK
+                else:
+                    result = self.bricklayer.get_free_resources(
+                        page_link=game["url"],
+                        ctx_cookies=ctx_cookies,
+                        ctx_session=self.challenger,
+                    )
+                _runtime = {
+                    "status": result,
+                    "name": game["name"],
+                    "dlc": game.get("dlc", False),
+                }
+                self.message_queue.put_nowait(_runtime)
 
 
 class UnrealClaimerInstance(ClaimerInstance):