diff --git a/src/services/bricklayer/bricklayer.py b/src/services/bricklayer/bricklayer.py index 6655ae0bb0..58e9029dea 100644 --- a/src/services/bricklayer/bricklayer.py +++ b/src/services/bricklayer/bricklayer.py @@ -37,6 +37,7 @@ def __init__(self, auth_str="games"): self.action_name = "CookieManager" self.service_mode = auth_str + self.ctx_session = None def _t(self) -> str: return ( @@ -121,11 +122,12 @@ def is_available_cookie(self, ctx_cookies: Optional[List[dict]] = None) -> bool: return False def refresh_ctx_cookies( - self, silence: bool = True, _ctx_session=None + self, silence: bool = True, _ctx_session=None, _keep_live=None ) -> Optional[bool]: """ 更新上下文身份信息 + :param _keep_live: keep actively to the challenger context :param _ctx_session: 泛型开发者参数 :param silence: :return: @@ -225,7 +227,10 @@ def refresh_ctx_cookies( return self.is_available_cookie(ctx_cookies=ctx.get_cookies()) finally: if _ctx_session is None: - ctx.quit() + if not _keep_live: + ctx.quit() + else: + self.ctx_session = ctx # {{< Done >}} return True @@ -248,13 +253,13 @@ def get_free_game( ctx_cookies: List[dict] = None, refresh: bool = True, challenge: Optional[bool] = None, - _ctx_session=None, + ctx_session=None, ) -> Optional[bool]: """ 获取免费游戏 部署后必须传输有效的 `page_link` 参数。 - :param _ctx_session: + :param ctx_session: :param challenge: :param page_link: 游戏购买页链接 zh-CN :param refresh: 当 COOKIE 失效时主动刷新 COOKIE @@ -284,10 +289,10 @@ def get_free_game( return False # [🚀] 常驻免费（General）周免（Challenge） - if _ctx_session is None: + if ctx_session is None: ctx = get_challenge_ctx(self.silence) if challenge else get_ctx(self.silence) else: - ctx = _ctx_session + ctx = ctx_session # [🚀] 认领游戏 try: @@ -337,31 +342,20 @@ def get_free_game( ) return False finally: - if _ctx_session is None: + if ctx_session is None: ctx.quit() def get_free_dlc_details( self, ctx_url: str, ctx_cookies: List[dict] ) -> Optional[List[Dict[str, Union[str, bool]]]]: - """ - - :param ctx_cookies: - :param ctx_url: 游戏本体链接 - :return: - """ + """获取免费附加内容信息""" dlc_details = self._get_free_dlc_details(ctx_url, ctx_cookies) if not dlc_details: return [] return dlc_details - def get_free_resources(self, page_link: str, ctx_cookies: List[dict], ctx_session): - """ - - :param ctx_cookies: - :param page_link: - :param ctx_session: - :return: - """ - return self._get_free_resources( + def get_free_games(self, page_link: str, ctx_cookies: List[dict], ctx_session): + """获取周免资源 游戏本体/附加内容 集成接口""" + return self._get_free_resource( page_link=page_link, ctx_cookies=ctx_cookies, ctx=ctx_session ) diff --git a/src/services/bricklayer/core.py b/src/services/bricklayer/core.py index 9189af8ad1..d67a2bf669 100644 --- a/src/services/bricklayer/core.py +++ b/src/services/bricklayer/core.py @@ -1039,7 +1039,7 @@ def handle_html(url_): return list(dlc_details.values()) - def _get_free_resources(self, page_link: str, ctx_cookies: List[dict], ctx: Chrome): + def _get_free_resource(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( diff --git a/src/services/bricklayer/unreal.py b/src/services/bricklayer/unreal.py index 26c0c7187d..2c46ef2942 100644 --- a/src/services/bricklayer/unreal.py +++ b/src/services/bricklayer/unreal.py @@ -3,7 +3,7 @@ # Author : QIN2DIM # Github : https://github.com/QIN2DIM # Description: -from typing import List, Optional, Dict +from typing import List, Optional, Dict, Union from bs4 import BeautifulSoup from cloudscraper import create_scraper @@ -31,7 +31,9 @@ class UnrealClaimer(Bricklayer): def __init__(self, silence: Optional[bool] = None): super().__init__(silence=silence, auth_str="unreal") - def get_claimer_response(self, ctx_cookies: List[dict]) -> List[Dict[str, str]]: + def get_claimer_response( + self, ctx_cookies: List[dict] + ) -> List[Dict[str, Union[str, bool]]]: """领取任务后审查资源的在库状态""" headers = {"cookie": ToolBox.transfer_cookies(ctx_cookies)} scraper = create_scraper() @@ -51,9 +53,9 @@ def get_claimer_response(self, ctx_cookies: List[dict]) -> List[Dict[str, str]]: return details - def get_free_resource(self, ctx, ctx_cookies): + def get_free_unreal_content(self, ctx_session, ctx_cookies): try: - self._unreal_get_free_resource(ctx=ctx, ctx_cookies=ctx_cookies) + self._unreal_get_free_resource(ctx=ctx_session, ctx_cookies=ctx_cookies) except AssertTimeout: logger.debug( ToolBox.runtime_report( diff --git a/src/services/deploy.py b/src/services/deploy.py index 125db620d1..aed1dcc8bc 100644 --- a/src/services/deploy.py +++ b/src/services/deploy.py @@ -5,7 +5,7 @@ # Description: import random from datetime import datetime, timedelta -from typing import Optional, List, Dict, Any +from typing import Optional, List, Dict, Union import apprise import pytz @@ -135,18 +135,20 @@ def __init__(self, silence: bool, log_ignore: Optional[bool] = False, _auth_str= auth_str = "games" if _auth_str is None else _auth_str self.bricklayer = Bricklayer(silence=silence, auth_str=auth_str) self.explorer = Explorer(silence=silence) + self._ctx_session = None # 任务队列 按顺缓存周免游戏及其免费附加内容的认领任务 - self.task_queue = Queue() + self.task_queue_pending = Queue() + self.task_queue_worker = Queue() # 消息队列 按序缓存认领任务的执行状态 self.message_queue = Queue() # 内联数据容器 编排推送模版 self.inline_docker = [] def __enter__(self): - # 集成统一的驱动上下文，减少内存占用 - self.challenger = get_challenge_ctx(silence=self.silence) - + if self.bricklayer.cookie_manager.refresh_ctx_cookies(_keep_live=True): + self._ctx_session = self.bricklayer.cookie_manager.ctx_session + self._ctx_cookies = self.bricklayer.cookie_manager.load_ctx_cookies() return self def __exit__(self, exc_type, exc_val, exc_tb): @@ -154,8 +156,15 @@ def __exit__(self, exc_type, exc_val, exc_tb): self._pusher_wrapper() # 缓存卸载 - if hasattr(self, "challenger"): - self.challenger.quit() + try: + if hasattr(self, "_ctx_session"): + self._ctx_session.quit() + except AttributeError: + pass + + def _pusher_putter(self, result: str, obj: Dict[str, Union[bool, str]]): + _runtime = {"status": result, "name": obj["name"], "dlc": obj.get("dlc", False)} + self.message_queue.put_nowait(_runtime) def _pusher_wrapper(self): while not self.message_queue.empty(): @@ -239,9 +248,17 @@ def _push(self, inline_docker: list, pusher_settings: Optional[dict] = None): ) ) - def promotions_filter( - self, promotions: Optional[Dict[str, Any]], ctx_cookies: List[dict] - ) -> Optional[List[Dict[str, Any]]]: + def get_promotions(self) -> Optional[Dict[str, Union[List[str], str]]]: + """获取促销信息的顶级接口""" + try: + return self.explorer.get_promotions(ctx_cookies=self._ctx_cookies) + except Exception as err: # noqa + self.logger.exception(err) + return self.explorer.get_promotions_by_stress_expressions( + _ctx_session=self._ctx_session + ) + + def promotions_filter(self): """ 促销实体过滤器 @@ -249,12 +266,10 @@ def promotions_filter( 2. 判断是否存在免费附加内容 3. 识别并弹出已在库资源 4. 返回待认领的实体资源 - :param promotions: - :param ctx_cookies: :return: """ - def in_library(page_link: str, name: str): + def in_library(page_link: str, name: str) -> bool: response = self.explorer.game_manager.is_my_game( ctx_cookies=ctx_cookies, page_link=page_link ) @@ -279,61 +294,63 @@ def in_library(page_link: str, name: str): ) return True + promotions = self.get_promotions() if not isinstance(promotions, dict) or not promotions["urls"]: return promotions + ctx_cookies = self._ctx_cookies # 过滤资源实体 - pending_objs = [] for url in promotions["urls"]: # 标记已在库游戏本体 job_name = promotions[url] - pending_objs.append( + self.task_queue_pending.put( {"url": url, "name": job_name, "in_library": in_library(url, job_name)} ) - # 识别免费附加内容 dlc_details = self.bricklayer.get_free_dlc_details( ctx_url=url, ctx_cookies=ctx_cookies ) - # 标记已在库的免费附加内容 for dlc in dlc_details: dlc.update({"in_library": in_library(dlc["url"], dlc["name"])}) - pending_objs.append(dlc) - - return pending_objs + self.task_queue_pending.put(dlc) def just_do_it(self): - """单步子任务 认领周免游戏""" - # 检查并更新身份令牌 - if self.bricklayer.cookie_manager.refresh_ctx_cookies( - _ctx_session=self.challenger - ): - # 读取有效的身份令牌 - ctx_cookies = self.bricklayer.cookie_manager.load_ctx_cookies() - - # 扫描商城促销活动，返回“0折”商品的名称与商城链接 - promotions = self.explorer.get_promotions(ctx_cookies) - - # 资源聚合过滤 从顶级接口剔除已在库资源 - game_objs = self.promotions_filter(promotions, 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) + """认领周免游戏及其免费附加内容""" + # ====================================== + # [🚀] 你以为是武器吧？但是居然是讯息…… + # ====================================== + # 1. 获取资源<本周免费> + # 2. 剔除资源<已在库中> + # ====================================== + self.promotions_filter() + + while not self.task_queue_pending.empty(): + game_obj = self.task_queue_pending.get() + if game_obj["in_library"]: + result = self.bricklayer.assert_.GAME_OK + self._pusher_putter(result=result, obj=game_obj) + else: + self.task_queue_worker.put(game_obj) + + # ====================================== + # [🚀] 前有重要道具！但是人机挑战…… + # ====================================== + # 1. 启动消息队列 编排消息模版 + # 2. 启动任务队列 领取周免游戏 + # ====================================== + if self.task_queue_worker.empty(): + return + if self._ctx_session is None: + self._ctx_session = get_challenge_ctx(self.silence) + while not self.task_queue_worker.empty(): + job = self.task_queue_worker.get() + result = self.bricklayer.get_free_games( + page_link=job["url"], + ctx_cookies=self._ctx_cookies, + ctx_session=self._ctx_session, + ) + self._pusher_putter(result=result, obj=job) class UnrealClaimerInstance(ClaimerInstance): @@ -343,22 +360,69 @@ def __init__(self, silence: bool, log_ignore: Optional[bool] = False): super().__init__(silence=silence, log_ignore=log_ignore) self.bricklayer = UnrealClaimer(silence=silence) + self.depth = 0 + + def promotions_filter(self): + def in_library(name: str, status: str) -> bool: + # 资源待认领 + if status == self.bricklayer.assert_.GAME_PENDING: + self.logger.debug( + ToolBox.runtime_report( + motive="STARTUP", + action_name="ScaffoldClaim", + message="🍜 正在为玩家领取周免游戏", + game=f"『{name}』", + ) + ) + return False + self.logger.info( + ToolBox.runtime_report( + motive="GET", + action_name=self.action_name, + message="🛴 资源已在库", + game=f"『{name}』", + ) + ) + return True + + content_objs = self.bricklayer.get_claimer_response(self._ctx_cookies) + for content_obj in content_objs: + content_obj.update( + {"in_library": in_library(content_obj["name"], content_obj["status"])} + ) + self.task_queue_pending.put(content_obj) def just_do_it(self): """虚幻商城月供砖家""" - # 检查并更新身份令牌 - if self.bricklayer.cookie_manager.refresh_ctx_cookies( - _ctx_session=self.challenger - ): - # 读取有效的身份令牌 - ctx_cookies = self.bricklayer.cookie_manager.load_ctx_cookies() - - # 释放 Claimer 认领免费内容 - self.bricklayer.get_free_resource( - ctx=self.challenger, ctx_cookies=ctx_cookies - ) + # ====================================== + # [🚀] 你以为是武器吧？但是居然是讯息…… + # ====================================== + # 1. 获取资源<本周免费> + # 2. 剔除资源<已在库中> + # ====================================== + self.promotions_filter() + + while not self.task_queue_pending.empty(): + content_obj = self.task_queue_pending.get() + if content_obj["in_library"]: + self._pusher_putter(result=content_obj["status"], obj=content_obj) + else: + self.task_queue_worker.put(content_obj) + + # ====================================== + # [🚀] 前有重要道具！但是人机挑战…… + # ====================================== + # 1. 启动消息队列 编排消息模版 + # 2. 启动任务队列 领取周免游戏 + # ====================================== + if self.task_queue_worker.empty() or self.depth >= 2: + return + if self._ctx_session is None: + self._ctx_session = get_challenge_ctx(self.silence) + self.bricklayer.get_free_unreal_content( + ctx_session=self._ctx_session, ctx_cookies=self._ctx_cookies + ) - # 检查运行结果 - details = self.bricklayer.get_claimer_response(ctx_cookies) - for detail in details: - self.message_queue.put_nowait(detail) + # [🛵] 接下来，跳跃很有用 + self.depth += 1 + return self.just_do_it() diff --git a/src/services/explorer/explorer.py b/src/services/explorer/explorer.py index 6fa3e096fb..dfce9f9915 100644 --- a/src/services/explorer/explorer.py +++ b/src/services/explorer/explorer.py @@ -4,7 +4,7 @@ # Github : https://github.com/QIN2DIM # Description: import json.decoder -from typing import List, Optional, Union, Dict, Any +from typing import List, Optional, Union, Dict import cloudscraper import yaml @@ -192,14 +192,14 @@ def discovery_free_games( # 返回链接 return [game_obj.get("url") for game_obj in game_objs] - def get_promotions(self, ctx_cookies: List[dict]) -> Dict[str, Any]: + def get_promotions(self, ctx_cookies: List[dict]) -> Dict[str, Union[List[str], str]]: """ 获取周免游戏数据 <即将推出> promotion["promotions"]["upcomingPromotionalOffers"] <本周免费> promotion["promotions"]["promotionalOffers"] :param ctx_cookies: - :return: + :return: {"urls": [], "pageLink1": "pageTitle1", "pageLink2": "pageTitle2", ...} """ free_game_objs = {"urls": []} headers = { @@ -230,3 +230,25 @@ def get_promotions(self, ctx_cookies: List[dict]) -> Dict[str, Any]: free_game_objs[url] = promotion["title"] return free_game_objs + + def get_promotions_by_stress_expressions( + self, _ctx_session=None + ) -> Dict[str, Union[List[str], str]]: + """使用应力表达式萃取商品链接""" + free_game_objs = {"urls": []} + if _ctx_session: + critical_memory = _ctx_session.current_window_handle + try: + _ctx_session.switch_to.new_window("tab") + pending_games: Dict[str, str] = self.stress_expressions(ctx=_ctx_session) + finally: + _ctx_session.switch_to.window(critical_memory) + else: + with get_ctx(silence=self.silence) as ctx: + pending_games: Dict[str, str] = self.stress_expressions(ctx=ctx) + + if pending_games: + for url, title in pending_games.items(): + free_game_objs[url] = title + free_game_objs["urls"].append(url) + return free_game_objs diff --git a/src/services/settings.py b/src/services/settings.py index 8245428a2b..09058a7f4e 100644 --- a/src/services/settings.py +++ b/src/services/settings.py @@ -6,9 +6,10 @@ import os import random import sys +from datetime import datetime from os.path import join, dirname from typing import Dict, Any, Optional -from datetime import datetime + from services.utils import ToolBox __all__ = [ @@ -127,10 +128,8 @@ sys.exit() _CONVERTER = ["沫雯喂", "辰丽", "荪彦孜", "有坷唯", "郑姊祺", "弹蓶蓶", "王飛"] -try: - if not PLAYER: - PLAYER = os.environ["PLAYER"] -except KeyError: +PLAYER = os.getenv("PLAYER", "") if not PLAYER else PLAYER +if PLAYER in ["", None]: PLAYER = f"{random.choice(_CONVERTER)}({datetime.now().day})" # -------------------------------- # [√] 阻止缺省配置