opt/model/helper.py

import ctypes
import os
import threading
import time

from dataclasses import dataclass
from model.custom_struct import GameConfig, AccountInfo
from model.global_manager import GM
from scripts.script import Script
from tools.dm_operate import Dm
from tools.emulator.emulator import Emulator
from tools.emulator.ld_operate import LD
from tools.emulator.ys_operate import YS
from tools.upload_log import LogInfo
from tools.utils import Utils


@dataclass
class ScriptHelper:
    dm: Dm = None  # 大漠对象
    bind_handle: int = None  # 绑定句柄
    top_handle: int = None  # 顶层句柄
    emulator_index: int = None  # 模拟器下标
    timeout: int = None  # 超时时间
    game_id: str = None  # 游戏id
    game_type: str = None  # 游戏类型
    account: str = None  # 账号
    password: str = None  # 密码
    retained: int = None  # 新增留存
    log_uuid: str = None  # 日志uuid
    log_port: int = None  # 日志端口
    emu: Emulator = None  # 模拟器对象


class MyHelper:
    """模拟器任务助手"""

    _lock = threading.Lock()

    def __init__(self, window_id: int, game_config: GameConfig, account_info: AccountInfo):
        self.window_id = window_id
        self.game_config = game_config
        self.account_info = account_info
        self.emulator_type = self.game_config.emulator_type
        self.emu = self._initialize_emulator_instance()
        self.emulator_index = self._get_emulator()
        self.dm = Dm()
        self.emulator_info = None
        self.wy_ip = GM.wy_ip
        self.log_port = GM.script_log_port
        self.script = Script(self.game_config.script)
        self.log_uuid = f'{time.time()}_{self.account_info.account}'
        self.upload_log = LogInfo(self.log_uuid)
        self.upload_account_log(GM.device_info.pc_name, GM.device_info.officer, GM.device_mac)

    def _initialize_emulator_instance(self):
        emulator_mapping = {
            '雷电模拟器4-32位': lambda: LD(4),
            '雷电模拟器4-64位': lambda: LD(64),
            '雷电模拟器9': lambda: LD(9),
            '夜神模拟器7-32位': lambda: YS(7),
            '夜神模拟器7-64位': lambda: YS(8),
            '夜神模拟器9': lambda: YS(9)
        }
        # 仅当查找到类型时才会实例化，否则返回默认的 LD(9)
        return emulator_mapping.get(self.emulator_type, lambda: LD(9))()

    def _get_emulator(self):
        with MyHelper._lock:
            index = self.window_id
            while GM.get_global_control():

                if self.emu.is_exists(index):
                    if GM.get_emulator_status(f'{self.emulator_type}-{index}') == 0:
                        GM.set_emulator_status(f'{self.emulator_type}-{index}', 1)
                        return index
                    raise Exception(f'模拟器-{index}状态异常')
                else:
                    # 模拟器不存在，添加新的模拟器
                    self.emu.add(f'new-{index}')
                    time.sleep(3)

    def get_emulator_index(self):
        return self.emulator_index

    def set_emulator_index(self, index):
        self.emulator_index = index

    def restore_emulator(self):
        try:
            self.emu.restore(self.emulator_index, os.path.join(GM.image_path, self.game_config.image))
            time.sleep(1)
            self.emu.rename(self.emulator_index, f'{self.game_config.task_id}-{self.emulator_index}')
            return True, '还原模拟器成功'
        except Exception as e:
            return False, str(e)

    def modify_emulator(self):
        try:
            self.emu.modify(self.emulator_index,
                            cpu=self.game_config.cpu,
                            memory=self.game_config.memory,
                            resolution=self.game_config.resolution,
                            manufacturer=self.account_info.manufacturer,
                            model=self.account_info.model,
                            pnumber=self.account_info.pnumber,
                            imei=self.account_info.imei,
                            imsi=self.account_info.imsi,
                            simserial=self.account_info.simserial,
                            androidid=self.account_info.androidid,
                            mac=self.account_info.mac if self.emu.emulator_type == 'ld' else self.account_info.mac_colon
                            )
            return True, '修改模拟器信息成功'
        except Exception as e:
            return False, str(e)

    def switch_emulator_area(self):
        if self.wy_ip is None:
            return False, "未初始化无忧IP"

        self.wy_ip.set_wy_token()

        if len(self.account_info.number_operatorTypes.split('_')) < 1:
            return False, "number_operatorTypes参数错误"

        return self.wy_ip.switch_emulator_area(self.emulator_index,
                                               self.account_info.number_operatorTypes.split('_')[1],
                                               self.account_info.number_operatorTypes.split('_')[0])

    def confirm_emulator_area(self):
        result, message = self.wy_ip.queryProcessProxyRegion(self.emulator_info.vbox_pid)
        if not result:
            return result, message
        if message != self.account_info.number_operatorTypes.split('_')[0]:
            return False, f'当前模拟器切换地区失败,目标地区：{self.account_info.number_operatorTypes.split("_")[0]},当前地区：{message}'
        return True, '当前模拟器切换地区成功'

    def start_emulator_and_set_position(self):
        """启动模拟器并设置窗口位置"""
        try:
            # 获取共享路径
            share_path = GM.get_share_path()
            emulator_share_dir = os.path.join(share_path, str(self.window_id))

            # 如果目录不存在则创建
            if not os.path.exists(emulator_share_dir):
                os.makedirs(emulator_share_dir)

            # 设置共享目录
            self.emu.set_share_dir(self.emulator_index, emulator_share_dir)

            # 启动模拟器并确认启动状态
            result, message = self.emu.start_and_confirm(self.emulator_index)

            if result:
                self.emulator_info = message
                # 设置模拟器窗口位置
                self.emu.set_emulator_position(self.emulator_info.top_handle, self.window_id)
                time.sleep(1)
                return True, f"模拟器编号：{self.emulator_index}，启动成功"
            return False, f"模拟器编号：{self.emulator_index}，,启动失败,错误信息：{message}"
        except Exception as e:
            return False, f"模拟器编号：{self.emulator_index}，启动失败，错误信息：{str(e)}"

    def close_emulator_and_confirm(self):
        """关闭模拟器并确保所有相关进程被终止。"""
        # 检查模拟器是否正在运行
        if self.emu.is_running(self.emulator_index):
            self.emu.close(self.emulator_index)
            time.sleep(3)  # 等待模拟器关闭的缓冲时间

        # 获取模拟器信息
        emu_info = self.emu.get_info(self.emulator_index)
        if emu_info:
            # 确认主进程 pid 有效
            if emu_info.pid > 0:
                Utils.kill_process(emu_info.pid)
            # 确认 VBox 进程 pid 有效
            if emu_info.vbox_pid > 0:
                Utils.kill_process(emu_info.vbox_pid)

    def bind_emulator(self):
        try:
            if self.game_config.script.endswith('.dll'):
                result = self.script.execute('dm_bind',
                                             self.emulator_index,
                                             self.emulator_info.bind_handle,
                                             timeout=10
                                             )
                return True, result
            elif self.game_config.script.endswith('.py'):
                script_helper = ScriptHelper(dm=self.dm,
                                             emu=self.emu,
                                             emulator_index=self.emulator_index,
                                             top_handle=self.emulator_info.top_handle,
                                             bind_handle=self.emulator_info.bind_handle,
                                             game_type=self.game_config.game_type,
                                             game_id=self.game_config.task_id,
                                             account=self.account_info.account,
                                             password=self.account_info.password,
                                             retained=self.account_info.retained,
                                             log_uuid=self.log_uuid,
                                             log_port=self.log_port,
                                             timeout=self.game_config.timeout * 60
                                             )
                result = self.script.execute('dm_bind', script_helper, timeout=10)
                return True, result
            else:
                return False, '脚本类型错误'

        except Exception as e:
            return False, str(e)

    def start_game(self):
        """启动游戏并处理可能的异常。

        Returns:
            tuple: 成功与否的标志和结果或错误信息。
        """
        try:
            if self.game_config.script.endswith('.dll'):
                # 获取字符串的字节数据
                task_id_bytes = self.game_config.task_id.encode('gbk')
                account_bytes = self.account_info.account.encode('gbk')
                script_path_bytes = self.script.script_path.encode('gbk')
                log_uuid_bytes = self.log_uuid.encode('gbk')

                # 创建 c_char_p 指针
                task_id_ptr = ctypes.c_char_p(task_id_bytes)
                account_ptr = ctypes.c_char_p(account_bytes)
                script_path_ptr = ctypes.c_char_p(script_path_bytes)
                log_uuid_ptr = ctypes.c_char_p(log_uuid_bytes)

                result = self.script.execute(
                    'script_start_game',
                    self.emulator_index,
                    task_id_ptr,
                    self.emulator_info.bind_handle,
                    account_ptr,
                    self.game_config.timeout * 60 * 1000,
                    script_path_ptr,
                    GM.script_log_port,
                    log_uuid_ptr,
                    timeout=30 * 60
                )
                return True, result
            elif self.game_config.script.endswith('.py'):
                script_helper = ScriptHelper(dm=self.dm,
                                             emu=self.emu,
                                             emulator_index=self.emulator_index,
                                             top_handle=self.emulator_info.top_handle,
                                             bind_handle=self.emulator_info.bind_handle,
                                             game_id=self.game_config.task_id,
                                             game_type=self.game_config.game_type,
                                             account=self.account_info.account,
                                             password=self.account_info.password,
                                             retained=self.account_info.retained,
                                             log_uuid=self.log_uuid,
                                             log_port=self.log_port,
                                             timeout=self.game_config.timeout * 60
                                             )
                result = self.script.execute('script_start_game',script_helper, timeout=30*60)
                return True, result
            else:
                return False, "脚本文件格式错误"

        except Exception as e:
            return False, f"脚本_启动游戏,错误: {e}"

    def login_game(self):
        """登录游戏并处理可能的异常。

        Returns:
            tuple: 成功与否的标志和结果或错误信息。
        """
        try:
            if self.game_config.script.endswith('.dll'):
                # 获取字符串的字节数据
                account_bytes = self.account_info.account.encode('gbk')
                password_bytes = self.account_info.password.encode('gbk')
                wechat_order_id_bytes = self.account_info.wechat_order_id.encode('gbk')
                task_id_bytes = self.game_config.task_id.encode('gbk')
                script_path_bytes = self.script.script_path.encode('gbk')
                channel_id_bytes = self.game_config.channel_id.encode('gbk')

                # 创建 c_char_p 指针
                account_ptr = ctypes.c_char_p(account_bytes)
                password_ptr = ctypes.c_char_p(password_bytes)
                wechat_order_id_ptr = ctypes.c_char_p(wechat_order_id_bytes)
                task_id_ptr = ctypes.c_char_p(task_id_bytes)
                script_path_ptr = ctypes.c_char_p(script_path_bytes)
                channel_id_ptr = ctypes.c_char_p(channel_id_bytes)

                result = self.script.execute(
                    'script_login_game',
                    self.emulator_index,  # 窗口下标
                    self.game_config.timeout * 60 * 1000,  # 超时时间
                    self.emulator_info.bind_handle,
                    account_ptr,
                    password_ptr,
                    self.account_info.retained,
                    wechat_order_id_ptr,
                    1,
                    task_id_ptr,
                    script_path_ptr,
                    GM.script_log_port,
                    channel_id_ptr,
                    timeout=30 * 60
                )
                return True, result
            elif self.game_config.script.endswith('.py'):
                script_helper = ScriptHelper(dm=self.dm,
                                             emu=self.emu,
                                             emulator_index=self.emulator_index,
                                             top_handle=self.emulator_info.top_handle,
                                             bind_handle=self.emulator_info.bind_handle,
                                             game_id=self.game_config.task_id,
                                             game_type=self.game_config.game_type,
                                             account=self.account_info.account,
                                             password=self.account_info.password,
                                             retained=self.account_info.retained,
                                             log_uuid=self.log_uuid,
                                             log_port=self.log_port,
                                             timeout=self.game_config.timeout * 60
                                             )
                result = self.script.execute('script_login_game',script_helper, timeout=30 * 60)
                return True, result
            else:
                return False, "脚本文件格式错误"

        except Exception as e:
            return False, f'脚本_登陆游戏,错误: {e}'

    def main_task(self):
        """执行主线任务并处理可能的异常。

        Returns:
            tuple: 成功与否的标志和结果或错误信息。
        """
        try:
            if self.game_config.script.endswith('.dll'):
                # 获取字符串的字节数据
                account_bytes = self.account_info.account.encode('gbk')
                task_id_bytes = self.game_config.task_id.encode('gbk')
                script_path_bytes = self.script.script_path.encode('gbk')

                # 创建 c_char_p 指针
                account_ptr = ctypes.c_char_p(account_bytes)
                task_id_ptr = ctypes.c_char_p(task_id_bytes)
                script_path_ptr = ctypes.c_char_p(script_path_bytes)

                result = self.script.execute(
                    'script_main_task',
                    self.emulator_index,
                    self.game_config.timeout * 60 * 1000,
                    account_ptr,  # 直接传递数据地址
                    0,
                    self.emulator_info.bind_handle,
                    task_id_ptr,  # 直接传递数据地址
                    script_path_ptr,  # 直接传递数据地址
                    GM.script_log_port,
                    timeout=60 * 60
                )
                return True, result
            elif self.game_config.script.endswith('.py'):
                script_helper = ScriptHelper(dm=self.dm,
                                             emu=self.emu,
                                             emulator_index=self.emulator_index,
                                             top_handle=self.emulator_info.top_handle,
                                             bind_handle=self.emulator_info.bind_handle,
                                             game_id=self.game_config.task_id,
                                             game_type=self.game_config.game_type,
                                             account=self.account_info.account,
                                             password=self.account_info.password,
                                             retained=self.account_info.retained,
                                             log_uuid=self.log_uuid,
                                             log_port=self.log_port,
                                             timeout=self.game_config.timeout * 60
                                             )
                result = self.script.execute('script_main_task', script_helper, timeout=60 * 60)
                return True, result
            else:
                return False, "脚本文件格式错误"
        except Exception as e:
            return False, f'脚本_教程主线,错误: {e}'

    def upload_account_log(self, pc_code: str, operator: str, pc_mac: str):
        self.upload_log.set_account_info(game_id=int(self.game_config.task_id),
                                         account_type=self.account_info.game_type,
                                         pwd=self.account_info.password,
                                         account=self.account_info.account,
                                         task_type=self.account_info.retained)
        self.upload_log.set_pc_info(pc_code=pc_code, operator=operator, pc_mac=pc_mac, pc_ip='')
        self.upload_log.upload_pull_account_log(1)

    def upload_device_log(self, status: int):
        try:
            self.upload_log.set_device_info(device_id=self.emu.get_android_id(self.emulator_index),
                                            device_manufacturer=self.emu.get_manufacturer(self.emulator_index),
                                            device_model=self.emu.get_model(self.emulator_index),
                                            device_imei=self.emu.get_IMEI(self.emulator_index),
                                            device_sdk=self.emu.get_android_version(self.emulator_index),
                                            device_mac=self.emu.get_mac(self.emulator_index),
                                            device_number=self.emu.get_phone_number(self.emulator_index),
                                            script_device_id='')

            self.upload_log.set_simulator_info(simulator_code=str(self.emulator_index),
                                               simulator_mac='',
                                               simulator_ip_city='',
                                               simulator_ip=self.emu.get_net_ip(self.emulator_index)
                                               )

            ret = self.upload_log.upload_start_simulator_log(status=status)
            return ret

        except Exception as e:
            return False, f'脚本_设备信息,错误: {e}'