add task_type python_code & fix bug: mcdr_command task can't be able to process vars successfully

Author: OptiJava

.gitignore

@@ -1,2 +1,4 @@
 .idea/
-debug.sh
+debug.sh
+
+test/.pytest_cache/

hooks/__init__.py

@@ -10,6 +10,9 @@
 from mcdreforged.api.utils import serializer
 from ruamel import yaml
 
+from hooks import utils
+from hooks.utils import process_arg_server
+
 scripts_folder: str = ''
 
 
@@ -40,6 +43,8 @@ class TaskType(Enum):
     shell_command = 'shell_command'
     server_command = 'server_command'
     mcdr_command = 'mcdr_command'
+    
+    python_code = 'python_code'
 
 
 class Task(Serializable):
@@ -56,6 +61,12 @@ def execute_task(self, server: PluginServerInterface, hook: str, var_dict: dict
 
         start_time = time.time()
 
+        if self.command is None:
+            server.logger.error(
+                f'Task state is not correct! Task: {self.name} Hooks: {hook} TaskType: {self.task_type} '
+                f'command: {self.command}')
+            return
+        
         if self.task_type == TaskType.undefined:
             server.logger.error(
                 f'Task state is not correct! Task: {self.name} Hooks: {hook} TaskType: {self.task_type} '
@@ -94,7 +105,32 @@ def execute_task(self, server: PluginServerInterface, hook: str, var_dict: dict
                 for key in var_dict.keys():
                     command = command.replace('{$' + key + '}', str(var_dict.get(key)))
 
-            server.execute_command(self.command)
+            server.execute_command(command)
+            
+        elif self.task_type == TaskType.python_code:
+            # 注入变量
+            finally_command = self.command
+            
+            # 要被注入的赋值语句
+            command_input = StringIO()
+            
+            if var_dict is not None:
+                for key in var_dict.keys():
+                    command_input.write(key)
+                    command_input.write('=')
+                    
+                    var_value = var_dict.get(key)
+                    
+                    if (not utils.is_int_var(var_value)) and (not utils.is_list_var(var_value)) and (not utils.is_dict_var(var_value)):
+                        var_value = '"' + str(var_value) + '"'
+                    
+                    command_input.write(str(var_value))
+                    command_input.write(';')
+            command_input.write('\n')
+            
+            finally_command = finally_command.replace('# variable injection here', command_input.getvalue())
+            
+            exec(finally_command)
 
         server.logger.debug(f'Task finished, name: {self.name}, task_type: {self.task_type}, command: {self.command}, '
                             f'costs {time.time() - start_time} seconds.')
@@ -370,23 +406,6 @@ def list_all_files(root_dir) -> list[str]:
         parse_and_apply_scripts(script, server)
 
 
-def process_arg_server(server: PluginServerInterface) -> PluginServerInterface:
-    server.func_is_server_running = server.is_server_running()
-    server.func_is_server_startup = server.is_server_startup()
-    server.func_is_rcon_running = server.is_rcon_running()
-    server.func_get_server_pid = server.get_server_pid()
-    server.func_get_server_pid_all = server.get_server_pid_all()
-    server.func_get_server_information = str(serialize(server.get_server_information()))
-    server.func_get_data_folder = server.get_data_folder()
-    server.func_get_plugin_file_path = server.get_plugin_file_path('hooks')
-    server.func_get_plugin_list = str(server.get_plugin_list())
-    server.func_get_unloaded_plugin_list = str(server.get_unloaded_plugin_list())
-    server.func_get_disabled_plugin_list = str(server.get_disabled_plugin_list())
-    server.func_get_mcdr_language = server.get_mcdr_language()
-    server.func_get_mcdr_config = str(server.get_mcdr_config())
-    return server
-
-
 def on_load(server: PluginServerInterface, old_module):
     global config, scripts_folder, temp_config
 

hooks/utils.py

@@ -0,0 +1,52 @@
+from typing import Any
+
+from mcdreforged.plugin.server_interface import PluginServerInterface
+from mcdreforged.utils.serializer import serialize
+
+
+def is_int_var(obj: Any) -> bool:
+    if type(obj) == int:
+        return True
+    else:
+        try:
+            int(str(obj))
+        except ValueError:
+            return False
+        return True
+
+
+def is_dict_var(obj: Any) -> bool:
+    if type(obj) == dict:
+        return True
+    else:
+        try:
+            return isinstance(eval(str(obj)), dict)
+        except:
+            return False
+
+
+def is_list_var(obj: Any) -> bool:
+    if type(obj) == list:
+        return True
+    else:
+        try:
+            return isinstance(eval(str(obj)), list)
+        except:
+            return False
+
+
+def process_arg_server(server: PluginServerInterface) -> PluginServerInterface:
+    server.func_is_server_running = server.is_server_running()
+    server.func_is_server_startup = server.is_server_startup()
+    server.func_is_rcon_running = server.is_rcon_running()
+    server.func_get_server_pid = server.get_server_pid()
+    server.func_get_server_pid_all = server.get_server_pid_all()
+    server.func_get_server_information = str(serialize(server.get_server_information()))
+    server.func_get_data_folder = server.get_data_folder()
+    server.func_get_plugin_file_path = server.get_plugin_file_path('hooks')
+    server.func_get_plugin_list = str(server.get_plugin_list())
+    server.func_get_unloaded_plugin_list = str(server.get_unloaded_plugin_list())
+    server.func_get_disabled_plugin_list = str(server.get_disabled_plugin_list())
+    server.func_get_mcdr_language = server.get_mcdr_language()
+    server.func_get_mcdr_config = str(server.get_mcdr_config())
+    return server

test/test.py

@@ -0,0 +1,88 @@
+import pytest as pytest
+
+from hooks import utils
+
+
+def test_type_list_detect():
+    input_list = [1, 2, 3]
+    assert utils.is_list_var(input_list) is True
+    
+    input_str_list = '[1, 2, 3]'
+    assert utils.is_list_var(input_str_list) is True
+    
+    input_str_list_invalid = '["a]'
+    assert utils.is_list_var(input_str_list_invalid) is False
+    
+    input_dict = {"key": "value"}
+    assert utils.is_list_var(input_dict) is False
+    
+    input_str_dict = '{"key": "value"}'
+    assert utils.is_list_var(input_str_dict) is False
+    
+    input_str_dict_invalid = '{"a}'
+    assert utils.is_list_var(input_str_dict_invalid) is False
+    
+    input_int = 1
+    assert utils.is_list_var(input_int) is False
+    
+    input_bool = True
+    assert utils.is_list_var(input_bool) is False
+
+
+def test_type_dict_detect():
+    input_list = [1, 2, 3]
+    assert utils.is_dict_var(input_list) is False
+    
+    input_str_list = '[1, 2, 3]'
+    assert utils.is_dict_var(input_str_list) is False
+    
+    input_str_list_invalid = '["a]'
+    assert utils.is_dict_var(input_str_list_invalid) is False
+    
+    input_dict = {"key": "value"}
+    assert utils.is_dict_var(input_dict) is True
+    
+    input_str_dict = '{"key": "value"}'
+    assert utils.is_dict_var(input_str_dict) is True
+    
+    input_str_dict_invalid = '{"a}'
+    assert utils.is_dict_var(input_str_dict_invalid) is False
+    
+    input_int = 1
+    assert utils.is_dict_var(input_int) is False
+    
+    input_bool = True
+    assert utils.is_dict_var(input_bool) is False
+
+
+def test_type_int_detect():
+    input_list = [1, 2, 3]
+    assert utils.is_int_var(input_list) is False
+    
+    input_str_list = '[1, 2, 3]'
+    assert utils.is_dict_var(input_str_list) is False
+    
+    input_str_list_invalid = '["a]'
+    assert utils.is_int_var(input_str_list_invalid) is False
+    
+    input_dict = {"key": "value"}
+    assert utils.is_int_var(input_dict) is False
+    
+    input_str_dict = '{"key": "value"}'
+    assert utils.is_int_var(input_str_dict) is False
+    
+    input_str_dict_invalid = '{"a}'
+    assert utils.is_int_var(input_str_dict_invalid) is False
+    
+    input_int = 1
+    assert utils.is_int_var(input_int) is True
+    
+    input_str_int = '1'
+    assert utils.is_int_var(input_str_int) is True
+    
+    input_bool = True
+    assert utils.is_int_var(input_bool) is False
+
+
+if __name__ == "__main__":
+    pytest.main()