"""商城英雄招募的限时 UP 神灯任务与免费神灯领取。""" import time from .engine import WorkflowError class RecruitRunner: def __init__(self, desktop, matcher, flow, region_provider, log=print, timeout=12): self.desktop = desktop self.matcher = matcher self.flow = flow self.region_provider = region_provider self.log = log self.timeout = timeout def wait(self, *names): end = time.monotonic() + self.timeout previous, stable = None, 0 while time.monotonic() < end: shot = self.desktop.screenshot() region = self.region_provider(shot) frame = self.matcher.frame(shot, region) found = next((name for name in names if self.matcher.matches(frame, self.flow['states'][name])), None) stable = stable + 1 if found and found == previous else (1 if found else 0) previous = found if stable >= 2: return found, shot, region, frame time.sleep(.25) raise WorkflowError('未识别到预期页面 ' + '/'.join(names) + '，已停止。') def observe(self): shot = self.desktop.screenshot() region = self.region_provider(shot) return shot, region, self.matcher.frame(shot, region) def click(self, name, observation, hold=.8): shot, region, _ = observation[-3:] if self.region_provider(shot) != region: raise WorkflowError('识别后游戏窗口移动，已停止，请重新运行。') self.desktop.click(self.flow['clicks'][name], region, shot.size, hold=hold) self.log('已执行：' + name) def click_location(self, location, observation): shot, region, _ = observation[-3:] x, y, width, height = location self.desktop.click(((x + width / 2) / 562, (y + height / 2) / 1002), region, shot.size, hold=.8) def swipe(self, name, observation): shot, region, _ = observation[-3:] start, end = self.flow['swipes'][name] self.desktop.drag(start, end, region, shot.size) time.sleep(self.flow.get('swipe_settle_seconds', .5)) def has_dot(self, frame, spec_name, search=None): spec = dict(self.flow[spec_name]) if search: spec['search'] = search score, _ = self.matcher.locate(frame, spec) return score >= spec.get('threshold', .86) def task_buttons_with_dot(self, frame): buttons = self.matcher.locate_all(frame, self.flow['task_button'], limit=8) for _, location in buttons: x, y, width, _ = location # The red dot sits just above the task-button's top-right corner. dot_search = [x + width - 32, max(0, y - 31), x + width + 8, y + 5] if self.has_dot(frame, 'card_task_dot', dot_search): yield location def dismiss_reward(self): reward = self.wait('reward') self.click('dismiss_reward', reward, hold=.2) self.wait('lamp') def claim_visible_tasks(self): claimed = 0 empty_scrolls = 0 while claimed < self.flow['limits']['task_claims']: lamp = self.wait('lamp') spec = self.flow['task_claim'] matches = self.matcher.locate_all(lamp[3], spec, limit=8) if matches: _, location = matches[0] self.click_location(location, lamp) self.log('已领取神灯任务奖励。') self.dismiss_reward() claimed += 1 empty_scrolls = 0 continue if empty_scrolls >= self.flow['limits']['task_scrolls']: break self.swipe('task_list_up', lamp) empty_scrolls += 1 return claimed def process_lamp(self): lamp = self.wait('lamp') if self.has_dot(lamp[3], 'task_tab_dot'): self.click('task_tab', lamp) self.wait('lamp') self.claim_visible_tasks() lamp = self.wait('lamp') if self.has_dot(lamp[3], 'gift_tab_dot'): self.click('gift_tab', lamp) gift = self.wait('gift') score, location = self.matcher.locate(gift[3], self.flow['free_button']) if score >= self.flow['free_button'].get('threshold', .86): self.click_location(location, gift) self.log('已领取免费神灯。') self.dismiss_reward() else: self.log('神灯礼包没有可领取的免费神灯，跳过。') lamp = self.wait('lamp') self.click('close_lamp', lamp, hold=.2) time.sleep(.5) def return_home(self): observation = self.observe() self.click('shop_back', observation, hold=.2) self.wait('home') self.log('已确认返回首页。') def run(self, dry_run=False): home = self.wait('home') if dry_run: self.log('首页识别通过。计划：商城 → 英雄招募限时 UP → 神灯任务/礼包 → 返回首页。未点击。') return 'preview' self.click('shop', home, hold=.8) shop = self.wait('shop') for _ in range(self.flow['limits']['scroll_to_recruit']): self.swipe('shop_down', shop) shop = self.observe() if self.matcher.locate_all(shop[2], self.flow['task_button'], limit=1): break handled, blank_pages = 0, 0 while handled < self.flow['limits']['cards']: _, _, frame = self.observe() task_button = next(self.task_buttons_with_dot(frame), None) if task_button: # Click the button itself, rather than its red dot. x, y, width, height = task_button shot, region, current = self.observe() self.desktop.click(((x + width / 2) / 562, (y + height / 2) / 1002), region, shot.size, hold=.8) self.log('已打开限时 UP 卡片的神灯召唤。') self.process_lamp() handled += 1 blank_pages = 0 continue if blank_pages >= self.flow['limits']['card_pages']: break # The card stack precedes 图腾寻宝. Pulling the list down shows # earlier UP cards without ever touching the treasure buttons. self.swipe('shop_down', self.observe()) blank_pages += 1 self.return_home() if handled: self.log('已处理 %s 张限时 UP 卡片。' % handled) return 'claimed' self.log('没有带红点的限时 UP 任务卡片，未领取奖励。') return 'no_card_dot'