feat(Python): add ruff + pyright and complete type annotations

.gitignore

@@ -369,6 +369,9 @@ paket-files/
 # Python Tools for Visual Studio (PTVS)
 __pycache__/
 *.pyc
+# Python virtual environments and tool caches (e.g. source/binding/Python)
+.venv/
+.ruff_cache/
 
 # Cake - Uncomment if you are using it
 # tools/**

.vscode/settings.json

@@ -21,7 +21,10 @@
         "editor.defaultFormatter": "vscode.json-language-features"
     },
     "[python]": {
-        "editor.defaultFormatter": "ms-python.black-formatter"
+        "editor.defaultFormatter": "charliermarsh.ruff",
+        "editor.codeActionsOnSave": {
+            "source.organizeImports": "explicit"
+        }
     },
     "[yaml]": {
         "editor.insertSpaces": true,

source/binding/Python/maa/__init__.py

@@ -4,9 +4,6 @@
 from .library import Library
 
 env_path = os.environ.get("MAAFW_BINARY_PATH")
-if env_path:
-    __PATH = Path(env_path)
-else:
-    __PATH = Path(Path(__file__).parent, "bin")
+path = Path(env_path) if env_path else Path(Path(__file__).parent, "bin")
 
-Library.open(__PATH, agent_server=False)
+Library.open(path, agent_server=False)

source/binding/Python/maa/agent/__init__.py

@@ -4,9 +4,6 @@
 from ..library import Library
 
 env_path = os.environ.get("MAAFW_BINARY_PATH")
-if env_path:
-    __PATH = Path(env_path)
-else:
-    __PATH = Path(Path(__file__).parent.parent, "bin")
+path = Path(env_path) if env_path else Path(Path(__file__).parent.parent, "bin")
 
-Library.open(__PATH, agent_server=True)
+Library.open(path, agent_server=True)

source/binding/Python/maa/agent/agent_server.py

@@ -1,8 +1,17 @@
 import ctypes
+from typing import TYPE_CHECKING, Callable
 
 from ..define import *
-from ..library import Library
 from ..event_sink import EventSink
+from ..library import Library
+
+if TYPE_CHECKING:
+    from ..context import ContextEventSink
+    from ..controller import ControllerEventSink
+    from ..custom_action import CustomAction
+    from ..custom_recognition import CustomRecognition
+    from ..resource import ResourceEventSink
+    from ..tasker import TaskerEventSink
 
 
 class AgentServer:
@@ -24,29 +33,33 @@ def analyze(self, context, argv):
     """
 
     @staticmethod
-    def custom_recognition(name: str):
+    def custom_recognition(
+        name: str,
+    ) -> Callable[[type["CustomRecognition"]], type["CustomRecognition"]]:
         """自定义识别器装饰器 / Custom recognition decorator
 
         Args:
-            name: 识别器名称，需与 Pipeline 中的 custom_recognition 字段匹配 / Recognition name, should match the custom_recognition field in Pipeline
+            name: 识别器名称，需与 Pipeline 中的 custom_recognition 字段匹配
+            Recognition name, should match the custom_recognition field in Pipeline
 
         Returns:
             装饰器函数 / Decorator function
         """
 
-        def wrapper_recognition(recognition):
-            AgentServer.register_custom_recognition(
-                name=name, recognition=recognition()
-            )
+        def wrapper_recognition(
+            recognition: type["CustomRecognition"],
+        ) -> type["CustomRecognition"]:
+            AgentServer.register_custom_recognition(name=name, recognition=recognition())
             return recognition
 
         return wrapper_recognition
 
-    _custom_recognition_holder = {}
+    _custom_recognition_holder: dict[str, "CustomRecognition"] = {}
 
     @staticmethod
     def register_custom_recognition(
-        name: str, recognition: "CustomRecognition"  # type: ignore
+        name: str,
+        recognition: "CustomRecognition",
     ) -> bool:
         """注册自定义识别器 / Register custom recognition
 
@@ -72,26 +85,29 @@ def register_custom_recognition(
         )
 
     @staticmethod
-    def custom_action(name: str):
+    def custom_action(
+        name: str,
+    ) -> Callable[[type["CustomAction"]], type["CustomAction"]]:
         """自定义动作装饰器 / Custom action decorator
 
         Args:
-            name: 动作名称，需与 Pipeline 中的 custom_action 字段匹配 / Action name, should match the custom_action field in Pipeline
+            name: 动作名称，需与 Pipeline 中的 custom_action 字段匹配
+            Action name, should match the custom_action field in Pipeline
 
         Returns:
             装饰器函数 / Decorator function
         """
 
-        def wrapper_action(action):
+        def wrapper_action(action: type["CustomAction"]) -> type["CustomAction"]:
             AgentServer.register_custom_action(name=name, action=action())
             return action
 
         return wrapper_action
 
-    _custom_action_holder = {}
+    _custom_action_holder: dict[str, "CustomAction"] = {}
 
     @staticmethod
-    def register_custom_action(name: str, action: "CustomAction") -> bool:  # type: ignore
+    def register_custom_action(name: str, action: "CustomAction") -> bool:
         """注册自定义动作 / Register custom action
 
         Args:
@@ -120,7 +136,8 @@ def start_up(identifier: str) -> bool:
         """启动 Agent 服务 / Start Agent service
 
         Args:
-            identifier: 连接标识符，用于与 AgentClient 匹配 / Connection identifier for matching with AgentClient
+            identifier: 连接标识符，用于与 AgentClient 匹配
+            Connection identifier for matching with AgentClient
 
         Returns:
             bool: 是否成功 / Whether successful
@@ -162,17 +179,19 @@ def detach() -> None:
 
         Library.agent_server().MaaAgentServerDetach()
 
-    _sink_holder: Dict[int, "EventSink"] = {}
+    _sink_holder: dict[int, "EventSink"] = {}
 
     @staticmethod
-    def resource_sink():
+    def resource_sink() -> Callable[[type["ResourceEventSink"]], type["ResourceEventSink"]]:
         """资源事件监听器装饰器 / Resource event sink decorator
 
         Returns:
             装饰器函数 / Decorator function
         """
 
-        def wrapper_sink(sink):
+        def wrapper_sink(
+            sink: type["ResourceEventSink"],
+        ) -> type["ResourceEventSink"]:
             AgentServer.add_resource_sink(sink=sink())
             return sink
 
@@ -185,25 +204,23 @@ def add_resource_sink(sink: "ResourceEventSink") -> None:
         Args:
             sink: 资源事件监听器 / Resource event sink
         """
-        sink_id = int(
-            Library.agent_server().MaaAgentServerAddResourceSink(
-                *EventSink._gen_c_param(sink)
-            )
-        )
+        sink_id = int(Library.agent_server().MaaAgentServerAddResourceSink(*EventSink._gen_c_param(sink)))
         if sink_id == MaaInvalidId:
             return None
 
         AgentServer._sink_holder[sink_id] = sink
 
     @staticmethod
-    def controller_sink():
+    def controller_sink() -> Callable[[type["ControllerEventSink"]], type["ControllerEventSink"]]:
         """控制器事件监听器装饰器 / Controller event sink decorator
 
         Returns:
             装饰器函数 / Decorator function
         """
 
-        def wrapper_sink(sink):
+        def wrapper_sink(
+            sink: type["ControllerEventSink"],
+        ) -> type["ControllerEventSink"]:
             AgentServer.add_controller_sink(sink=sink())
             return sink
 
@@ -216,25 +233,21 @@ def add_controller_sink(sink: "ControllerEventSink") -> None:
         Args:
             sink: 控制器事件监听器 / Controller event sink
         """
-        sink_id = int(
-            Library.agent_server().MaaAgentServerAddControllerSink(
-                *EventSink._gen_c_param(sink)
-            )
-        )
+        sink_id = int(Library.agent_server().MaaAgentServerAddControllerSink(*EventSink._gen_c_param(sink)))
         if sink_id == MaaInvalidId:
             return None
 
         AgentServer._sink_holder[sink_id] = sink
 
     @staticmethod
-    def tasker_sink():
+    def tasker_sink() -> Callable[[type["TaskerEventSink"]], type["TaskerEventSink"]]:
         """任务器事件监听器装饰器 / Tasker event sink decorator
 
         Returns:
             装饰器函数 / Decorator function
         """
 
-        def wrapper_sink(sink):
+        def wrapper_sink(sink: type["TaskerEventSink"]) -> type["TaskerEventSink"]:
             AgentServer.add_tasker_sink(sink=sink())
             return sink
 
@@ -247,25 +260,21 @@ def add_tasker_sink(sink: "TaskerEventSink") -> None:
         Args:
             sink: 任务器事件监听器 / Tasker event sink
         """
-        sink_id = int(
-            Library.agent_server().MaaAgentServerAddTaskerSink(
-                *EventSink._gen_c_param(sink)
-            )
-        )
+        sink_id = int(Library.agent_server().MaaAgentServerAddTaskerSink(*EventSink._gen_c_param(sink)))
         if sink_id == MaaInvalidId:
             return None
 
         AgentServer._sink_holder[sink_id] = sink
 
     @staticmethod
-    def context_sink():
+    def context_sink() -> Callable[[type["ContextEventSink"]], type["ContextEventSink"]]:
         """上下文事件监听器装饰器 / Context event sink decorator
 
         Returns:
             装饰器函数 / Decorator function
         """
 
-        def wrapper_sink(sink):
+        def wrapper_sink(sink: type["ContextEventSink"]) -> type["ContextEventSink"]:
             AgentServer.add_context_sink(sink=sink())
             return sink
 
@@ -278,11 +287,7 @@ def add_context_sink(sink: "ContextEventSink") -> None:
         Args:
             sink: 上下文事件监听器 / Context event sink
         """
-        sink_id = int(
-            Library.agent_server().MaaAgentServerAddContextSink(
-                *EventSink._gen_c_param(sink)
-            )
-        )
+        sink_id = int(Library.agent_server().MaaAgentServerAddContextSink(*EventSink._gen_c_param(sink)))
         if sink_id == MaaInvalidId:
             return None