171 lines
6.7 KiB
Python
171 lines
6.7 KiB
Python
import json
|
|
import queue
|
|
import threading
|
|
import tkinter as tk
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
from tkinter import filedialog, messagebox, ttk
|
|
|
|
from PIL import Image, ImageTk
|
|
|
|
from .engine import Desktop, Matcher, Region, Runner, load_flow
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
LOCAL = ROOT / 'local'
|
|
|
|
|
|
class Selector(tk.Toplevel):
|
|
def __init__(self, parent, screenshot, on_select):
|
|
super().__init__(parent)
|
|
self.title('拖动框选游戏内容：不包含微信标题栏和窗口边框')
|
|
self.original = screenshot
|
|
self.on_select = on_select
|
|
ratio = min(1000 / screenshot.width, 700 / screenshot.height, 1)
|
|
self.preview_size = (round(screenshot.width * ratio), round(screenshot.height * ratio))
|
|
self.photo = ImageTk.PhotoImage(screenshot.resize(self.preview_size))
|
|
self.canvas = tk.Canvas(self, width=self.preview_size[0], height=self.preview_size[1], highlightthickness=0)
|
|
self.canvas.pack()
|
|
self.canvas.create_image(0, 0, image=self.photo, anchor='nw')
|
|
self.start = None
|
|
self.rect = None
|
|
self.canvas.bind('<ButtonPress-1>', self.press)
|
|
self.canvas.bind('<B1-Motion>', self.drag)
|
|
self.canvas.bind('<ButtonRelease-1>', self.release)
|
|
|
|
def press(self, event):
|
|
self.start = (event.x, event.y)
|
|
if self.rect:
|
|
self.canvas.delete(self.rect)
|
|
self.rect = self.canvas.create_rectangle(event.x, event.y, event.x, event.y, outline='#ff3333', width=3)
|
|
|
|
def drag(self, event):
|
|
if self.start:
|
|
self.canvas.coords(self.rect, *self.start, event.x, event.y)
|
|
|
|
def release(self, event):
|
|
if not self.start:
|
|
return
|
|
x1, x2 = sorted((self.start[0], event.x))
|
|
y1, y2 = sorted((self.start[1], event.y))
|
|
sx = self.original.width / self.preview_size[0]
|
|
sy = self.original.height / self.preview_size[1]
|
|
region = Region(round(x1 * sx), round(y1 * sy), round((x2 - x1) * sx), round((y2 - y1) * sy))
|
|
try:
|
|
region.validate(self.original.size)
|
|
except Exception as exc:
|
|
messagebox.showerror('重新框选', str(exc), parent=self)
|
|
return
|
|
self.on_select(region)
|
|
self.destroy()
|
|
|
|
|
|
class App(tk.Tk):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.title('蔚蓝星球 · 本地自动化')
|
|
self.geometry('720x520')
|
|
self.region = None
|
|
self.events = queue.Queue()
|
|
self.stop = threading.Event()
|
|
self.busy = False
|
|
self.flow = load_flow(ROOT / 'workflows' / 'pig_claim.json')
|
|
self.matcher = Matcher(ROOT / 'assets')
|
|
self.status = tk.StringVar(value='请先将游戏放在主屏幕，打开首页并校准游戏内容区域。')
|
|
ttk.Label(self, text='第一条流程：小金猪挂机奖励领取', font=('', 18)).pack(pady=16)
|
|
ttk.Label(self, text='开始前请保持游戏窗口可见且位置不变。鼠标移到主屏幕左上角可紧急停止。').pack()
|
|
bar = ttk.Frame(self)
|
|
bar.pack(pady=14)
|
|
self.buttons = []
|
|
for name, cmd in [('校准游戏区域', self.capture), ('导入截图校准', self.import_image), ('预览流程', lambda: self.launch(True)), ('执行领取', lambda: self.launch(False))]:
|
|
button = ttk.Button(bar, text=name, command=cmd)
|
|
button.pack(side='left', padx=4)
|
|
self.buttons.append(button)
|
|
ttk.Button(self, text='停止', command=self.stop.set).pack()
|
|
ttk.Label(self, textvariable=self.status, wraplength=680).pack(pady=10)
|
|
self.output = tk.Text(self, height=16, wrap='word', state='disabled')
|
|
self.output.pack(fill='both', expand=True, padx=16, pady=12)
|
|
self.protocol('WM_DELETE_WINDOW', self.close)
|
|
config = LOCAL / 'region.json'
|
|
if config.exists():
|
|
try:
|
|
self.region = Region(**json.loads(config.read_text()))
|
|
self.status.set('已加载游戏区域；移动游戏窗口或更换屏幕后请重新校准。')
|
|
except (ValueError, TypeError):
|
|
pass
|
|
self.after(100, self.poll)
|
|
|
|
def log(self, text):
|
|
self.events.put(('log', text))
|
|
|
|
def set_region(self, region):
|
|
LOCAL.mkdir(exist_ok=True)
|
|
(LOCAL / 'region.json').write_text(json.dumps(asdict(region)), encoding='utf-8')
|
|
self.region = region
|
|
self.status.set('校准完成。先预览，确认首页识别通过后再执行。')
|
|
|
|
def capture(self):
|
|
self.iconify()
|
|
self.after(1500, self.take_capture)
|
|
|
|
def take_capture(self):
|
|
try:
|
|
screenshot = Desktop().screenshot()
|
|
except Exception as exc:
|
|
self.deiconify()
|
|
messagebox.showerror('截图失败', '请检查系统屏幕录制权限。\n' + str(exc))
|
|
return
|
|
self.deiconify()
|
|
Selector(self, screenshot, self.set_region)
|
|
|
|
def import_image(self):
|
|
name = filedialog.askopenfilename(filetypes=[('主屏幕完整截图', '*.png *.jpg *.jpeg')])
|
|
if name:
|
|
with Image.open(name) as image:
|
|
Selector(self, image.convert('RGB').copy(), self.set_region)
|
|
|
|
def launch(self, dry_run):
|
|
if self.busy:
|
|
return
|
|
if self.region is None:
|
|
messagebox.showinfo('需要校准', '请先校准游戏内容区域。')
|
|
return
|
|
self.busy = True
|
|
self.stop.clear()
|
|
for button in self.buttons:
|
|
button.configure(state='disabled')
|
|
self.status.set('正在预览…' if dry_run else '3 秒后执行，请将游戏放到前台。')
|
|
region = self.region
|
|
def work():
|
|
try:
|
|
if self.stop.wait(3):
|
|
return
|
|
Runner(Desktop(), self.matcher, region, self.log, self.stop).run(self.flow, dry_run)
|
|
except Exception as exc:
|
|
self.log('流程停止：' + str(exc))
|
|
finally:
|
|
self.events.put(('done', None))
|
|
threading.Thread(target=work, daemon=True).start()
|
|
|
|
def poll(self):
|
|
while not self.events.empty():
|
|
kind, value = self.events.get_nowait()
|
|
if kind == 'log':
|
|
self.output.configure(state='normal')
|
|
self.output.insert('end', value + '\n')
|
|
self.output.see('end')
|
|
self.output.configure(state='disabled')
|
|
elif kind == 'done':
|
|
self.busy = False
|
|
for button in self.buttons:
|
|
button.configure(state='normal')
|
|
self.status.set('本轮已结束，请查看日志中的实际结果。')
|
|
self.after(100, self.poll)
|
|
|
|
def close(self):
|
|
self.stop.set()
|
|
self.destroy()
|
|
|
|
|
|
def main():
|
|
App().mainloop()
|