yys_auto/scriptBase/comon.py

# -*- coding: utf-8 -*-
from datetime import datetime
import threading 
import pyautogui
import win32com
from scriptBase.imgFind import pcacc_img
from scriptBase.mouseClick import pcacc_mouse
import time
import random
import pygetwindow as gw
import re
from PIL import Image
from PIL import ImageFilter
import io
from scriptBase.ocr import *
from scriptBase.record import ScreenRecorder
import win32gui
import win32con

g_recorder = ScreenRecorder()

def compress_image(image, target_size = 100 * 1024):
    # 调整图像尺寸，保持纵横比
    image.thumbnail((1024, 1024))

    # 尝试不同的压缩质量
    quality = 80
    compressed_image = io.BytesIO()
    while compressed_image.tell() < target_size and quality >= 10:
        image.save(compressed_image, format='JPEG', quality=quality)
        quality -= 10
        compressed_image.seek(0)
        compressed_size = len(compressed_image.getvalue())

    return compressed_image.getvalue()

'''
def binarize_image(image):
    # 将图片转换为黑白格式（二值化）
    # 这里使用Pillow库来进行图片二值化
    image = image.convert("L")  # 将图片转换为灰度图像
    threshold = 128  # 设定二值化阈值，0~255之间，通常取128
    image = image.point(lambda x: 0 if x < threshold else 255, '1')  # 进行二值化处理
    return image
'''

def binarize_image(image):
    # 将图片转换为灰度图像
    image = image.convert("L")
    # 降噪处理（中值滤波）
    image = image.filter(ImageFilter.MedianFilter(size=3))
    # 自适应阈值二值化
    threshold = image.point(lambda x: 0 if x < 128 else 255, '1')
    
    return threshold


def waitFindImg(imgPath, waitTime, preWaitTime, isCV=False):
    pre = time.time()
    if preWaitTime > 0:
        myTimeSleep(preWaitTime)
    #sleep 0.2s per times
    curTime = 0
    while waitTime == -1 or curTime < waitTime:
        ret, pos_cur = pcacc_img.find_imgs(imgPath, isCV)
        if ret:
            return ret, pos_cur
        else:
            myTimeSleep(0.1)
            curTime = time.time() - pre
    costTime = time.time() - pre
    print(f"waitFindImg failed, cost:{costTime:.2f}, img:{imgPath}")
    return False, None


def waitClickImg_noWait(imgPath, waitTime, preWaitTime, isCV=False):
    pre = time.time()
    if preWaitTime > 0:
        myTimeSleep(preWaitTime)
    #sleep 0.2s per times
    curTime = 0
    while waitTime == -1 or curTime < waitTime:
        ret, pos_cur = pcacc_img.find_imgs(imgPath, isCV)
        if ret:
            pcacc_mouse.click(pos_cur[0], pos_cur[1])
            myTimeSleep(0.5)
            return True, pos_cur
        else:
            myTimeSleep(0.1)
            curTime = time.time() - pre
    costTime = time.time() - pre
    print(f"waitClickImg_noWait failed, cost:{costTime:.2f}, img:{imgPath}")
    return False, (-1, -1)

def waitFindImg_withBool(imgPath, waitTime, preWaitTime, isCV=False):
    ret, pos = waitFindImg(imgPath, waitTime, preWaitTime, isCV)
    return ret

def waitClickImg_noWait_withBool(imgPath, waitTime, preWaitTime, isCV=False):
    ret, pos = waitClickImg_noWait(imgPath, waitTime, preWaitTime, isCV)
    return ret


def waitclickMultiImg(imgArr, waitTime, preWaitTime):
    if preWaitTime > 0:
        myTimeSleep(preWaitTime)
    #sleep 0.2s per times
    curTime = 0
    while waitTime == -1 or curTime < waitTime:
        ret, pos_cur = pcacc_img.find_imgArr(imgArr)
        if ret:
            pcacc_mouse.click(pos_cur[0], pos_cur[1])
            myTimeSleep(0.5)
            ret, pos_cur = pcacc_img.find_imgArr(imgArr)
            if ret == False:
                return True, pos_cur
            pcacc_mouse.click(pos_cur[0], pos_cur[1])
            myTimeSleep(0.5)
            ret, pos_cur = pcacc_img.find_imgArr(imgArr)
            if ret == False:
                return True, pos_cur
            else:
                return True, pos_cur
        else:
            myTimeSleep(0.2)
            curTime = curTime + 0.2
    print("waitclickMultiImg failed", imgArr)
    return False, (-1, -1)


def waitClickImg(imgPath, waitTime, preWaitTime, isCV=False):
    if preWaitTime > 0:
        myTimeSleep(preWaitTime)
    #sleep 0.2s per times
    curTime = 0
    while waitTime == -1 or curTime < waitTime:
        ret, pos_cur = pcacc_img.find_imgs(imgPath, isCV)
        if ret:
            pcacc_mouse.move_mouse(pos_cur[0], pos_cur[1])
            ret, pos_cur2 = pcacc_img.find_imgs(imgPath, isCV)
            if ret:
                pcacc_mouse.click(pos_cur2[0], pos_cur2[1])
            else:
                pcacc_mouse.click(pos_cur[0], pos_cur[1])
            myTimeSleep(0.5)
            ret, pos_cur = pcacc_img.find_imgs(imgPath, isCV)
            if ret == False:
                return True, pos_cur
            
            pcacc_mouse.move_mouse(pos_cur[0], pos_cur[1])
            ret, pos_cur2 = pcacc_img.find_imgs(imgPath, isCV)
            if ret:
                pcacc_mouse.click(pos_cur2[0], pos_cur2[1])
            else:
                pcacc_mouse.click(pos_cur[0], pos_cur[1])

            myTimeSleep(0.5)
            ret, pos_cur = pcacc_img.find_imgs(imgPath, isCV)
            if ret == False:
                return True, pos_cur
            else:
                return True, pos_cur
        else:
            myTimeSleep(0.2)
            curTime = curTime + 0.2
    print("waitClickImg failed", imgPath)
    return False, (-1, -1)

def waitClickImg_withBool(imgPath, waitTime, preWaitTime, isCV=False):
    ret, pos = waitClickImg(imgPath, waitTime, preWaitTime, isCV)
    return ret

def myTimeSleep(curTime , FunStatus=None):
    tmpTime = curTime * 1000
    randomTime = random.randint(int(tmpTime - tmpTime / 9), int(tmpTime + tmpTime / 9)) / 1000
    randomTime = round(randomTime)
    while randomTime > 50:
        time.sleep(50)
        randomTime -= 50
        showStr = f'下一轮{randomTime}s'
        if FunStatus is not None:
            FunStatus(showStr)
        print(showStr)
    time.sleep(randomTime)



def myTimeSleep_small():
    time.sleep(random.randint(200, 1000) / 1000)

def myTimeSleep_big():
    time.sleep(random.randint(1200, 2000) / 1000)

def get_random_point(region):
    left, top, width, height = region
    x = random.randint(left, left + width)
    y = random.randint(top, top + height)
    return x, y

def get_yys_random_point(region):
    dstRegion = yys_getRegion(region, True)
    print(dstRegion)
    return get_random_point(dstRegion)


def maintain_state(min):
    state = True
    last_reset_time = time.time()

    def inner_func():
        nonlocal state, last_reset_time

        # 获取当前时间
        current_time = time.time()

        # 如果距离上次重置的时间大于等于min秒，则将状态重置为True
        if current_time - last_reset_time >= min:
            state = True
            last_reset_time = current_time

        # 返回状态值，并将状态置为False
        result = state
        state = False
        return result

    return inner_func

def find_window_pos(title):
    # 查找窗口
    winList = gw.getWindowsWithTitle(title)
    if len(winList) == 0:
        return False, None
    window = gw.getWindowsWithTitle(title)[0]  # 替换为你要查找的窗口标题

    # 获取位置信息
    left = window.left
    top = window.top
    width = window.width
    height = window.height

    # 打印位置信息
    #print(f'Left: {left}, Top: {top}, Width: {width}, Height: {height}')
    if left > 0 and top > 0 and width > 0 and height > 0:
        return True, (left, top, width, height)
    else:
        return False, None

def find_yys_pos():
    return find_window_pos('MuMu安卓设备')

def get_first_single_digit(string):
    pattern = r'\d'  # 匹配一个独立的单个数字
    match = re.search(pattern, string)
    if match:
        return int(match.group())  # 将匹配到的字符串转换为整数
    else:
        return None

def dragAuto(gamestr, positions, isQuick=False):
    tmpRet, tmpPos = find_window_pos(gamestr)
    if tmpRet == False:
        return False
    
    x1 = positions[0] + tmpPos[0] + random.randint(-20, 20)
    y1 = positions[1] + tmpPos[1] + random.randint(-20, 20)
    x2 = positions[2] + tmpPos[0] + random.randint(-20, 20)
    y2 = positions[3] + tmpPos[1] + random.randint(-20, 20)
    pcacc_mouse.drag_from(x1, y1, x2, y2, isQuick)
    return True

def ocrAuto(gamestr, positions):
    try:
        tmpRet, tmpPos = find_window_pos(gamestr)
        if tmpRet == False:
            return []
        
        x1 = positions[0] + tmpPos[0]
        y1 = positions[1] + tmpPos[1]
        x2 = positions[2] + tmpPos[0]
        y2 = positions[3] + tmpPos[1]
        region = (x1, y1, x2, y2)
        lineStr = capture_and_ocr(region)
        print("识别到：", lineStr)
    except:
        lineStr = ""
    return lineStr

def tryMoveTopWindow(gameStr):
    """尝试查找并置顶包含指定关键词的窗口"""
    mumu_windows = []
    
    def enum_callback(hwnd, param):
        """枚举回调函数"""
        if win32gui.IsWindowVisible(hwnd):
            window_title = win32gui.GetWindowText(hwnd)
            if gameStr in window_title:
                param.append((hwnd, window_title))
        return True
    
    # 枚举所有顶级窗口
    win32gui.EnumWindows(enum_callback, mumu_windows)
    
    if not mumu_windows:
        print(f"未找到包含 '{gameStr}' 的窗口")
        return False
    
    # 输出找到的窗口信息（调试用）
    print(f"找到 {len(mumu_windows)} 个匹配窗口:")
    for i, (hwnd, title) in enumerate(mumu_windows):
        print(f"  {i+1}. 句柄: {hwnd}, 标题: {title}")
    
    # 处理第一个找到的窗口
    target_hwnd, target_title = mumu_windows[0]
    
    try:
        # 设置窗口置顶
        win32gui.SetWindowPos(target_hwnd, win32con.HWND_TOPMOST, 0, 0, 0, 0,
                             win32con.SWP_NOMOVE | win32con.SWP_NOSIZE)
        
        # 激活窗口到前台
        try:
            win32gui.SetForegroundWindow(target_hwnd)
        except Exception as e:
            print(f"设置前台窗口时警告: {e}")
        
        print(f"窗口已置顶: {target_title}")
        return True
        
    except Exception as e:
        print(f"设置窗口置顶时出错: {e}")
        return False

def yys_ocrAuto(positions):
    return ocrAuto('MuMu安卓设备', positions)

def yys_dragAuto(positions, isQuick=False):
    return dragAuto('MuMu安卓设备', positions, isQuick)

def getRegion(gamestr, region, isWidth):
    tmpRet, tmpPos = find_window_pos(gamestr)
    if tmpRet == False:
        return False
    if isWidth:
        return (region[0] + tmpPos[0], region[1] + tmpPos[1], region[2] - region[0], region[3] - region[1])
    else:
        return (region[0] + tmpPos[0], region[1] + tmpPos[1], region[2] + tmpPos[0], region[3] + tmpPos[1])
def game_region():
    tmpRet, tmpPos = find_window_pos('MuMu安卓设备',)
    return tmpRet, tmpPos

def save_game_screen(fileHeader=""):
    try:
        regionRet, regionPos = game_region()
        screenshot = pyautogui.screenshot(region=regionPos)
        curData = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
        fileName = "temp_pictures/" + fileHeader + "_" + curData + ".png"
        screenshot.save(fileName)
        return True
    except Exception as e:
        print(f"截图保存失败: {e}")
        return False

# 创建并启动截图线程
def start_screenshot_thread(fileHeader=""):
    screenshot_thread = threading.Thread(
        target=save_game_screen, 
        args=(fileHeader,),
        name=f"ScreenshotThread_{fileHeader}"
    )
    screenshot_thread.start()
    return screenshot_thread

def start_recording():
    global g_recorder
    regionRet, regionPos = game_region()
    print("record region,", regionPos)
    g_recorder.start_recording(regionPos)

def stop_recording():
    global g_recorder
    g_recorder.stop_recording()


def yys_getRegion(region, isWidth= False):
    return getRegion('MuMu安卓设备', region, isWidth)

def click_waitimg(pos_x,pos_y,img):
    retry_times = 3
    while  retry_times > 0 and waitFindImg_withBool(img,0.4,1) == False:
        pcacc_mouse.click(pos_x,pos_y)
        retry_times-=1
    return retry_times > 0

if __name__ == '__main__':
    tryMoveTopWindow('MuMu安卓设备')