112 lines
4.8 KiB
Python
112 lines
4.8 KiB
Python
"""Verified offer-claim workflow, with an explicit already-claimed branch."""
|
|
import time
|
|
|
|
from .engine import Desktop, Region, WorkflowError
|
|
from .permissions import mouse_permission_granted
|
|
|
|
|
|
class MacGameDesktop(Desktop):
|
|
def window_bounds(self):
|
|
import Quartz
|
|
windows = Quartz.CGWindowListCopyWindowInfo(
|
|
Quartz.kCGWindowListOptionOnScreenOnly, Quartz.kCGNullWindowID)
|
|
matches = [w for w in windows
|
|
if w.get('kCGWindowName') == '永远的蔚蓝星球'
|
|
and w.get('kCGWindowOwnerName') == '微信']
|
|
if len(matches) != 1:
|
|
raise WorkflowError('请在微信中正常打开一个游戏窗口，不要直接启动 WeApp。')
|
|
return dict(matches[0]['kCGWindowBounds'])
|
|
|
|
def region_for(self, screenshot):
|
|
b = self.window_bounds()
|
|
width, height = self.api.size()
|
|
sx, sy = screenshot.width / width, screenshot.height / height
|
|
# This observed WeChat header is 44 logical points high. Expanded
|
|
# sidebars change aspect ratio and are rejected by Region.validate.
|
|
region = Region(round(b['X'] * sx), round((b['Y'] + 44) * sy),
|
|
round(b['Width'] * sx), round((b['Height'] - 44) * sy))
|
|
region.validate(screenshot.size)
|
|
return region
|
|
|
|
def activate(self):
|
|
if not mouse_permission_granted():
|
|
raise WorkflowError('请先为运行宿主授予 macOS 辅助功能权限。')
|
|
b = self.window_bounds()
|
|
self.region_for(self.screenshot())
|
|
# Focus with a click in the observed empty part of the header, so
|
|
# the first game-button click is not consumed by window activation.
|
|
self.api.moveTo(b['X'] + 80, b['Y'] + 22, duration=0.2)
|
|
self.api.mouseDown()
|
|
try:
|
|
time.sleep(0.2)
|
|
finally:
|
|
saved = self.api.FAILSAFE
|
|
self.api.FAILSAFE = False
|
|
try:
|
|
self.api.mouseUp()
|
|
finally:
|
|
self.api.FAILSAFE = saved
|
|
time.sleep(0.3)
|
|
|
|
|
|
class OfferRunner:
|
|
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):
|
|
deadline = time.monotonic() + self.timeout
|
|
previous, count = None, 0
|
|
while time.monotonic() < deadline:
|
|
screenshot = self.desktop.screenshot()
|
|
region = self.region_provider(screenshot)
|
|
frame = self.matcher.frame(screenshot, region)
|
|
detected = next((name for name in names
|
|
if self.matcher.matches(frame, self.flow['states'][name])), None)
|
|
count = count + 1 if detected and detected == previous else 1 if detected else 0
|
|
previous = detected
|
|
if count >= 2:
|
|
return detected, screenshot, region
|
|
time.sleep(0.25)
|
|
raise WorkflowError('未识别到预期页面 ' + '/'.join(names) + '，已停止，不重复领取。')
|
|
|
|
def click(self, action, observation):
|
|
_, screenshot, region = observation
|
|
# Stop on movement after the recognition frame instead of clicking
|
|
# stale coordinates. A future run can re-locate the moved window.
|
|
if self.region_provider(screenshot) != region:
|
|
raise WorkflowError('识别后游戏窗口移动，已停止，请重新运行。')
|
|
self.desktop.click(self.flow['clicks'][action], region, screenshot.size)
|
|
self.log('已执行：' + action)
|
|
|
|
def run(self, dry_run=False):
|
|
home = self.wait('home')
|
|
if dry_run:
|
|
self.log('首页识别通过。计划：特惠 → 一键领取 → 关闭奖励 → 验证按钮变灰 → 返回首页。未点击。')
|
|
return 'preview'
|
|
self.click('open', home)
|
|
offer = self.wait('available', 'claimed', 'menu')
|
|
if offer[0] == 'menu':
|
|
self.click('battlepass', offer)
|
|
offer = self.wait('available', 'claimed')
|
|
if offer[0] == 'claimed':
|
|
self.log('一键领取已变灰，当前没有可领取奖励；跳过领取。')
|
|
self.click('back', offer)
|
|
self.wait('home')
|
|
self.log('已确认返回首页。')
|
|
return 'already_claimed'
|
|
self.click('claim', offer)
|
|
reward = self.wait('reward')
|
|
self.log('已确认“恭喜获得”奖励弹框。')
|
|
self.click('dismiss', reward)
|
|
completed = self.wait('claimed')
|
|
self.log('已确认一键领取按钮变灰。')
|
|
self.click('back', completed)
|
|
self.wait('home')
|
|
self.log('领取完成，已确认返回首页。')
|
|
return 'claimed'
|