From 58b77e30abc714a5d04e7467281a9e1c17d6d76a Mon Sep 17 00:00:00 2001 From: AquamarineCyan <1057424730@qq.com> Date: Thu, 30 Nov 2023 22:50:05 +0800 Subject: [PATCH] =?UTF-8?q?:recycle:=20=E5=87=8F=E5=B0=91=E5=9B=BE?= =?UTF-8?q?=E5=83=8F=E8=BD=AE=E8=AF=A2=E4=BA=A7=E7=94=9F=E7=9A=84=E6=97=A5?= =?UTF-8?q?=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.MD | 1 + src/package/baiguiyexing.py | 126 +++++++---------- src/package/daoguantupo.py | 269 ++++++++++++++++++----------------- src/package/huodong.py | 7 +- src/package/jiejietupo.py | 270 +++++++++++++++++------------------- src/package/qiling.py | 39 +++--- src/package/rilun.py | 89 ++++++------ src/package/utils.py | 58 +++++++- src/package/yeyuanhuo.py | 81 +++++------ src/utils/coordinate.py | 2 + src/utils/function.py | 47 +++---- src/utils/gui.py | 16 +-- 12 files changed, 496 insertions(+), 509 deletions(-) diff --git a/CHANGELOG.MD b/CHANGELOG.MD index fcfa9e6..736c5d1 100644 --- a/CHANGELOG.MD +++ b/CHANGELOG.MD @@ -5,6 +5,7 @@ ### 优化 - 移除更新升级过程中的`任意键` +- 减少图像轮询产生的日志 ## v1.7.10 - 2023-11-14 diff --git a/src/package/baiguiyexing.py b/src/package/baiguiyexing.py index 010d052..f3e1707 100644 --- a/src/package/baiguiyexing.py +++ b/src/package/baiguiyexing.py @@ -3,18 +3,13 @@ # baiguiyexing.py """百鬼夜行""" -import random -import time - -import pyautogui - -from ..utils.decorator import log_function_call, run_in_thread, time_count +from ..utils.coordinate import RelativeCoor +from ..utils.decorator import log_function_call from ..utils.event import event_thread from ..utils.function import ( - check_click, - check_scene, - get_coor_info, + click, random_coor, + random_num, random_sleep ) from ..utils.log import logger @@ -24,37 +19,37 @@ class BaiGuiYeXing(Package): """百鬼夜行""" + scene_name = "百鬼夜行" + resource_path = "baiguiyexing" + resource_list = [ + "title", # 标题 + "jinru", # 进入 + "ya", # 押选 + "kaishi", # 开始 + "baiguiqiyueshu", # 百鬼契约书 + ] @log_function_call def __init__(self, n: int = 0) -> None: super().__init__(n) - self.scene_name: str = "百鬼夜行" - self.n: int = 0 # 当前次数 - self.max: int = n # 总次数 - self.resource_path = "baiguiyexing" # 资源路径 - self.resource_list: list = [ # 资源列表 - "title", # 标题 - "jinru", # 进入 - "ya", # 押选 - "kaishi", # 开始 - "baiguiqiyueshu" # 百鬼契约书 - ] - def title(self) -> bool: + def check_title(self) -> bool: """场景""" - flag_title = True # 场景提示 + _flag_title_msg = True while True: if event_thread.is_set(): return - if check_scene(f"{self.resource_path}/title", self.scene_name): - return True - elif flag_title: - flag_title = False + coor = self.get_coor_info("title") + if coor.is_effective: + logger.scene(self.scene_name) + return + elif _flag_title_msg: + _flag_title_msg = False logger.ui("请检查游戏场景", "warn") def start(self): """开始""" - check_click(f"{self.resource_path}/jinru") + self.check_click("jinru") def choose(self): """鬼王选择""" @@ -69,9 +64,7 @@ def choose(self): while True: if event_thread.is_set(): return - # 获取系统当前时间戳 - random.seed(time.time_ns()) - m = random.random() * 3 + 1 + m = random_num(1, 4) if m < 2: x1 = _x1_left x2 = _x1_right @@ -82,22 +75,17 @@ def choose(self): x1 = _x3_left x2 = _x3_right x, y = random_coor(x1, x2, _y1, _y2).coor - pyautogui.moveTo( - x + window.window_left, - y + window.window_top, - duration=0.5 - ) - pyautogui.click() - time.sleep(2) - x, y = get_coor_info(f"{self.resource_path}/ya").coor - if x != 0 and y != 0: + click(RelativeCoor(x, y)) + random_sleep() + coor = self.get_coor_info("ya") + if coor.is_effective: logger.ui("已选择鬼王") break - check_click(f"{self.resource_path}/kaishi", dura=0.5) + self.check_click("kaishi") def fighting(self): """砸豆子""" - time.sleep(2) + random_sleep() for _ in range(250, 0, -5): if event_thread.is_set(): return @@ -108,48 +96,38 @@ def fighting(self): 300, window.absolute_window_height - 100 ).coor - pyautogui.moveTo( - x + window.window_left, - y + window.window_top, - duration=0.25 - ) - pyautogui.click() + click(RelativeCoor(x, y), dura=0.25) def finish(self): """结束""" while True: if event_thread.is_set(): return - coor = get_coor_info(f'{self.resource_path}/baiguiqiyueshu') - time.sleep(2) + coor = self.get_coor_info("baiguiqiyueshu") + random_sleep() if coor.is_effective: self.screenshot() - pyautogui.moveTo(coor.x, coor.y, duration=0.5) - pyautogui.click() + click(coor) break - @run_in_thread - @time_count - @log_function_call def run(self): - if self.title(): - logger.num(f"0/{self.max}") + self.check_title() + logger.num(f"0/{self.max}") + random_sleep(1, 3) + + while self.n < self.max: + if event_thread.is_set(): + return + random_sleep(0, 2) + self.start() random_sleep(1, 3) - while self.n < self.max: - if event_thread.is_set(): - return - random_sleep(0, 2) - self.start() - random_sleep(1, 3) - self.choose() - random_sleep(2, 4) - self.fighting() - random_sleep(2, 4) - self.finish() - self.n += 1 - logger.num(f"{self.n}/{self.max}") - time.sleep(4) - # TODO 更新随机判断 - if self.n in {12, 25, 39}: - random_sleep(10, 20) - logger.ui(f"已完成 {self.scene_name} {self.n}次") + self.choose() + random_sleep(2, 4) + self.fighting() + random_sleep(2, 4) + self.finish() + self.done() + random_sleep(3) + # TODO 更新随机判断 + if self.n in {12, 25, 39}: + random_sleep(10, 20) diff --git a/src/package/daoguantupo.py b/src/package/daoguantupo.py index c53d8e1..2610244 100644 --- a/src/package/daoguantupo.py +++ b/src/package/daoguantupo.py @@ -3,12 +3,10 @@ # daoguantupo.py """道馆突破""" -from ..utils.decorator import log_function_call, run_in_thread, time_count +from ..utils.decorator import log_function_call from ..utils.event import event_thread from ..utils.function import ( RESOURCE_FIGHT_PATH, - check_click, - check_scene, check_scene_multiple_once, click, finish_random_left_right, @@ -21,83 +19,79 @@ class DaoGuanTuPo(Package): """道馆突破""" + scene_name = "道馆突破" + resource_path = "daoguantupo" + resource_list= [ + "button_zhuwei", # TODO 助威开关 + "chuzhan", # 出战 + "daojishi", # 倒计时 + "guanzhan", # 观战 + "guanzhuzhan", # TODO 馆主战 + "jijie", # 集结 + "qianwang", # 前往-助威 + "shengyutuposhijian", # 剩余突破时间 + "tiaozhan", # 挑战 + "title", # 标题 + "zhanbao", # 战报 + ] @log_function_call def __init__(self, flag_guanzhan: bool = False) -> None: - self.scene_name = "道馆突破" - self.n = 0 # 当前次数 - self.resource_path = "daoguantupo" # 路径 - self.resource_list: list = [ - "button_zhuwei", # TODO 助威开关 - "chuzhan", # 出战 - "daojishi", # 倒计时 - "guanzhan", # 观战 - "guanzhuzhan", # TODO 馆主战 - "jijie", # 集结 - "qianwang", # 前往-助威 - "shengyutuposhijian", # 剩余突破时间 - "tiaozhan", # 挑战 - "title", # 标题 - "zhanbao" # 战报 - ] - + super().__init__() self.flag_guanzhan = flag_guanzhan # 是否观战 self.flag_fighting = False # 是否进行中 - def get_coor_info(self, file: str): - return get_coor_info(f"{self.resource_path}/{file}") - - def title(self) -> bool: + def check_title(self) -> None: """场景""" - flag_title = True # 场景提示 + _flag_title_msg = True self.flag_fighting = False # 进行中 - flag_daojishi = True # 倒计时 + _flag_daojishi = True # 倒计时 while True: if event_thread.is_set(): return - if check_scene(f"{self.resource_path}/title", self.scene_name): + coor_title = self.get_coor_info("title") + coor_fighting = self.get_coor_info("button_zhuwei") + + if coor_title.is_effective: + logger.scene(self.scene_name) + self.current_resource_list = [ + f"{self.resource_path}/daojishi", + f"{self.resource_path}/shengyutuposhijian", + f"{self.resource_path}/guanzhuzhan", + f"{self.resource_path}/button_zhuwei", + ] + self.log_current_scene_list() while True: if event_thread.is_set(): return - # 等待倒计时自动进入 - if self.judge_scene_daoguantupo() == "倒计时": - if flag_daojishi: - logger.ui("等待倒计时自动进入") - flag_daojishi = False - self.flag_fighting = True - break - elif self.judge_scene_daoguantupo() == "可进攻": - self.flag_fighting = False - break - # 馆主战 - elif self.judge_scene_daoguantupo() == "馆主战": - logger.ui("待开发", "warn") - break - return True - # 已进入道馆进攻状态 - elif self.judge_scene_daoguantupo() == "进行中": + scene, _ = check_scene_multiple_once(self.current_resource_list) + if scene is None: + continue + scene = self.scene_handle(scene) + + match scene: + case "daojishi": # 等待倒计时自动进入 + if _flag_daojishi: + logger.ui("等待倒计时自动进入") + _flag_daojishi = False + self.flag_fighting = True + return + case "shengyutuposhijian": # 可进攻 + self.flag_fighting = False + return + case "guanzhuzhan": # 馆主战 + logger.ui("馆主战，待开发", "warn") + return + + # 道馆突破进行中 + elif coor_fighting.is_effective: logger.ui("道馆突破进行中") self.flag_fighting = True return True - elif flag_title: - flag_title = False + elif _flag_title_msg: + _flag_title_msg = False logger.ui("请检查游戏场景", "warn") - def judge_scene_daoguantupo(self) -> str: - """场景判断""" - scene = { - "daojishi.png": "倒计时", - "shengyutuposhijian.png": "可进攻", - "guanzhuzhan.png": "馆主战", - "button_zhuwei.png": "进行中" - } # TODO"可进攻"未实现 - for item in scene: - if event_thread.is_set(): - return - coor = self.get_coor_info(item) - if coor.is_effective: - return scene[item] - def guanzhan(self): """观战""" logger.ui("观战中，暂无法自动退出，可手动退出", "warn") @@ -108,8 +102,8 @@ def guanzhan(self): coor = self.get_coor_info("qianwang") if coor.is_effective: break - check_click(f"{self.resource_path}/zhanbao") - random_sleep(1, 2) + self.check_click("zhanbao") + random_sleep() # 前往按钮 while True: if event_thread.is_set(): @@ -117,92 +111,97 @@ def guanzhan(self): coor = self.get_coor_info("jijie") if coor.is_effective: break - check_click(f"{self.resource_path}/qianwang") - random_sleep(1, 2) - flag_zhuwei = False # 是否能够助威 + self.check_click("qianwang") + random_sleep() + + self.current_resource_list = [ + f"{self.resource_path}/zhuwei", + f"{self.resource_path}/test_zhuwei_gray", + f"{RESOURCE_FIGHT_PATH}/finish", + f"{RESOURCE_FIGHT_PATH}/fail", + ] + _flag_zhuwei_disable = False # 是否能够助威 + while True: if event_thread.is_set(): return - coor1 = self.get_coor_info("zhuwei") - coor2 = self.get_coor_info("test_zhuwei_gray") - # 可助威 - if coor1.is_effective: - click(coor) - logger.ui("助威成功") - flag_zhuwei = True - # 不可助威 - elif coor2.is_effective: - if flag_zhuwei: - logger.ui("无法助威") - flag_zhuwei = False - # 结束观战 - else: - coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/finish") - if coor.is_effective: - logger.ui("胜利") - finish_random_left_right() - break - coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/fail") - if coor.is_effective: + scene, coor = self.check_scene_multiple_once() + if scene is None: + continue + logger.info(f"current scene: {scene}") + if "/" in scene: + scene = scene.split("/")[-1] + + match scene: + case "zhuwei": + click(coor) + logger.ui("助威成功") + _flag_zhuwei_disable = True + case "test_zhuwei_gray": + if _flag_zhuwei_disable: + logger.ui("无法助威") + _flag_zhuwei_disable = False + case "finish": + self.ensure_finish() + case "fail": logger.ui("失败", "warn") + random_sleep(0.4, 0.8) finish_random_left_right() - break - random_sleep(1, 2) + while True: + if event_thread.is_set(): + return + coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/fail") + # 未重复检测到，表示成功点击 + if coor.is_zero: + self.done() + break + click() + random_sleep(0.4, 0.8) + random_sleep(4) - # TODO 馆主战 - def guanzhuzhan(self) -> None: + def guanzhuzhan(self) -> None: # TODO """馆主战""" pass - @run_in_thread - @time_count - @log_function_call - def run(self) -> None: - if self.title(): - logger.num(0) - random_sleep(2, 4) - if not self.flag_fighting: - check_click(f"{self.resource_path}/tiaozhan") - random_sleep(2, 4) - # TODO调整预设队伍 - # 开始 - while True: - if event_thread.is_set(): - return - - _resource_list = [ - f"{RESOURCE_FIGHT_PATH}/ready_old", - f"{RESOURCE_FIGHT_PATH}/ready_new", - f"{RESOURCE_FIGHT_PATH}/victory", - f"{RESOURCE_FIGHT_PATH}/fail", - f"{RESOURCE_FIGHT_PATH}/finish" - ] - scene, coor = check_scene_multiple_once(_resource_list) - if scene is None: - continue - - logger.info(f"scene: {scene}") - logger.info(f"coor: {coor.coor}") - if scene in [ - f"{RESOURCE_FIGHT_PATH}/ready_old", - f"{RESOURCE_FIGHT_PATH}/ready_new", - ]: + def run(self): + self.check_title() + logger.num(0) + random_sleep(2, 4) + if not self.flag_fighting: + self.check_click("tiaozhan") + random_sleep(2, 4) + # TODO调整预设队伍 + # 开始 + while True: + if event_thread.is_set(): + return + self.current_resource_list = [ + f"{RESOURCE_FIGHT_PATH}/ready_old", + f"{RESOURCE_FIGHT_PATH}/ready_new", + f"{RESOURCE_FIGHT_PATH}/victory", + f"{RESOURCE_FIGHT_PATH}/fail", + f"{RESOURCE_FIGHT_PATH}/finish" + ] + scene, coor = self.check_scene_multiple_once() + if scene is None: + continue + logger.info(f"coor: {coor.coor}") + scene = self.scene_handle(scene) + + match scene: + case "ready_old" | "read_new": logger.ui("准备") click(coor) self.n += 1 logger.num(str(self.n)) random_sleep(1, 2) - elif scene == f"{RESOURCE_FIGHT_PATH}/victory": + case "victory": random_sleep(1, 2) - elif scene == f"{RESOURCE_FIGHT_PATH}/fail": - logger.ui("失败", "warn") - finish_random_left_right() - break - elif scene == f"{RESOURCE_FIGHT_PATH}/finish": - logger.ui("结束") - finish_random_left_right() + case "finish": + self.ensure_finish() + case "fail": + logger.ui("失败，需要手动处理", "warn") break - if self.flag_guanzhan: - self.guanzhan() - logger.ui(f"已完成 {self.scene_name} 胜利{self.n}次") + if self.flag_guanzhan: + self.guanzhan() diff --git a/src/package/huodong.py b/src/package/huodong.py index 4344905..89743db 100644 --- a/src/package/huodong.py +++ b/src/package/huodong.py @@ -59,9 +59,6 @@ def timer_check_start(self): if coor.is_effective: self._flag_timer_check_start = True - # @run_in_thread - # @time_count - # @log_function_call def run(self) -> None: _g_resource_list: list = [ f"{self.resource_path}/title", @@ -80,9 +77,7 @@ def run(self) -> None: if scene is None: continue - logger.info(f"scene:{scene}") - if "/" in scene: - scene = scene.split("/")[-1] + scene = self.scene_handle(scene) if self._flag_timer_check_start: self._flag_timer_check_start = False diff --git a/src/package/jiejietupo.py b/src/package/jiejietupo.py index f0f0fec..ad3e4d5 100644 --- a/src/package/jiejietupo.py +++ b/src/package/jiejietupo.py @@ -3,15 +3,13 @@ # jiejietupo.py """结界突破""" -import math -import time from pathlib import Path import pyautogui from ..utils.application import RESOURCE_DIR_PATH from ..utils.coordinate import Coor, RelativeCoor -from ..utils.decorator import log_function_call, run_in_thread, time_count +from ..utils.decorator import log_function_call from ..utils.event import event_thread from ..utils.function import ( RESOURCE_FIGHT_PATH, @@ -27,33 +25,32 @@ ) from ..utils.log import logger from ..utils.window import window +from .utils import Package -class JieJieTuPo: +class JieJieTuPo(Package): """结界突破""" - - def __init__(self): - self.scene_name: str = "结界突破" - self.resource_path: str = "jiejietupo" # 路径 - self.resource_list: list = [ - "fail", # 失败 - "fangshoujilu", # 防守记录-个人突破 - "fighting_fail", # TODO 战斗失败 - "geren", # 个人突破 - "jingong", # 进攻 - "queding", - "shuaxin", # 刷新-个人突破 - "title", # 突破界面 - "tupojilu", # 突破记录-阴阳寮突破 - "victory", # 攻破 - "xunzhang_0", # 勋章数0 - "xunzhang_1", # 勋章数1 - "xunzhang_2", # 勋章数2 - "xunzhang_3", # 勋章数3 - "xunzhang_4", # 勋章数4 - "xunzhang_5", # 勋章数5 - "yinyangliao" # 阴阳寮突破 - ] + scene_name: str = "结界突破" + resource_path: str = "jiejietupo" + resource_list: list = [ + "fail", # 失败 + "fangshoujilu", # 防守记录-个人突破 + "fighting_fail", # TODO 战斗失败 + "geren", # 个人突破 + "jingong", # 进攻 + "queding", + "shuaxin", # 刷新-个人突破 + "title", # 突破界面 + "tupojilu", # 突破记录-阴阳寮突破 + "victory", # 攻破 + "xunzhang_0", # 勋章数0 + "xunzhang_1", # 勋章数1 + "xunzhang_2", # 勋章数2 + "xunzhang_3", # 勋章数3 + "xunzhang_4", # 勋章数4 + "xunzhang_5", # 勋章数5 + "yinyangliao", # 阴阳寮突破 + ] def get_coor_info_tupo(self, x1: int, y1: int, file: Path | str) -> Coor: """图像识别，返回图像的局部相对坐标 @@ -67,7 +64,7 @@ def get_coor_info_tupo(self, x1: int, y1: int, file: Path | str) -> Coor: Coor: 坐标 """ _file_name = image_file_format(RESOURCE_DIR_PATH / self.resource_path / file) - logger.info(f"looking for file: {_file_name}") + logger.info(f"looking for file: {_file_name}") # TODO if "xunzhang" in file: # 个人突破 try: @@ -112,6 +109,33 @@ def get_coor_info_tupo(self, x1: int, y1: int, file: Path | str) -> Coor: coor = Coor(0, 0) return coor + def check_title(self) -> None: + """场景""" + _flag_title_msg = True + while True: + if event_thread.is_set(): + return + coor = self.get_coor_info("title") + if coor.is_effective: + logger.scene(JieJieTuPo.scene_name) + if isinstance(self, JieJieTuPoGeRen): + file_1 = "fangshoujilu" + file_2 = "geren" + if isinstance(self, JieJieTuPoYinYangLiao): + file_1 = "tupojilu" + file_2 = "yinyangliao" + while True: + coor = self.get_coor_info(file_1) + if coor.is_effective: + logger.scene(self.scene_name) + return + random_sleep(0.4, 0.8) + self.check_click(file_2) + random_sleep() + elif _flag_title_msg: + _flag_title_msg = False + logger.ui("请检查游戏场景", "warn") + def fighting_tupo(self, x0: int, y0: int) -> None: """结界突破战斗 @@ -125,10 +149,10 @@ def fighting_tupo(self, x0: int, y0: int) -> None: while True: if event_thread.is_set(): return - coor = get_coor_info(f"{self.resource_path}/jingong") + coor = self.get_coor_info("jingong") if coor.is_effective: click(coor) - break + return def fighting_fail_again(self): # TODO 主动失败 @@ -152,43 +176,25 @@ class JieJieTuPoGeRen(JieJieTuPo): 间隔宽115 间隔高30 """ + scene_name = "个人突破" + tupo_geren_x = { + 1: 215, + 2: 515, + 3: 815, + } + tupo_geren_y = { + 1: 175, + 2: 295, + 3: 415 + } @log_function_call def __init__(self, n: int = 0) -> None: - super().__init__() - self.tupo_geren_x = { - 1: 215, - 2: 515, - 3: 815, - } - self.tupo_geren_y = { - 1: 175, - 2: 295, - 3: 415 - } - self.n: int = 0 # 当前次数 - self.max: int = n # 总次数 + super().__init__(n) self.list_xunzhang: list = None # 勋章列表 self.tupo_victory: int = None # 攻破次数 self.time_refresh: int = 0 # 记录刷新时间 - def title(self) -> bool: - """场景""" - flag_title = True # 场景提示 - while True: - if event_thread.is_set(): - return - if check_scene(f"{self.resource_path}/title", "结界突破"): - while True: - if check_scene(f"{self.resource_path}/fangshoujilu.png", "个人突破"): - return True - random_sleep(0, 1) - check_click(f"{self.resource_path}/geren.png") - random_sleep(3, 4) - elif flag_title: - flag_title = False - logger.ui("请检查游戏场景", "warn") - def list_num_xunzhang(self) -> list[int]: """ 创建列表，返回每个结界的勋章数 @@ -299,32 +305,31 @@ def fighting(self) -> None: self.tupo_geren_y[(k + 2) // 3] ) # TODO 待优化，利用时间戳的间隔判断 - # time.sleep(4) + # random_sleep(4) # if self.judge_click("zhunbei.png",click=False): # self.judge_click("zhunbei.png") if finish(): flag_victory = True - self.n += 1 - logger.num(f"{self.n}/{self.max}") + self.done() else: flag_victory = False - random_sleep(0, 1) + random_sleep(0.4, 0.8) # 结束界面 coor = finish_random_left_right() - random_sleep(1, 2) + random_sleep() # 3胜奖励 if self.tupo_victory == 2 and flag_victory: - random_sleep(1, 2) + random_sleep() while True: if event_thread.is_set(): return check_click(f"{RESOURCE_FIGHT_PATH}/finish") - random_sleep(1, 2) + random_sleep() coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/finish") if coor.is_zero: break logger.ui("成功攻破3次") - random_sleep(2, 3) + random_sleep(2) if flag_victory: return @@ -332,6 +337,8 @@ def refresh(self) -> None: """刷新""" flag_refresh = False # 刷新提醒 random_sleep(4, 8) # 强制等待 + import time + import math while True: if event_thread.is_set(): return @@ -340,9 +347,9 @@ def refresh(self) -> None: if self.time_refresh == 0 or self.time_refresh + 5 * 60 < timenow: logger.ui("刷新中") random_sleep(3, 6) - check_click(f"{self.resource_path}/shuaxin", sleep_time=2) + self.check_click("shuaxin", sleep_time=2) random_sleep(2, 4) - check_click(f"{self.resource_path}/queding", sleep_time=0.5) + self.check_click("queding", sleep_time=0.5) self.time_refresh = timenow random_sleep(2, 6) break @@ -352,31 +359,26 @@ def refresh(self) -> None: flag_refresh = True random_sleep(time_wait, time_wait+5) - @run_in_thread - @time_count - @log_function_call - def run(self) -> None: - if self.title(): - logger.num(f"0/{self.max}") - while self.n < self.max: - if event_thread.is_set(): - return - random_sleep(2, 4) - self.list_xunzhang = self.list_num_xunzhang() - # 胜利次数 - self.tupo_victory = self.list_xunzhang.count(-1) - # 刷新 - if self.tupo_victory == 3: - self.refresh() - # 挑战 - elif self.tupo_victory < 3: - logger.ui(f"已攻破{self.tupo_victory}个") - self.fighting() - elif self.tupo_victory > 3: - logger.ui("暂不支持大于3个，请自行处理", "warn") - break - random_sleep(2, 3) - logger.ui(f"已完成 个人突破 {self.n}次") + def run(self): + self.check_title() + logger.num(f"0/{self.max}") + while self.n < self.max: + if event_thread.is_set(): + return + self.list_xunzhang = self.list_num_xunzhang() + # 胜利次数 + self.tupo_victory = self.list_xunzhang.count(-1) + # 刷新 + if self.tupo_victory == 3: + self.refresh() + # 挑战 + elif self.tupo_victory < 3: + logger.ui(f"已攻破{self.tupo_victory}个") + self.fighting() + elif self.tupo_victory > 3: + logger.ui("暂不支持大于3个，请自行处理", "warn") + break + random_sleep(2) class JieJieTuPoYinYangLiao(JieJieTuPo): @@ -387,39 +389,21 @@ class JieJieTuPoYinYangLiao(JieJieTuPo): 间隔宽115 间隔高40 """ + scene_name = "阴阳寮突破" + tupo_yinyangliao_x = { + 1: 460, + 2: 760 + } + tupo_yinyangliao_y = { + 1: 170, + 2: 290, + 3: 410, + 4: 530 + } @log_function_call def __init__(self, n: int = 0) -> None: - super().__init__() - self.tupo_yinyangliao_x = { - 1: 460, - 2: 760 - } - self.tupo_yinyangliao_y = { - 1: 170, - 2: 290, - 3: 410, - 4: 530 - } - self.n: int = 0 # 当前次数 - self.max: int = n # 总次数 - - def title(self) -> bool: - """场景""" - flag_title = True # 场景提示 - while True: - if event_thread.is_set(): - return - if check_scene(f"{self.resource_path}/title", "结界突破"): - while True: - if check_scene(f"{self.resource_path}/tupojilu", "阴阳寮突破"): - return True - random_sleep(0, 1) - check_click(f"{self.resource_path}/yinyangliao.png") - random_sleep(3, 4) - elif flag_title: - flag_title = False - logger.ui("请检查游戏场景", "warn") + super().__init__(n) def jibaicishu(self) -> bool: """剩余次数判断""" @@ -472,8 +456,8 @@ def fighting(self) -> int: ) # TODO 多人攻破同一寮突后，无法再次进入，通过加定时器5秒判断当前是否还是寮突界面提前退出战斗循环 # 延迟等待，判断当前寮突是否有效 - random_sleep(3, 4) - coor = get_coor_info(f"{self.resource_path}/jingong") + random_sleep(3) + coor = self.get_coor_info("jingong") if coor.is_effective: logger.ui("当前结界已被攻破", "warn") i += 1 @@ -481,7 +465,7 @@ def fighting(self) -> int: continue flag = 1 if finish() else 0 - random_sleep(1, 2) + random_sleep() # 结束界面 coor = finish_random_left_right() return flag @@ -506,21 +490,15 @@ def page_down(self, rows: int = 1): import pyautogui pyautogui.scroll(-(rows * 240)) # 2*pis(pis=2*120) - @run_in_thread - @time_count - @log_function_call - def run(self) -> None: - if self.title(): - logger.num(f"0/{self.max}") - random_sleep(1, 3) - while self.n < self.max: - if event_thread.is_set(): - break - random_sleep(0, 1) - if flag := self.fighting(): - self.n += 1 - logger.num(f"{self.n}/{self.max}") - elif flag == -1: - break - random_sleep(2, 3) - logger.ui(f"已完成 阴阳寮突破 {self.n}次") + def run(self): + self.check_title() + logger.num(f"0/{self.max}") + while self.n < self.max: + if event_thread.is_set(): + break + random_sleep(0.4, 0.8) + if flag := self.fighting(): + self.done() + elif flag == -1: + break + random_sleep() diff --git a/src/package/qiling.py b/src/package/qiling.py index 0bcb92c..1922029 100644 --- a/src/package/qiling.py +++ b/src/package/qiling.py @@ -3,10 +3,8 @@ # qiling.py """契灵""" -from threading import Timer - from ..utils.coordinate import RelativeCoor -from ..utils.decorator import log_function_call, run_in_thread, time_count +from ..utils.decorator import log_function_call from ..utils.event import event_thread from ..utils.function import ( RESOURCE_FIGHT_PATH, @@ -18,6 +16,7 @@ random_sleep ) from ..utils.log import logger +from ..utils.mythread import WorkTimer from .utils import Package @@ -60,7 +59,7 @@ def fighting(self): @log_function_call def timer_start(self): - coor = get_coor_info(f"{self.resource_path}/start_tancha") + coor = self.get_coor_info("start_tancha") if coor.is_effective: self._flag_finish = True @@ -79,8 +78,8 @@ def check_pokemon(self) -> bool: logger.info(f"_pokemon_address_count: {self._pokemon_address_count}") click(_pokemon_list[i]) self._pokemon_address_count += 1 - random_sleep(2, 3) - coor = get_coor_info(f"{self.resource_path}/start_jieqi") + random_sleep(2) + coor = self.get_coor_info("start_jieqi") if coor.is_effective: return True else: @@ -100,14 +99,13 @@ def run_tancha(self): if event_thread.is_set(): return scene, coor = check_scene_multiple_once(_resource_list, self.resource_path) - if scene is None: - continue - self.scene_print(scene) + scene = self.scene_handle(scene) + match scene: # case "title": # pass case "start_tancha": - Timer(5, self.timer_start).start() + WorkTimer(5, self.timer_start).start() # _timestamp = int(time.time()) # if _timestamp - self._timestamp < 3: # _flag_finish = True @@ -115,7 +113,7 @@ def run_tancha(self): # else: # self._timestamp = _timestamp click(coor) - random_sleep(1, 2) + random_sleep() self.fighting() self.done() random_sleep(2, 4) @@ -127,18 +125,16 @@ def run_tancha(self): def run_jieqi(self): """结契""" _n: int = 0 - _resource_list = ["start_tancha", "start_jieqi"] + self.current_resource_list = ["start_tancha", "start_jieqi"] _flag_done_once: bool = False while _n <= 5: if event_thread.is_set(): return - scene, coor = check_scene_multiple_once(_resource_list, self.resource_path) + scene, coor = self.check_scene_multiple_once() if scene is None: continue - logger.info(f"scene: {scene}") - if "/" in scene: - scene = scene.split("/")[-1] + scene = self.scene_handle(scene) match scene: # 确保在探查界面点击契灵小图标 @@ -150,14 +146,14 @@ def run_jieqi(self): logger.ui(f"结契第{_n}只成功") if not self.check_pokemon(): break - random_sleep(1, 2) + random_sleep() continue case "start_jieqi": logger.scene("契灵探查") click(coor) - random_sleep(10, 11) + random_sleep(10) _flag_first: bool = False - _timer = Timer(2 * 60, self.timer_jieqi_finish) + _timer = WorkTimer(2 * 60, self.timer_jieqi_finish) _timer.start() while True: @@ -172,12 +168,9 @@ def run_jieqi(self): finish_random_left_right() elif _flag_first: _flag_done_once = True - random_sleep(2, 3) + random_sleep(2) break - @run_in_thread - @time_count - @log_function_call def run(self): if self._flag_tancha: self.run_tancha() diff --git a/src/package/rilun.py b/src/package/rilun.py index dd54d81..9bfaa42 100644 --- a/src/package/rilun.py +++ b/src/package/rilun.py @@ -3,11 +3,11 @@ # rilun.py """组队日轮副本""" -import pyautogui -from src.utils.event import event_thread -from ..utils.decorator import log_function_call, run_in_thread, time_count +from ..utils.decorator import log_function_call +from ..utils.event import event_thread from ..utils.function import ( + RESOURCE_FIGHT_PATH, check_click, check_scene, click, @@ -16,46 +16,43 @@ get_coor_info, is_passengers_on_position, random_sleep, - result, - RESOURCE_FIGHT_PATH + result ) from ..utils.log import logger from ..utils.window import window +from .utils import Package - -class RiLun: +class RiLun(Package): """日轮副本""" + scene_name = "日轮副本" + resource_path = "rilun" + resource_yuhun_path: str = "yuhun" # 御魂路径，复用资源 + resource_list: list = [ + "fighting", # 对局进行中 + ] @log_function_call def __init__(self, n: int = 0, flag_driver: bool = False, flag_passengers: int = 2) -> None: - self.scene_name: str = "日轮副本" - self.n: int = 0 # 当前次数 - self.max: int = n # 总次数 - self.resource_path: str = "rilun" # 路径 - self.resource_yuhun_path: str = "yuhun" # 御魂路径，复用资源 - self.resource_list: list = [ # 资源列表 - "fighting" # 对局进行中 - ] - + super().__init__(n) self.flag_driver: bool = flag_driver # 是否为司机（默认否） self.flag_passengers: bool = flag_passengers # 组队人数 self.flag_driver_start: bool = False # 司机待机 self.flag_fighting: bool = False # 是否进行中对局（默认否） - def title(self) -> bool: + def check_title(self) -> bool: """场景""" - flag_title = True # 场景提示 + _flag_title_msg = True # 场景提示 while True: if event_thread.is_set(): return if check_scene(f"{self.resource_yuhun_path}/xiezhanduiwu", "组队御魂准备中"): self.flag_driver_start = True - return True + return elif check_scene(f"{self.resource_yuhun_path}/fighting", "组队御魂进行中"): self.flag_fighting = True - return True - elif flag_title: - flag_title = False + return + elif _flag_title_msg: + _flag_title_msg = False logger.ui("请检查游戏场景", "warn") def finish(self) -> None: @@ -63,6 +60,7 @@ def finish(self) -> None: result() random_sleep(1.5, 3) coor = finish_random_left_right(is_click=False) + import pyautogui while True: if event_thread.is_set(): return @@ -76,36 +74,31 @@ def finish(self) -> None: while True: if event_thread.is_set(): return - random_sleep(1, 2) + random_sleep() click() - random_sleep(1, 2) + random_sleep() coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/finish") if coor.is_zero: break break - random_sleep(0, 1) + random_sleep(0.4, 0.8) - @run_in_thread - @time_count - @log_function_call def run(self) -> None: - if self.title(): - logger.num(f"0/{self.max}") - while self.n < self.max: - if event_thread.is_set(): - return - # 司机 - if self.flag_driver and self.flag_driver_start: - is_passengers_on_position(self.flag_passengers) - # 开始挑战 - check_click(f"{RESOURCE_FIGHT_PATH}/start_team", dura=0.25) - logger.ui("开始") - if not self.flag_fighting: - check_click(f"{self.resource_path}/fighting", False) - self.flag_fighting = False - logger.ui("对局进行中") - self.finish() - self.n += 1 - logger.num(f"{self.n}/{self.max}") - random_sleep(1, 2) - logger.ui(f"已完成 组队日轮副本 {self.n}次") + self.check_title() + logger.num(f"0/{self.max}") + while self.n < self.max: + if event_thread.is_set(): + return + # 司机 + if self.flag_driver and self.flag_driver_start: + is_passengers_on_position(self.flag_passengers) + # 开始挑战 + check_click(f"{RESOURCE_FIGHT_PATH}/start_team", dura=0.25) + logger.ui("开始") + if not self.flag_fighting: + check_click(f"{self.resource_path}/fighting", False) + self.flag_fighting = False + logger.ui("对局进行中") + self.finish() + self.done() + random_sleep() diff --git a/src/package/utils.py b/src/package/utils.py index 5e489a3..447f865 100644 --- a/src/package/utils.py +++ b/src/package/utils.py @@ -1,7 +1,18 @@ import time from ..utils.decorator import run_in_thread -from ..utils.function import click, random_coor, screenshot +from ..utils.event import event_thread +from ..utils.function import ( + RESOURCE_FIGHT_PATH, + check_click, + check_scene_multiple_once, + click, + finish_random_left_right, + get_coor_info, + random_coor, + random_sleep, + screenshot +) from ..utils.log import logger from ..utils.mysignal import global_ms as ms from ..utils.toast import toast @@ -15,7 +26,7 @@ class Package: """路径""" resource_list: list = [] """资源列表""" - description : str = None + description: str = None """功能描述""" def __init__(self, n: int = 0) -> None: @@ -23,13 +34,38 @@ def __init__(self, n: int = 0) -> None: """当前次数""" self.max: int = n """总次数""" + self.current_resource_list: list = None + """当前使用的资源列表""" - def scene_print(self, scene: str = None) -> None: + def get_coor_info(self, file): + return get_coor_info(f"{self.resource_path}/{file}") + + def check_click(self, file, *args, **kwargs): + return check_click(f"{self.resource_path}/{file}", *args, **kwargs) + + def check_scene_multiple_once(self, *args, **kwargs): + return check_scene_multiple_once(self.current_resource_list, *args, **kwargs) + + def scene_print(self, scene: str = None) -> None: # FIXME remove """打印当前场景""" if "/" in scene: scene = scene.split("/")[-1] logger.scene(scene) + def scene_handle(self, scene: str = None) -> str: + logger.info(f"current scene: {scene}") + if "/" in scene: + scene = scene.split("/")[-1] + return scene + + def log_current_scene_list(self) -> None: + """记录当前匹配的资源列表""" + if self.current_resource_list is None: + return + logger.info(f"current_resource_list: {len(self.current_resource_list)}") + for item in self.current_resource_list: + logger.info(item) + def start(self, sleeptime: float = 0.4) -> None: """挑战开始""" coor = random_coor(1067 - 50, 1067 + 50, 602 - 50, 602 + 50) @@ -46,6 +82,22 @@ def done(self) -> None: self.n += 1 logger.num(f"{self.n}/{self.max}") + def ensure_finish(self): + """确保结束""" + logger.ui("结束") + random_sleep(0.4, 0.8) + finish_random_left_right() + while True: + if event_thread.is_set(): + return + coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/finish") + # 未重复检测到，表示成功点击 + if coor.is_zero: + self.done() + break + click() + random_sleep(0.4, 0.8) + def run(self): """任务内容""" pass diff --git a/src/package/yeyuanhuo.py b/src/package/yeyuanhuo.py index 8599d31..1505104 100644 --- a/src/package/yeyuanhuo.py +++ b/src/package/yeyuanhuo.py @@ -3,15 +3,13 @@ # yeyuanhuo.py """业原火副本""" -from src.utils.event import event_thread -from ..utils.decorator import log_function_call, run_in_thread, time_count +from ..utils.decorator import log_function_call +from ..utils.event import event_thread from ..utils.function import ( + RESOURCE_FIGHT_PATH, check_click, - check_finish_once, check_scene_multiple_once, - check_scene_multiple_while, click, - finish_random_left_right, random_sleep ) from ..utils.log import logger @@ -20,9 +18,9 @@ class YeYuanHuo(Package): """业原火副本""" - scene_name: str = "业原火副本" - resource_path: str = "yeyuanhuo" - resource_list: list = [ + scene_name = "业原火副本" + resource_path = "yeyuanhuo" + resource_list = [ "title", # 标题 "start" # 挑战 ] @@ -32,46 +30,49 @@ class YeYuanHuo(Package): def __init__(self, n: int = 0) -> None: super().__init__(n) - @log_function_call - def title(self) -> bool: - """场景""" - check_scene_multiple_while(self.resource_list, self.resource_path, text="请检查游戏场景") - return True - @log_function_call def start(self): """挑战开始""" check_click(f"{self.resource_path}/start") - @run_in_thread - @time_count - @log_function_call def run(self): + self.current_resource_list = [ + f"{self.resource_path}/title", + f"{self.resource_path}/start", + f"{RESOURCE_FIGHT_PATH}/finish", + f"{RESOURCE_FIGHT_PATH}/fail", + f"{RESOURCE_FIGHT_PATH}/victory", + ] + _flag_title_msg: bool = True logger.num(f"0/{self.max}") + self.log_current_scene_list() + while self.n < self.max: if event_thread.is_set(): return - - scene, coor = check_scene_multiple_once(self.resource_list, self.resource_path) - # if scene is None: - # continue - if scene: - self.scene_print(scene) - - if coor.is_effective: - match scene: - case "title": - self.start() - random_sleep(self.fast_time, self.fast_time+1) - case "start": - click(coor) - random_sleep(self.fast_time, self.fast_time+1) - if result := check_finish_once(): - self.done() - random_sleep(1, 2) - finish_random_left_right() - random_sleep(2, 4) + scene, coor = check_scene_multiple_once(self.current_resource_list) + if scene is None: continue - elif result == False: - logger.ui("失败，需要手动处理", "warn") - logger.ui(f"已完成 {self.scene_name} {self.n}次") + scene = self.scene_handle(scene) + + match scene: + case "title": + logger.scene(self.scene_name) + _flag_title_msg = False + self.start() + random_sleep(self.fast_time) + case "start": + click(coor) + random_sleep(self.fast_time) + case "fail": + logger.ui("失败，需要手动处理", "warn") + break + case "victory": + logger.ui("胜利") + random_sleep() + case "finish": + self.ensure_finish() + case _: + if _flag_title_msg: + logger.ui("请检查游戏场景", "warn") + _flag_title_msg = False diff --git a/src/utils/coordinate.py b/src/utils/coordinate.py index 1a1b069..be7507e 100644 --- a/src/utils/coordinate.py +++ b/src/utils/coordinate.py @@ -23,7 +23,9 @@ def __init__(self, x: int | float = 0, y: int | float = 0) -> None: self.y: int | float = y self.coor: tuple[int | float, int | float] = self._coor_tuple_format() self.is_zero: bool = self._is_zero_func() + """全0，未找到图像""" self.is_effective: bool = self._is_effective_func() + """坐标有效""" def _coor_tuple_format(self) -> tuple[int | float, int | float]: return (self.x, self.y) diff --git a/src/utils/function.py b/src/utils/function.py index 1ea040b..9e89f98 100644 --- a/src/utils/function.py +++ b/src/utils/function.py @@ -7,12 +7,10 @@ import time from datetime import datetime, timezone from pathlib import Path -from subprocess import Popen import pyautogui from .application import ( - APP_EXE_NAME, RESOURCE_DIR_PATH, RESOURCE_FIGHT_PATH, SCREENSHOT_DIR_PATH @@ -21,7 +19,6 @@ from .decorator import log_function_call from .event import event_thread, event_xuanshang, event_xuanshang_enable from .log import logger -from .mysignal import global_ms as ms from .window import window RESOURCE_FIGHT_PATH = RESOURCE_FIGHT_PATH @@ -32,25 +29,23 @@ class FightResource: """通用战斗资源""" - - def __init__(self): - self.resource_path: str = "fight" # 路径 - self.resource_list: list = [ # 资源列表 - "accept_invitation", # 接受邀请 - "fail", # 失败 - "finish", # 结束 - "fighting_friend_default", # 战斗中好友图标-怀旧/简约 - "fighting_friend_linshuanghanxue", # 战斗中好友图标-凛霜寒雪 - "fighting_friend_chunlvhanqing", # 战斗中好友图标-春缕含青 - "passenger_2", # 队员2 - "passenger_3", # 队员3 - "start_single", # 单人挑战 - "start_team", # 组队挑战 - "tanchigui", # 贪吃鬼 - "victory", # 成功 - "ready_old", # 准备-怀旧主题 - "ready_new", # 准备-简约主题 - ] + resource_path: str = "fight" # 路径 + resource_list: list = [ # 资源列表 + "accept_invitation", # 接受邀请 + "fail", # 失败 + "finish", # 结束 + "fighting_friend_default", # 战斗中好友图标-怀旧/简约 + "fighting_friend_linshuanghanxue", # 战斗中好友图标-凛霜寒雪 + "fighting_friend_chunlvhanqing", # 战斗中好友图标-春缕含青 + "passenger_2", # 队员2 + "passenger_3", # 队员3 + "start_single", # 单人挑战 + "start_team", # 组队挑战 + "tanchigui", # 贪吃鬼 + "victory", # 成功 + "ready_old", # 准备-怀旧主题 + "ready_new", # 准备-简约主题 + ] @log_function_call @@ -223,7 +218,7 @@ def get_coor_info_center(file: Path | str, is_log: bool = True) -> Coor: return Coor(0, 0) -def check_scene(file: str, scene_name: str = None) -> bool: +def check_scene(file: str, scene_name: str = None) -> bool: # FIXME remove """场景判断 参数: @@ -248,7 +243,7 @@ def check_scene_multiple_once(scene: list, resource_path: str = None) -> tuple[s """ 多场景判断，仅遍历一次 - 可传带RESOURCE_FIGHT_PATH资源， + 可传带`RESOURCE_FIGHT_PATH`资源 参数: scene (list): 多场景列表 @@ -262,7 +257,7 @@ def check_scene_multiple_once(scene: list, resource_path: str = None) -> tuple[s return None, Coor(0, 0) """ 1.如果没传路径，说明全部文件名自带路径 - 2.传参路径，可能存在RESOURCE_FIGHT_PAHT的资源，用斜杠判断列表值 + 2.传参路径，可能存在`RESOURCE_FIGHT_PAHT`的资源，用斜杠判断列表值 3.剩下的便是普通情况，即路径+文件 多数情况下会是第2种 """ @@ -339,7 +334,7 @@ def result() -> bool: if coor.is_effective: logger.ui("结束") return True - coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/fail") # TODO `fail` 需要更新素材 + coor = get_coor_info(f"{RESOURCE_FIGHT_PATH}/fail") # TODO `fail` 需要更新资源 if coor.is_effective: logger.ui("失败", "warn") return False diff --git a/src/utils/gui.py b/src/utils/gui.py index 024a298..bbf1efe 100644 --- a/src/utils/gui.py +++ b/src/utils/gui.py @@ -311,7 +311,7 @@ def __init__(self, n: int = 0) -> None: # 检查资源文件 for item in P.resource_list: if not Path(RESOURCE_DIR_PATH/P.resource_path/f"{item}.png").exists(): - logger.ui(f"无{P.resource_path}/{item}.png资源文件", "error") + logger.ui(f"未找到资源：{P.resource_path}/{item}.png", "error") ms.main.qmessagbox_update.emit("ERROR", f"无{P.resource_path}/{item}.png资源文件") return False logger.info("资源完整") @@ -525,20 +525,20 @@ def start() -> None: case "单人": pass case 3: # 业原火 - yeyuanhuo.YeYuanHuo(n=_n).run() + yeyuanhuo.YeYuanHuo(n=_n).task_start() case 4: # 御灵 yuling.YuLing(n=_n).run() case 5: # 个人突破 - jiejietupo.JieJieTuPoGeRen(n=_n).run() + jiejietupo.JieJieTuPoGeRen(n=_n).task_start() case 6: # 寮突破 - jiejietupo.JieJieTuPoYinYangLiao(n=_n).run() + jiejietupo.JieJieTuPoYinYangLiao(n=_n).task_start() case 7: # 道馆突破 flag_guanzhan = self.ui.button_guanzhan.isChecked() - daoguantupo.DaoGuanTuPo(flag_guanzhan=flag_guanzhan).run() + daoguantupo.DaoGuanTuPo(flag_guanzhan=flag_guanzhan).task_start() case 8: # 普通召唤 zhaohuan.ZhaoHuan(n=_n).run() case 9: # 百鬼夜行 - baiguiyexing.BaiGuiYeXing(n=_n).run() + baiguiyexing.BaiGuiYeXing(n=_n).task_start() case 10: # 限时活动 huodong.HuoDong(n=_n).task_start() case 11: # 组队日轮副本 @@ -553,7 +553,7 @@ def start() -> None: n=_n, flag_driver=_flag_driver, flag_passengers=_flag_passengers, - ).run() + ).task_start() case 12: # 单人探索 tansuo.TanSuo(n=_n).run() case 13: # 契灵 @@ -563,7 +563,7 @@ def start() -> None: n=_n, _flag_tancha=_flag_tancha, _flag_jieqi=_flag_jieqi - ).run() + ).task_start() def stop() -> None: """停止函数"""