pythonTest/MuMu.py

import subprocess
import json
import time
import os
import tkinter as tk

class MuMuEmulatorManager:
    def __init__(self, manager_path=r"D:\MuMuPlayer\nx_main\MuMuManager.exe"):
        self.manager_path = manager_path
        if not os.path.exists(manager_path):
            raise FileNotFoundError(f"找不到 MuMuManager.exe: {manager_path}")
    
    def get_emulator_list(self):
        """获取所有模拟器列表"""
        cmd = [self.manager_path, "info", "-v", "all"]
        result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8')
        
        if result.returncode != 0:
            print(f"获取列表失败: {result.stderr}")
            return []
        
        try:
            data = json.loads(result.stdout)
            emulators = []
            for key, value in data.items():
                if isinstance(value, dict):
                    value['index'] = key
                    emulators.append(value)
            return emulators
        except json.JSONDecodeError as e:
            print(f"JSON解析失败: {e}")
            return []
    
    def start_emulator(self, index):
        """启动指定索引的模拟器 - 使用正确的 launch 命令"""
        print(f"正在启动模拟器 {index}...")
        
        # 正确的命令格式：control -v <index> launch
        cmd = [self.manager_path, "control", "-v", str(index), "launch"]
        print(f"执行命令: {' '.join(cmd)}")
        
        result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8')
        
        if result.returncode == 0:
            print(f"✅ 模拟器 {index} 启动命令已发送")
            return True
        else:
            print(f"❌ 启动失败")
            if result.stderr:
                print(f"错误信息: {result.stderr}")
            if result.stdout:
                print(f"输出: {result.stdout}")
            return False
    
    def wait_for_emulator_ready(self, index, timeout=120, check_interval=3):
        """等待模拟器启动完成（Android 系统就绪）"""
        print(f"\n等待模拟器 {index} 启动完成...")
        start_time = time.time()
        
        while time.time() - start_time < timeout:
            # 获取模拟器状态
            cmd = [self.manager_path, "info", "-v", str(index)]
            result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8')
            
            if result.returncode == 0:
                try:
                    data = json.loads(result.stdout)
                    # 检查 Android 是否已启动
                    if data.get('is_android_started') == True:
                        elapsed = time.time() - start_time
                        print(f"✅ 模拟器 {index} 已就绪！耗时: {elapsed:.1f} 秒")
                        return True
                    elif data.get('is_process_started') == True:
                        print(f"⏳ 进程已启动，等待 Android 系统... ({int(time.time() - start_time)}秒)")
                    else:
                        print(f"⏳ 等待进程启动... ({int(time.time() - start_time)}秒)")
                except:
                    pass
            
            time.sleep(check_interval)
        
        print(f"❌ 超时！模拟器 {index} 在 {timeout} 秒内未就绪")
        return False
    
    def get_adb_port(self, index):
        """获取模拟器的 ADB 端口"""
        # 通过 MuMuManager 获取
        cmd = [self.manager_path, "info", "-v", str(index)]
        result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8')
        
        if result.returncode == 0:
            try:
                data = json.loads(result.stdout)
                # 注意：从之前的输出看，info 命令没有直接返回 adb_port
                # 可能需要通过其他方式获取
                pass
            except:
                pass
        
        # 使用 MuMu 默认端口规律：16384 + 32 * index
        default_port = 16384 + 32 * int(index)
        print(f"估算 ADB 端口: {default_port}")
        return default_port
    
    def install_apk(self, index, apk_path):
        """通过 ADB 安装 APK 到模拟器"""
        # 检查 APK 是否存在
        if not os.path.exists(apk_path):
            print(f"❌ 错误：APK 文件不存在: {apk_path}")
            return False
        
        # 获取 ADB 端口
        adb_port = self.get_adb_port(index)
        target_device = f"127.0.0.1:{adb_port}"
        
        # 连接 ADB
        print(f"\n连接 ADB {target_device}...")
        connect_result = subprocess.run(f"adb connect {target_device}", 
                                    shell=True, capture_output=True, text=True)
        print(connect_result.stdout.strip())
        
        # 等待连接稳定
        time.sleep(2)
        
        # 检查设备连接
        devices_result = subprocess.run("adb devices", shell=True, capture_output=True, text=True)
        print(f"\n当前设备列表:\n{devices_result.stdout}")
        
        # 检查是否已安装 com.dragon.read
        print(f"\n检查是否已安装 com.dragon.read...")
        check_cmd = f"adb -s {target_device} shell pm list packages | findstr \"com.dragon.read\""
        check_result = subprocess.run(check_cmd, shell=True, capture_output=True, text=True)
        
        if "com.dragon.read" in check_result.stdout:
            print("✅ com.dragon.read 已安装，跳过安装步骤")
            return True
        
        # 安装 APK
        print(f"\n正在安装 APK: {os.path.basename(apk_path)}")
        install_cmd = f"adb -s {target_device} install -r \"{apk_path}\""
        result = subprocess.run(install_cmd, shell=True, capture_output=True, text=True)
        
        if "Success" in result.stdout:
            print("✅ APK 安装成功！")
            return True
        else:
            print("❌ APK 安装失败")
            print(f"输出: {result.stdout}")
            if result.stderr:
                print(f"错误: {result.stderr}")
            return False
    
    def display_emulator_list(self):
        """显示模拟器列表"""
        emulators = self.get_emulator_list()
        
        if not emulators:
            print("未找到任何模拟器")
            return
        
        print(f"\n{'='*60}")
        print(f"找到 {len(emulators)} 个模拟器:")
        print(f"{'='*60}\n")
        
        for i, emu in enumerate(emulators, 1):
            status = "🟢 运行中" if emu.get('is_process_started') else "🔴 未运行"
            android_status = "✅ 已启动" if emu.get('is_android_started') else "⏸️ 未启动"
            print(f"{i}. 索引 {emu.get('index')}: {emu.get('name')}")
            print(f"   进程状态: {status}")
            print(f"   Android: {android_status}")
            print()
    def open_app(self, index, package_name="com.dragon.read"):
        """打开指定包名的应用"""
        # 获取 ADB 端口
        adb_port = self.get_adb_port(index)
        target_device = f"127.0.0.1:{adb_port}"
        
        # 连接 ADB
        print(f"\n连接 ADB {target_device}...")
        subprocess.run(f"adb connect {target_device}", shell=True, capture_output=True)
        time.sleep(1)
        
        # 打开应用（使用 monkey 命令，最通用）
        print(f"\n正在打开应用: {package_name}")
        cmd = f"adb -s {target_device} shell monkey -p {package_name} -c android.intent.category.LAUNCHER 1"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        
        if "Events injected" in result.stdout or result.returncode == 0:
            print(f"✅ 应用已打开: {package_name}")
            return True
        else:
            print(f"❌ 打开失败")
            print(f"输出: {result.stdout}")
            if result.stderr:
                print(f"错误: {result.stderr}")
            return False
    def paste_text(self, index, text):
        """将文字放到系统剪贴板并粘贴"""
        # 获取 ADB 端口
        adb_port = self.get_adb_port(index)
        target_device = f"127.0.0.1:{adb_port}"
        
        # 连接 ADB
        subprocess.run(f"adb connect {target_device}", shell=True, capture_output=True)
        time.sleep(0.5)
        # 1. 复制到电脑剪贴板
        """使用 Windows clip 命令复制到剪贴板"""
        root = tk.Tk()
        root.withdraw()
        root.clipboard_clear()
        root.clipboard_append(text)
        root.update()
        root.destroy()
        # 将文字设置到剪贴板
        
        escape_text = text.replace('"', '\\"').replace("'", "\\'")
        set_clipboard_cmd = f'adb -s {target_device} shell am broadcast -a clipper.set -e text "{escape_text}"'
        subprocess.run(set_clipboard_cmd, shell=True, capture_output=True)
        time.sleep(0.3)
        
        # 执行粘贴 (Ctrl+V)
        subprocess.run(f"adb -s {target_device} shell input keyevent 279", shell=True)
        
        print(f"✅ 已粘贴: {text[:50]}{'...' if len(text) > 50 else ''}")
        return True
    def tap(self, index, x, y):
        """点击指定坐标"""
        # 获取 ADB 端口
        adb_port = self.get_adb_port(index)
        target_device = f"127.0.0.1:{adb_port}"
        
        # 连接 ADB
        subprocess.run(f"adb connect {target_device}", shell=True, capture_output=True)
        
        # 点击坐标
        cmd = f"adb -s {target_device} shell input tap {x} {y}"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        
        if result.returncode == 0:
            print(f"✅ 点击坐标 ({x}, {y})")
            return True
        else:
            print(f"❌ 点击失败: {result.stderr}")
            return False
def main():
    # 配置
    MANAGER_PATH = r"D:\MuMuPlayer\nx_main\MuMuManager.exe"
    APK_PATH = r"fanqie.apk"  # 请修改为你的 APK 实际路径
    TARGET_INDEX = 1  # 第二个模拟器的索引（索引从0开始，1是第二个）
    
    # 创建管理器实例
    try:
        manager = MuMuEmulatorManager(MANAGER_PATH)
    except FileNotFoundError as e:
        print(e)
        return
    
    # 1. 显示所有模拟器列表
    print("步骤1: 获取模拟器列表")
    manager.display_emulator_list()
    
    # 2. 启动第二个模拟器（索引为1）
    print(f"\n步骤2: 启动模拟器 {TARGET_INDEX}")
    if not manager.start_emulator(TARGET_INDEX):
        print("启动失败，退出")
        return
    
    # 3. 等待模拟器启动完成
    print(f"\n步骤3: 等待模拟器就绪")
    if not manager.wait_for_emulator_ready(TARGET_INDEX, timeout=180):
        print("模拟器未能在规定时间内就绪，退出")
        return
    time.sleep(5)
    # 4. 安装 APK
    print(f"\n步骤4: 安装 APK")
    if manager.install_apk(TARGET_INDEX, APK_PATH):
        print("\n🎉 全部步骤完成！")
        manager.open_app(TARGET_INDEX, package_name="com.dragon.read")
        time.sleep(40)
        manager.tap(TARGET_INDEX, 390, 90)  # 点击输入框坐标（示例）
        time.sleep(2)
        manager.paste_text(TARGET_INDEX, "玄幻战神：开局就得到大佬的守护")
        time.sleep(2)
        manager.tap(TARGET_INDEX, 655, 92)
        # 进入数目
        time.sleep(5)
        manager.tap(TARGET_INDEX, 355, 333)
        ind = 0
        while ind < 10:
            time.sleep(30)
            manager.tap(TARGET_INDEX, 710, 632)
            ind += 1
        # 评价
        time.sleep(2)
        manager.tap(TARGET_INDEX, 360, 690)
        time.sleep(2)
        manager.tap(TARGET_INDEX, 680, 95)
        time.sleep(2)
        manager.tap(TARGET_INDEX, 633, 930)
        # 发表
        time.sleep(2)
        manager.tap(TARGET_INDEX, 640, 95)
    else:
        print("\n❌ 运行失败")

if __name__ == "__main__":
    main()