fix: WSL-only support and emulator detection

autowsgr/configs.py

@@ -268,25 +268,37 @@ def __post_init__(self) -> None:
         object.__setattr__(self, 'os_type', OSType.auto())
 
         # 模拟器
-        if self.emulator_name is None:
-            object.__setattr__(
-                self,
-                'emulator_name',
-                self.emulator_type.default_emulator_name(self.os_type),
-            )
-        if self.emulator_start_cmd is None:
-            object.__setattr__(
-                self,
-                'emulator_start_cmd',
-                self.emulator_type.auto_emulator_path(self.os_type),
-            )
-        assert self.emulator_start_cmd is not None
-        if self.emulator_process_name is None:
-            object.__setattr__(
-                self,
-                'emulator_process_name',
-                os.path.basename(self.emulator_start_cmd),
-            )
+        if self.os_type == OSType.linux:
+            if self.emulator_name is None:
+                raise ValueError('WSL 需要显式设置 emulator_name')
+            if self.emulator_start_cmd is None:
+                raise ValueError('WSL 需要显式设置 emulator_start_cmd')
+            if self.emulator_process_name is None:
+                object.__setattr__(
+                    self,
+                    'emulator_process_name',
+                    os.path.basename(self.emulator_start_cmd),
+                )
+        else:
+            if self.emulator_name is None:
+                object.__setattr__(
+                    self,
+                    'emulator_name',
+                    self.emulator_type.default_emulator_name(self.os_type),
+                )
+            if self.emulator_start_cmd is None:
+                object.__setattr__(
+                    self,
+                    'emulator_start_cmd',
+                    self.emulator_type.auto_emulator_path(self.os_type),
+                )
+            assert self.emulator_start_cmd is not None
+            if self.emulator_process_name is None:
+                object.__setattr__(
+                    self,
+                    'emulator_process_name',
+                    os.path.basename(self.emulator_start_cmd),
+                )
 
         # 游戏
         object.__setattr__(self, 'app_name', self.game_app.app_name)

autowsgr/timer/controllers/os_controller.py

@@ -1,6 +1,7 @@
 import json
 import os
 import re
+import shlex
 import subprocess
 import time
 from typing import Protocol
@@ -12,7 +13,7 @@
 
 from autowsgr.configs import UserConfig
 from autowsgr.constants.custom_exceptions import CriticalErr
-from autowsgr.types import EmulatorType
+from autowsgr.types import EmulatorType, OSType
 from autowsgr.utils.logger import Logger
 
 
@@ -300,3 +301,112 @@ def __get_mumu_info(self) -> dict:
         except Exception as e:
             self.logger.error(f'{cmd} {e}')
         return {}
+
+
+class LinuxController(OSController):
+    def __init__(self, config: UserConfig, logger: Logger) -> None:
+        self.logger = logger
+        self.is_wsl = OSType._is_wsl()
+
+        self.emulator_type = config.emulator_type
+        if config.emulator_name is None:
+            raise CriticalErr('WSL 需要显式设置 emulator_name')
+        if config.emulator_start_cmd is None:
+            raise CriticalErr('WSL 需要显式设置 emulator_start_cmd')
+        if config.emulator_process_name is None:
+            raise CriticalErr('WSL 需要显式设置 emulator_process_name')
+        self.emulator_name = config.emulator_name
+        self.emulator_start_cmd = config.emulator_start_cmd
+        self.emulator_process_name = config.emulator_process_name
+        self.dev_name = f'Android:///{self.emulator_name}'
+
+    def is_android_online(self) -> bool:
+        """判断 timer 给定的设备是否在线"""
+        devices = self._adb_devices()
+        if self.emulator_name in devices:
+            return True
+        if self.is_wsl:
+            return self._is_windows_process_running()
+        try:
+            subprocess.run(
+                ['pgrep', '-f', self.emulator_process_name],
+                check=True,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+            return True
+        except subprocess.CalledProcessError:
+            return False
+
+    def kill_android(self) -> None:
+        """强制终止模拟器进程"""
+        try:
+            if self.is_wsl:
+                result = subprocess.run(
+                    ['taskkill.exe', '/f', '/im', self.emulator_process_name],
+                    capture_output=True,
+                    text=True,
+                )
+                if result.returncode != 0:
+                    raise CriticalErr(result.stderr.strip() or result.stdout.strip())
+            else:
+                subprocess.run(
+                    ['pkill', '-9', '-f', self.emulator_process_name],
+                    check=True,
+                )
+            self.logger.info(f'已终止模拟器进程: {self.emulator_process_name}')
+        except Exception as e:
+            raise CriticalErr(f'终止模拟器进程失败: {e}')
+
+    def start_android(self) -> None:
+        """启动模拟器"""
+        try:
+            subprocess.Popen(self._split_command(self.emulator_start_cmd))
+            self.logger.info(f'正在启动模拟器: {self.emulator_start_cmd}')
+            start_time = time.time()
+            while not self.is_android_online():
+                time.sleep(1)
+                if time.time() - start_time > 120:
+                    raise TimeoutError('模拟器启动超时！')
+        except Exception as e:
+            raise CriticalErr(f'启动模拟器失败: {e}')
+
+    def _adb_devices(self) -> list[str]:
+        try:
+            from airtest.core.android.adb import ADB
+
+            adb = ADB().get_adb_path()
+            result = subprocess.run(
+                [adb, 'devices'],
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+        except Exception as e:
+            self.logger.error(f'adb devices 失败: {e}')
+            return []
+
+        devices = []
+        for line in result.stdout.splitlines():
+            line = line.strip()
+            if not line or line.startswith('List of devices'):
+                continue
+            parts = line.split()
+            if len(parts) >= 2 and parts[1] == 'device':
+                devices.append(parts[0])
+        return devices
+
+    def _is_windows_process_running(self) -> bool:
+        result = subprocess.run(
+            ['tasklist.exe', '/fi', f'IMAGENAME eq {self.emulator_process_name}'],
+            capture_output=True,
+            text=True,
+        )
+        output = (result.stdout or '').lower()
+        if 'no tasks' in output:
+            return False
+        return self.emulator_process_name.lower() in output
+
+    @staticmethod
+    def _split_command(command: str) -> list[str]:
+        return shlex.split(command)

autowsgr/types.py

@@ -46,7 +46,7 @@ class OcrBackend(StrEnum):
 
 class OSType(StrEnum):
     windows = 'Windows'
-    linux = 'Linux'
+    linux = 'linux'
     macos = 'macOS'
 
     @classmethod
@@ -55,8 +55,25 @@ def auto(cls) -> 'OSType':
             return OSType.windows
         if sys.platform == 'darwin':
             return OSType.macos
+        if sys.platform.startswith('linux'):
+            if cls._is_wsl():
+                return OSType.linux
+            raise ValueError('暂不支持非 WSL 的 Linux 系统')
         raise ValueError(f'不支持的操作系统 {sys.platform}')
 
+    @staticmethod
+    def _is_wsl() -> bool:
+        if os.environ.get('WSL_DISTRO_NAME') or os.environ.get('WSL_INTEROP'):
+            return True
+        for path in ('/proc/sys/kernel/osrelease', '/proc/version'):
+            try:
+                with open(path, encoding='utf-8', errors='ignore') as handle:
+                    if 'microsoft' in handle.read().lower():
+                        return True
+            except OSError:
+                continue
+        return False
+
 
 class EmulatorType(StrEnum):
     leidian = '雷电'