Files
weilanxingqiu/automation/hero_rally.py
T

84 lines
4.7 KiB
Python

"""英雄集结中剑来献礼与骊珠画卷领奖流程。"""
import time
from .engine import WorkflowError
class HeroRallyRunner:
def __init__(self, desktop, matcher, flow, region_provider, log=print, timeout=12):
self.desktop, self.matcher, self.flow = desktop, matcher, flow
self.region_provider, self.log, self.timeout = region_provider, log, timeout
def observe(self):
shot = self.desktop.screenshot(); region = self.region_provider(shot)
return shot, region, self.matcher.frame(shot, region)
def wait(self, *names):
deadline = time.monotonic() + self.timeout; previous = None; stable = 0
while time.monotonic() < deadline:
shot, region, frame = self.observe()
found = next((n for n in names if self.matcher.matches(frame, self.flow['states'][n])), None)
stable = stable + 1 if found == previous and found else (1 if found else 0); previous = found
if stable >= 2: return found, shot, region, frame
time.sleep(.25)
raise WorkflowError('未识别到预期页面 ' + '/'.join(names) + ',已停止。')
def click(self, point, obs):
_, shot, region, _ = obs
if self.region_provider(shot) != region: raise WorkflowError('识别后游戏窗口移动,已停止,请重新运行。')
self.desktop.click(point, region, shot.size, hold=.8)
def action(self, name, obs): self.click(self.flow['clicks'][name], obs); self.log('已执行:' + name)
def rally(self, home): self.action('open_rally', home); return self.wait('rally')
def close_home(self, rally): self.action('close_rally', rally); self.wait('home'); self.log('已确认返回首页。')
class SwordGiftRunner(HeroRallyRunner):
def run(self, dry_run=False):
home = self.wait('home')
if dry_run: self.log('首页识别通过。计划:英雄集结 → 剑来献礼红点 → 一键领取 → 返回首页。未点击。'); return 'preview'
rally = self.rally(home)
if not self.matcher.matches(rally[3], self.flow['sword_dot']):
self.log('剑来献礼没有红点,本次跳过。'); self.close_home(rally); return 'no_dot'
self.action('open_sword', rally); page = self.wait('sword_available', 'sword_claimed')
if page[0] == 'sword_claimed':
self.log('剑来献礼已领完,本次跳过。')
else:
self.action('claim_sword', page); reward = self.wait('reward'); self.action('dismiss_reward', reward); page = self.wait('sword_claimed')
self.action('sword_back', page); self.close_home(self.wait('rally')); return 'claimed'
class LizhuTaskRunner(HeroRallyRunner):
def claim_button(self, frame):
spec = self.flow['claim_button']; score, loc = self.matcher.locate(frame, spec)
if score < spec.get('threshold', .86): return None
x, y, w, h = loc; return ((x+w/2)/562, (y+h/2)/1002)
def claim_panel(self, panel):
count = 0
while count < self.flow['max_claims_per_panel']:
obs = self.wait(panel); point = self.claim_button(obs[3])
if not point: return count
self.click(point, obs); count += 1; self.log('已执行:%s 第 %s 个领取按钮。' % (panel, count))
self.action('dismiss_reward', self.wait('reward'))
raise WorkflowError('单个面板领取次数超过安全上限,已停止。')
def run(self, dry_run=False):
home = self.wait('home')
if dry_run: self.log('首页识别通过。计划:英雄集结 → 骊珠画卷红点 → 活动/每日任务奖励 → 返回首页。未点击。'); return 'preview'
rally = self.rally(home)
if not self.matcher.matches(rally[3], self.flow['lizhu_dot']):
self.log('骊珠画卷没有红点,本次跳过。'); self.close_home(rally); return 'no_dot'
self.action('open_lizhu', rally); lizhu = self.wait('lizhu'); self.action('open_tasks', lizhu)
panel = self.wait('activity', 'daily'); total = 0
if self.matcher.matches(panel[3], self.flow['tab_dots']['activity']): total += self.claim_panel('activity')
else: self.log('活动任务没有红点,不领取。')
panel = self.wait('activity', 'daily')
if self.matcher.matches(panel[3], self.flow['tab_dots']['daily']):
if panel[0] != 'daily': self.action('open_daily', panel)
total += self.claim_panel('daily')
else: self.log('每日任务没有红点,不切换。')
panel = self.wait('activity', 'daily'); self.action('close_tasks', panel)
self.action('lizhu_back', self.wait('lizhu')); self.close_home(self.wait('rally'))
return 'claimed' if total else 'no_claim_button'