227 lines
9.3 KiB
Python
227 lines
9.3 KiB
Python
import json
|
|
import math
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import cv2
|
|
import numpy as np
|
|
from PIL import Image
|
|
|
|
|
|
class WorkflowError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Region:
|
|
# Coordinates in primary-screen screenshot pixels, not mouse points.
|
|
x: int
|
|
y: int
|
|
width: int
|
|
height: int
|
|
|
|
def validate(self, screen_size):
|
|
if min(self.x, self.y) < 0 or min(self.width, self.height) <= 0:
|
|
raise WorkflowError('请先框选有效的游戏内容区域。')
|
|
if self.x + self.width > screen_size[0] or self.y + self.height > screen_size[1]:
|
|
raise WorkflowError('屏幕尺寸已变化，请重新校准游戏区域。')
|
|
if abs(self.width / self.height - 562 / 1002) > 0.035:
|
|
raise WorkflowError('区域比例不符合竖屏游戏，请只框选游戏内容，不包含标题栏。')
|
|
|
|
def mouse_point(self, point, screenshot_size, mouse_size):
|
|
if len(point) != 2 or not all(math.isfinite(v) and 0 < v < 1 for v in point):
|
|
raise WorkflowError('点击位置必须位于游戏区域内部。')
|
|
self.validate(screenshot_size)
|
|
return (
|
|
round((self.x + point[0] * self.width) * mouse_size[0] / screenshot_size[0]),
|
|
round((self.y + point[1] * self.height) * mouse_size[1] / screenshot_size[1]),
|
|
)
|
|
|
|
|
|
class Matcher:
|
|
def __init__(self, assets):
|
|
self.assets = Path(assets)
|
|
self.cache = {}
|
|
|
|
def frame(self, screenshot, region):
|
|
region.validate(screenshot.size)
|
|
crop = screenshot.crop((region.x, region.y, region.x + region.width, region.y + region.height))
|
|
return np.asarray(crop.convert('RGB').resize((562, 1002), Image.Resampling.LANCZOS))
|
|
|
|
def score(self, frame, spec):
|
|
return self.locate(frame, spec)[0]
|
|
|
|
def locate(self, frame, spec):
|
|
name = spec['template']
|
|
if name not in self.cache:
|
|
p = self.assets / name
|
|
if not p.is_file():
|
|
raise WorkflowError('缺少识别模板：' + name)
|
|
self.cache[name] = np.asarray(Image.open(p).convert('RGB'))
|
|
template = self.cache[name]
|
|
x1, y1, x2, y2 = spec['search']
|
|
area = frame[y1:y2, x1:x2]
|
|
h, w = template.shape[:2]
|
|
if area.shape[0] < h or area.shape[1] < w:
|
|
raise WorkflowError('模板尺寸超出识别区域。')
|
|
result = cv2.matchTemplate(area, template, cv2.TM_CCOEFF_NORMED)
|
|
_, score, _, (x, y) = cv2.minMaxLoc(result)
|
|
candidate = area[y:y + h, x:x + w]
|
|
# Correlation alone can mistake a darkened background behind a popup
|
|
# for the active home page. Also require similar absolute colors.
|
|
difference = float(np.abs(candidate.astype(float) - template.astype(float)).mean())
|
|
if difference > spec.get('max_color_difference', 28):
|
|
return 0.0, None
|
|
if not math.isfinite(score):
|
|
return 0.0, None
|
|
return float(score), (x1 + x, y1 + y, w, h)
|
|
|
|
def locate_all(self, frame, spec, limit=20):
|
|
"""Return non-overlapping template matches, ordered from top to bottom."""
|
|
name = spec['template']
|
|
if name not in self.cache:
|
|
p = self.assets / name
|
|
if not p.is_file():
|
|
raise WorkflowError('缺少识别模板：' + name)
|
|
self.cache[name] = np.asarray(Image.open(p).convert('RGB'))
|
|
template = self.cache[name]
|
|
x1, y1, x2, y2 = spec['search']
|
|
area = frame[y1:y2, x1:x2]
|
|
h, w = template.shape[:2]
|
|
if area.shape[0] < h or area.shape[1] < w:
|
|
raise WorkflowError('模板尺寸超出识别区域。')
|
|
scores = cv2.matchTemplate(area, template, cv2.TM_CCOEFF_NORMED)
|
|
threshold = spec.get('threshold', 0.86)
|
|
matches = []
|
|
while len(matches) < limit:
|
|
_, score, _, (x, y) = cv2.minMaxLoc(scores)
|
|
if not math.isfinite(score) or score < threshold:
|
|
break
|
|
candidate = area[y:y + h, x:x + w]
|
|
difference = float(np.abs(candidate.astype(float) - template.astype(float)).mean())
|
|
# Suppress this neighbourhood regardless of the colour check so
|
|
# a rejected partial overlap cannot repeat forever.
|
|
scores[max(0, y - h // 2):min(scores.shape[0], y + h // 2 + 1),
|
|
max(0, x - w // 2):min(scores.shape[1], x + w // 2 + 1)] = -1
|
|
if difference <= spec.get('max_color_difference', 28):
|
|
matches.append((float(score), (x1 + x, y1 + y, w, h)))
|
|
return sorted(matches, key=lambda item: (item[1][1], item[1][0]))
|
|
|
|
def matches(self, frame, specs):
|
|
return all(self.score(frame, s) >= s.get('threshold', 0.86) for s in specs)
|
|
|
|
|
|
class Desktop:
|
|
def __init__(self):
|
|
import pyautogui
|
|
self.api = pyautogui
|
|
self.api.FAILSAFE = True
|
|
self.api.PAUSE = 0.1
|
|
|
|
def screenshot(self):
|
|
return self.api.screenshot()
|
|
|
|
def click(self, point, region, screenshot_size, hold=0.2):
|
|
from .permissions import mouse_permission_granted
|
|
if not mouse_permission_granted():
|
|
raise WorkflowError(
|
|
'macOS 辅助功能权限未授予，无法发送有效鼠标点击。'
|
|
'请在系统设置 → 隐私与安全性 → 辅助功能中启用运行程序的宿主'
|
|
'（如 ChatGPT/Codex 或终端），重启宿主后再测试。'
|
|
)
|
|
x, y = region.mouse_point(point, screenshot_size, tuple(self.api.size()))
|
|
self.api.moveTo(x, y, duration=0.2)
|
|
self.api.mouseDown()
|
|
try:
|
|
# The game did not react to instant down/up in the live test.
|
|
time.sleep(hold)
|
|
finally:
|
|
# Always release, including if the pointer moved to a failsafe
|
|
# corner during the hold. Restore the guard immediately afterward.
|
|
saved_failsafe = self.api.FAILSAFE
|
|
self.api.FAILSAFE = False
|
|
try:
|
|
self.api.mouseUp()
|
|
finally:
|
|
self.api.FAILSAFE = saved_failsafe
|
|
|
|
def drag(self, start, end, region, screenshot_size, hold=0.2, duration=0.7):
|
|
"""Perform a deliberate touch-like list swipe inside the game region."""
|
|
from .permissions import mouse_permission_granted
|
|
if not mouse_permission_granted():
|
|
raise WorkflowError('macOS 辅助功能权限未授予，无法发送有效鼠标拖拽。')
|
|
start_point = region.mouse_point(start, screenshot_size, tuple(self.api.size()))
|
|
end_point = region.mouse_point(end, screenshot_size, tuple(self.api.size()))
|
|
self.api.moveTo(*start_point, duration=0.2)
|
|
self.api.mouseDown()
|
|
try:
|
|
time.sleep(hold)
|
|
self.api.moveTo(*end_point, duration=duration)
|
|
time.sleep(hold)
|
|
finally:
|
|
saved = self.api.FAILSAFE
|
|
self.api.FAILSAFE = False
|
|
try:
|
|
self.api.mouseUp()
|
|
finally:
|
|
self.api.FAILSAFE = saved
|
|
|
|
|
|
class Runner:
|
|
def __init__(self, desktop, matcher, region, log=print, stop=None):
|
|
self.desktop = desktop
|
|
self.matcher = matcher
|
|
self.region = region
|
|
self.log = log
|
|
self.stop = stop or threading.Event()
|
|
|
|
def check_stop(self):
|
|
if self.stop.is_set():
|
|
raise WorkflowError('已停止。')
|
|
|
|
def wait_state(self, specs, timeout=10, absent=False):
|
|
end = time.monotonic() + timeout
|
|
stable = 0
|
|
while time.monotonic() < end:
|
|
self.check_stop()
|
|
screenshot = self.desktop.screenshot()
|
|
frame = self.matcher.frame(screenshot, self.region)
|
|
matched = self.matcher.matches(frame, specs)
|
|
stable = stable + 1 if matched != absent else 0
|
|
if stable >= 2:
|
|
return screenshot
|
|
self.stop.wait(0.35)
|
|
raise WorkflowError('等待页面状态超时，流程停止，请检查当前界面。')
|
|
|
|
def run(self, flow, dry_run=False):
|
|
self.check_stop()
|
|
# A dry run validates the CURRENT page and lists later actions;
|
|
# it cannot verify pages that have not actually been opened.
|
|
initial = self.wait_state(flow['steps'][0]['expect'])
|
|
if dry_run:
|
|
for step in flow['steps']:
|
|
self.log('计划：' + step['name'])
|
|
self.log('预览完成：首页识别通过，后续页面尚未验证；未执行点击。')
|
|
return
|
|
for step in flow['steps']:
|
|
self.log('检查：' + step['name'])
|
|
screenshot = self.wait_state(step['expect'], step.get('timeout', 10))
|
|
self.check_stop()
|
|
if 'click' in step:
|
|
# One click only; never retry a claim after an uncertain result.
|
|
self.desktop.click(step['click'], self.region, screenshot.size)
|
|
self.log('已点击：' + step['name'])
|
|
if 'wait_absent' in step:
|
|
self.wait_state(step['wait_absent'], step.get('timeout', 10), absent=True)
|
|
if step.get('settle'):
|
|
self.stop.wait(step['settle'])
|
|
self.check_stop()
|
|
self.wait_state(flow['finish'])
|
|
self.log(flow['result'])
|
|
|
|
|
|
def load_flow(path):
|
|
return json.loads(Path(path).read_text(encoding='utf-8'))
|