#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ MuMu模拟器管理模块 """ import subprocess import os import time import configparser import shutil import re import requests import logging import json import sys # 注意：日志配置现在由统一的 logging_config 模块管理 # 配置日志 # logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class MuMuManager: def __init__(self, config, vm_index=None): self.config = config # 优先从 mumu 子配置读取，兼容旧的扁平键名 mumu = config.get('mumu', {}) if isinstance(config, dict) else {} self.vm_index = ( vm_index or mumu.get('default_vm_index') or config.get('DEFAULT_VM_INDEX', 2) ) self.mumu_manager_path = ( mumu.get('manager_path') or config.get('MUMU_MANAGER_PATH') ) self.clean_backup_path = ( mumu.get('clean_backup') or config.get('CLEAN_BACKUP_PATH') or config.get('shared_paths', {}).get('clean_backup') ) self.local_import_dir = ( mumu.get('local_import_dir') or config.get('LOCAL_IMPORT_DIR') ) self.max_vm_index = ( mumu.get('max_vm_index') or config.get('MAX_VM_INDEX', 10) ) self.net_bridge_card = ( mumu.get('bridge_card') or config.get('NET_BRIDGE_CARD', 'Realtek PCIe GbE Family Controller') ) # 从 MuMuManager.exe 路径推导工作目录，回退到 C:\ if self.mumu_manager_path and os.path.exists(os.path.dirname(self.mumu_manager_path)): self._cwd = os.path.dirname(self.mumu_manager_path) else: self._cwd = 'C:\\' @property def device_type(self) -> str: return 'emulator' def get_active_network_adapter(self): """获取第一个有网的网卡名称""" try: output = subprocess.check_output( ['wmic', 'nic', 'where', 'NetConnectionStatus=2', 'get', 'Name', '/value'], encoding='utf-8', errors='ignore', timeout=10 ) for line in output.strip().split('\n'): if line.startswith('Name='): adapter_name = line.split('=', 1)[1].strip() if adapter_name: logger.info(f"检测到活动网卡: {adapter_name}") return adapter_name except Exception as e: logger.warning(f"获取网卡信息失败: {e}") # 如果获取失败，返回默认值 default_adapter = self.config.get('NET_BRIDGE_CARD', 'Realtek PCIe GbE Family Controller') logger.warning(f"无法获取活动网卡，使用默认值: {default_adapter}") return default_adapter def run_cmd(self, command, use_index=True): """执行 MuMuManager 命令""" # 路径需要用引号包裹，防止Windows路径中的空格导致命令解析错误 cmd = f'"{self.mumu_manager_path}" {command}' + (f" -v {self.vm_index}" if use_index else "") logger.info(f"执行: {cmd}") # 使用 utf-8 编码并忽略无法解码的字符，防止在 Windows GBK 环境下崩溃 res = subprocess.run(cmd, shell=True, capture_output=True, encoding='utf-8', errors='ignore', cwd=self._cwd, timeout=60) if res.returncode != 0: logger.error(f"命令失败: {res.stdout.strip()}\n {res.stderr.strip()}") return res.returncode == 0 def force_cleanup(self): """清理残留进程""" logger.info("清理 MuMu 进程...") for proc in ['MuMuPlayer.exe', 'MuMuNxDevice.exe', 'MuMuNxMain.exe']: # 统一使用 encoding 和 errors 参数 subprocess.run(f'taskkill /f /im {proc}', shell=True, capture_output=True, encoding='utf-8', errors='ignore') time.sleep(10) def get_player_state(self) -> str: """ 获取模拟器播放器状态 :return: 状态字符串 - 'starting_rom': 模拟器正在启动 - 'start_finished': 模拟器运行正常 - '': 空字符串表示模拟器卡死或未响应 - 'FROZEN': 检测到错误输出（如 result=-3），表示卡死 - None: 无法获取状态（命令执行失败或超时） """ try: cmd = f'"{self.mumu_manager_path}" api player_state -v {self.vm_index}' logger.info(f"检查模拟器状态: {cmd}") result = subprocess.run( cmd, shell=True, capture_output=True, encoding='utf-8', errors='ignore', cwd=self._cwd, timeout=30 ) output = result.stdout.strip() logger.debug(f"模拟器状态输出: {output}") # 检测错误输出，如 "check player state fail: result=-3" if 'fail' in output.lower() or 'result=-' in output: logger.warning(f"模拟器状态检测失败: {output}") return 'FROZEN' # 返回特殊标记表示卡死 # 解析 state=xxx 格式的输出 # 例如: "check player state: state=start_finished" match = re.search(r'state=([\w]*)', output) if match: state = match.group(1) logger.info(f"模拟器状态: {state if state else '(空/卡死)'}") return state logger.warning(f"无法解析模拟器状态输出: {output}") return None except subprocess.TimeoutExpired: logger.error("获取模拟器状态超时，视为卡死") return 'FROZEN' # 超时也视为卡死 except Exception as e: logger.error(f"获取模拟器状态失败: {e}") return None def is_player_frozen(self) -> bool: """ 检查模拟器是否处于卡死状态 :return: True 表示卡死，False 表示正常或正在启动 以下情况判定为卡死： - state='' (空字符串) - state='FROZEN' (检测到错误输出或超时) - state=None (命令执行失败) """ state = self.get_player_state() # 空字符串、FROZEN 标记、None 都视为卡死 frozen = state in ('', 'FROZEN', None) if frozen: logger.warning(f"判定模拟器卡死 (state={repr(state)})") return frozen def launch_player(self) -> bool: """ 启动模拟器播放器（不执行 shutdown） :return: 成功返回 True，失败返回 False """ logger.info("启动模拟器...") cmd = f'"{self.mumu_manager_path}" control -v {self.vm_index} launch' logger.info(f"执行: {cmd}") result = subprocess.run( cmd, shell=True, capture_output=True, encoding='utf-8', errors='ignore', cwd=self._cwd, timeout=60, ) if result.returncode == 0: # 等待启动完成并获取 IP for _ in range(10): time.sleep(15) ip = self.get_adb_ip() if ip: logger.info(f"获取到模拟器 IP: {ip}，尝试 ADB 连接...") # 执行 adb connect 连接设备 try: connect_result = subprocess.run( ['adb', 'connect', ip], capture_output=True, encoding='utf-8', errors='ignore', timeout=10 ) logger.info(f"ADB connect 输出: {connect_result.stdout.strip()}") # 验证连接是否成功 verify_result = subprocess.run( ['adb', '-s', ip, 'shell', 'getprop', 'ro.build.version.sdk'], capture_output=True, encoding='utf-8', errors='ignore', timeout=10 ) if verify_result.returncode == 0 and verify_result.stdout.strip(): logger.info(f"模拟器启动成功，ADB 连接验证通过: {ip} (SDK: {verify_result.stdout.strip()})") return True else: logger.warning(f"ADB 连接验证失败，继续等待...") except subprocess.TimeoutExpired: logger.warning(f"ADB 连接超时，继续等待...") except Exception as e: logger.warning(f"ADB 连接异常: {e}，继续等待...") else: logger.error(f"命令失败: {result.stdout.strip()}\n {result.stderr.strip()}") logger.error("模拟器启动失败") return False def recover_from_frozen(self) -> bool: """ 从卡死状态恢复模拟器 使用 force_cleanup() 强杀进程后重新启动 :return: 成功返回 True，失败返回 False """ logger.warning("检测到模拟器卡死，执行强制恢复...") self.force_cleanup() return self.launch_player() def get_adb_ip(self): """获取 ADB IP""" # 明确指定编码为 utf-8 并忽略错误，解决 Windows 下的 UnicodeDecodeError output = subprocess.check_output([self.mumu_manager_path, "info", "-v", str(self.vm_index)], encoding='utf-8', errors='ignore', timeout=30) match = re.search(r'"adb_host_ip":\s*"([^"]+)"', output) return match.group(1) if match else None def prepare_import_file(self): """同步备份文件到本地目录""" if not self.local_import_dir: # 如果没配置，回退到桌面 self.local_import_dir = os.path.join(os.path.expanduser("~"), "Desktop") if not os.path.exists(self.local_import_dir): os.makedirs(self.local_import_dir, exist_ok=True) target = os.path.join(self.local_import_dir, os.path.basename(self.clean_backup_path)) # 检查源文件是否存在 if not os.path.exists(self.clean_backup_path): error_msg = f"找不到源备份文件: {self.clean_backup_path}" logger.error(error_msg) raise FileNotFoundError(error_msg) if os.path.exists(target) and os.path.getsize(target) == os.path.getsize(self.clean_backup_path): logger.info(f"备份文件已存在且大小一致，跳过拷贝: {target}") return target logger.info(f"开始拷贝备份文件: {self.clean_backup_path} -> {target}") try: shutil.copy2(self.clean_backup_path, target) except Exception as e: error_msg = f"拷贝备份文件失败: {e}" logger.error(error_msg) raise return target def check_status(self) -> bool: """ 检查模拟器是否正常运行，并验证 ADB 连接是否可用 :return: 如果模拟器运行正常且 ADB 连接可用返回True，否则返回False """ try: # 1. 检查模拟器进程是否存在 ip = self.get_adb_ip() if not ip: logger.warning("无法获取模拟器ADB IP，模拟器可能未运行") return False # 2. 验证 ADB 实际连接（关键修复） logger.debug(f"检测到模拟器 IP: {ip}，验证 ADB 连接...") try: # 尝试连接设备 result = subprocess.run( ['adb', 'connect', ip], capture_output=True, encoding='utf-8', errors='ignore', timeout=10 ) # 执行测试命令验证设备确实可用 result = subprocess.run( ['adb', '-s', ip, 'shell', 'getprop', 'ro.build.version.sdk'], capture_output=True, encoding='utf-8', errors='ignore', timeout=5 ) if result.returncode == 0 and result.stdout.strip(): logger.debug(f"模拟器 ADB 连接验证成功: {ip} (SDK: {result.stdout.strip()})") return True else: logger.warning(f"ADB 连接验证失败: {ip}，设备未响应") return False except subprocess.TimeoutExpired: logger.warning(f"ADB 连接验证超时: {ip}") return False except Exception as adb_err: logger.warning(f"ADB 连接验证异常: {ip} - {adb_err}") return False except Exception as e: logger.warning(f"检查模拟器状态失败: {e}") return False def ensure_running(self) -> tuple: """ 确保模拟器正在运行，如果未运行则尝试启动 :return: (success, restarted) - success表示模拟器是否正常运行，restarted表示是否执行了重启 """ if self.check_status(): return (True, False) # 正常运行，未重启 logger.warning("模拟器未运行或 ADB 连接失败，检查模拟器状态...") # 检测是否卡死，选择恢复策略 if self.is_player_frozen(): # 模拟器卡死：使用 force_cleanup + launch_player logger.warning("检测到模拟器卡死状态 (state=空)，执行强制恢复...") success = self.recover_from_frozen() else: # 模拟器未卡死：使用常规重启 logger.info("模拟器状态正常，执行常规重启...") success = self.restart() return (success, True) # 返回重启结果和重启标志 def restart(self): """重启模拟器""" logger.info("正在重启模拟器...") cmd = f'"{self.mumu_manager_path}" control --vmindex {self.vm_index} restart' logger.info(f"执行: {cmd}") result = subprocess.run( cmd, shell=True, capture_output=True, encoding='utf-8', errors='ignore', cwd=self._cwd, timeout=120, ) if result.returncode != 0: logger.error(f"模拟器重启命令失败: {result.stdout.strip()}\n {result.stderr.strip()}") return False for _ in range(18): time.sleep(10) if self.check_status(): ip = self.get_adb_ip() logger.info(f"模拟器重启成功，ADB IP: {ip}") return True logger.error("模拟器重启失败：等待超时") return False def recover(self): """执行恢复流程""" import_path = self.prepare_import_file() for attempt in range(1, 11): logger.info(f"尝试恢复 #{attempt}") # self.force_cleanup() # 注释强杀，避免文件系统损坏 # 1. 重置环境 self.run_cmd('api shutdown_player') time.sleep(15) self.run_cmd('delete') time.sleep(15) for i in range(2, self.max_vm_index + 1): self.run_cmd(f'delete -v {i}') time.sleep(15) # 2. 导入与配置 if not self.run_cmd(f'import -p "{import_path}"'): continue time.sleep(15) net_adapter = self.get_active_network_adapter() self.run_cmd(f'setting -k "net_bridge_card" -val "{net_adapter}"') time.sleep(15) # 3. 启动并校验 if self.run_cmd('api launch_player'): for _ in range(10): time.sleep(15) ip = self.get_adb_ip() if ip: logger.info("恢复成功！") self.run_cmd('api shutdown_player') return True logger.warning(f"第 {attempt} 次失败") return False if __name__ == '__main__': sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../..")) from config_loader import load_config config = load_config() mumu_controller = MuMuManager(config) net_adapter = mumu_controller.get_active_network_adapter() print(net_adapter)