feat(MaaEnd): 重构自动代理任务配置管理功能

- 添加re、json5、shutil模块依赖
- 移除ScriptConfig.py中的重复函数定义，优化代码结构
- 重构MaaEnd配置文件处理逻辑，使用copytree替代原有的单例处理
- 更新任务字典数据结构，支持多层级任务状态管理
- 优化日志监控中任务完成状态的判断逻辑
- 添加i18n国际化配置加载功能
- 改进任务实例ID统一为'automas'
- 增强错误处理机制，提升代码健壮性
- 优化配置文件路径验证逻辑

Author: DLmaster361

app/task/MaaEnd/AutoProxy.py

@@ -19,8 +19,11 @@
 #   Contact: DLmaster_361@163.com
 
 
+import re
 import uuid
 import json
+import json5
+import shutil
 import asyncio
 from pathlib import Path
 from datetime import datetime, timedelta
@@ -34,7 +37,6 @@
 from app.utils import get_logger, LogMonitor, ProcessManager, skland_sign_in
 from app.utils.constants import UTC4, UTC8, MAAEND_KILLPROC_TASK
 from .tools import login, push_notification, wait_and_focus_window
-from .ScriptConfig import CONFIG_FILE_NAME, _keep_single_instance, _replace_config_dir
 
 logger = get_logger("MaaEnd 自动代理")
 
@@ -135,7 +137,7 @@ async def main_task(self):
         logger.info(f"开始代理用户 {self.cur_user_uid}")
         self.cur_user_item.status = "运行"
 
-        self.task_dict: dict[str, bool] | None = None
+        self.task_dict: dict[str, dict[str, bool]] | None = None
         self.unique_task: dict[str, str] = {}
 
         if (
@@ -253,9 +255,7 @@ async def main_task(self):
             self.wait_event.clear()
             t = datetime.now()
             await self.maaend_process_manager.open_process(
-                self.maaend_exe_path,
-                "--log-mode=verbose",
-                stdout=asyncio.subprocess.PIPE,
+                self.maaend_exe_path, stdout=asyncio.subprocess.PIPE
             )
 
             # 静默模式隐藏 MaaEnd 窗口
@@ -363,28 +363,26 @@ async def set_maaend(self, device_info: DeviceInfo | None) -> None:
         await self.maaend_process_manager.kill()
         await System.kill_process(self.maaend_exe_path)
 
-        source_config_path = None
-        if self.cur_user_config.get("Info", "Mode") == "简洁":
-            source_config_path = (
-                Path.cwd() / f"data/{self.script_info.script_id}/Default/ConfigFile"
-            )
-        elif self.cur_user_config.get("Info", "Mode") == "详细":
-            source_config_path = (
-                Path.cwd()
-                / f"data/{self.script_info.script_id}/{self.cur_user_uid}/ConfigFile"
-            )
-
-        if source_config_path is None:
-            raise RuntimeError("未找到 MaaEnd 配置目录")
-
-        source_config_file = source_config_path / CONFIG_FILE_NAME
-        if source_config_file.exists():
-            _keep_single_instance(source_config_file)
-        _replace_config_dir(source_config_path, self.maaend_set_path)
+        # 基础配置内容
+        maaend_config_path = (
+            Path.cwd()
+            / f"data/{self.script_info.script_id}/{'Default' if self.cur_user_config.get('Info', 'Mode') == '简洁' else self.cur_user_uid}/ConfigFile"
+        )
+        shutil.rmtree(self.maaend_set_path, ignore_errors=True)
+        shutil.copytree(maaend_config_path, self.maaend_set_path)
 
-        maaend_set, maaend_instance = _keep_single_instance(
-            self.maaend_set_path / CONFIG_FILE_NAME
+        # 初始化任务实例
+        maaend_set = json.loads(
+            (self.maaend_set_path / "mxu-MaaEnd.json").read_text(encoding="utf-8")
         )
+
+        # 获取任务项单例
+        for instance in maaend_set["instances"]:
+            if instance.get("id") == "automas":
+                maaend_instance = instance
+                break
+        else:
+            maaend_instance = {"id": "automas", "name": "AUTO-MAS", "tasks": []}
         maaend_tasks = maaend_instance["tasks"]
 
         settings = maaend_set.get("settings")
@@ -393,7 +391,7 @@ async def set_maaend(self, device_info: DeviceInfo | None) -> None:
             maaend_set["settings"] = settings
 
         # 直接运行任务
-        settings["autoStartInstanceId"] = maaend_instance["id"]
+        settings["autoStartInstanceId"] = "automas"
         settings["autoRunOnLaunch"] = True
 
         # 模拟器相关配置
@@ -408,32 +406,47 @@ async def set_maaend(self, device_info: DeviceInfo | None) -> None:
                 "adbDeviceName": (await MaaFWManager.convert_adb(device_info)).name
             }
 
+        # 加载 i18n 配置
+        if settings["language"] == "system":
+            settings["language"] = "zh-CN"
+        maaend_i18n_raw = json.loads(
+            (
+                self.maaend_root_path
+                / f"locales/interface/{settings['language'].lower().replace('-', '_')}.json"
+            ).read_text(encoding="utf-8")
+        )
+        maaend_i18n = {}
+        for task_definition_file in self.maaend_root_path.glob("tasks/*.json"):
+            task_definition = json5.loads(  # type: ignore
+                task_definition_file.read_text(encoding="utf-8")
+            )["task"][0]
+            if task_definition["label"].startswith("$"):
+                maaend_i18n[task_definition["name"]] = maaend_i18n_raw[
+                    task_definition["label"].lstrip("$")
+                ]
+            else:
+                maaend_i18n[task_definition["name"]] = task_definition["label"]
+
         # 配置任务启用状态
         if self.task_dict is None:
             # 任务列表为空则记录任务
             self.task_dict = {}
             task = {}
             for task in maaend_tasks:
-                self.task_dict[task["id"]] = task["enabled"]
-            if task.get("taskName") == "__MXU_KILLPROC__" and task.get(
-                "optionValues", {}
-            ).get("__MXU_KILLPROC_SELF_OPTION__", {}).get("value", False):
-                self.task_dict.popitem()
+                if task.get("taskName") == "__MXU_KILLPROC__" and task.get(
+                    "optionValues", {}
+                ).get("__MXU_KILLPROC_SELF_OPTION__", {}).get("value", False):
+                    continue
+                task_name = maaend_i18n.get(task["taskName"], task["taskName"])
+                if task_name not in self.task_dict:
+                    self.task_dict[task_name] = {}
+                self.task_dict[task_name][task["id"]] = task["enabled"]
         else:
             # 任务列表不为空则配置任务
             for task in maaend_tasks:
-                task["enabled"] = self.task_dict[task["id"]]
-
-        # 记录启用的无重复任务项以便简化判定
-        self.unique_task = {}
-        duplicate_task = set()
-        for task in maaend_tasks:
-            if task["enabled"] and task["id"] in self.task_dict:
-                if task["taskName"] in self.unique_task:
-                    self.unique_task.pop(task["taskName"])
-                    duplicate_task.add(task["taskName"])
-                elif task["taskName"] not in duplicate_task:
-                    self.unique_task[task["taskName"]] = task["id"]
+                task_name = maaend_i18n.get(task["taskName"], task["taskName"])
+                if task_name in self.task_dict:
+                    task["enabled"] = self.task_dict[task_name][task["id"]]
 
         # 配置协议空间
         for task in maaend_tasks:
@@ -480,16 +493,34 @@ async def check_log(self, log_content: list[str], latest_time: datetime) -> None
             if self.task_dict is None:
                 self.cur_user_log.status = "MaaEnd 未加载任何任务"
             else:
-                for id in self.task_dict.keys():
-                    if f"[{id}] 任务完成" in log:
-                        self.task_dict[id] = False
-                for task_name, task_id in self.unique_task.items():
-                    if f"任务完成: {task_name}" in log:
-                        self.task_dict[task_id] = False
-                if any(self.task_dict.values()):
-                    self.cur_user_log.status = "MaaEnd 部分任务执行失败"
-                else:
-                    self.cur_user_log.status = "Success!"
+                try:
+                    task_name = ""
+                    task_index = {
+                        k: {"index": 0, "list": list(v.keys())}
+                        for k, v in self.task_dict.items()
+                    }
+                    for log_line in self.cur_user_log.content:
+                        match = re.search(r"任务开始:\s*(.+)", log_line)
+                        task_name = match.group(1) if match else task_name
+                        if (
+                            task_name in self.task_dict
+                            and f"任务完成: {task_name}" in log_line
+                        ):
+                            self.task_dict[task_name][
+                                task_index[task_name]["list"][
+                                    task_index[task_name]["index"]
+                                ]
+                            ] = False
+                            task_index[task_name]["index"] += 1
+                        elif f"任务失败: {task_name}" in log_line:
+                            task_index[task_name]["index"] += 1
+
+                    if any(any(_.values()) for _ in self.task_dict.values()):
+                        self.cur_user_log.status = "MaaEnd 部分任务执行失败"
+                    else:
+                        self.cur_user_log.status = "Success!"
+                except:
+                    self.cur_user_log.status = "MaaEnd 任务执行情况解析失败"
 
         elif datetime.now() - latest_time > timedelta(
             minutes=self.script_config.get("Run", "RunTimeLimit")

app/task/MaaEnd/ScriptConfig.py

@@ -23,7 +23,7 @@
 import shutil
 import asyncio
 from pathlib import Path
-from typing import Any
+from copy import deepcopy
 
 from app.core import Config
 from app.models.task import TaskExecuteBase, ScriptItem
@@ -34,65 +34,6 @@
 from app.utils import get_logger, ProcessManager
 
 logger = get_logger("MaaEnd 脚本设置")
-CONFIG_FILE_NAME = "mxu-MaaEnd.json"
-
-
-def _load_config(config_path: Path) -> dict[str, Any]:
-    return json.loads(config_path.read_text(encoding="utf-8"))
-
-
-def _dump_config(config_path: Path, data: dict[str, Any]) -> None:
-    config_path.write_text(
-        json.dumps(data, ensure_ascii=False, indent=4), encoding="utf-8"
-    )
-
-
-def _keep_single_instance(config_path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
-    data = _load_config(config_path)
-
-    instances = data.get("instances", [])
-
-    selected_instance: dict[str, Any] | None = None
-
-    for instance in instances:
-        if instance.get("name") == "AUTO-MAS":
-            selected_instance = dict(instance)
-            break
-
-    if selected_instance is None:
-        active_id = str(data.get("lastActiveInstanceId", "")).strip()
-        if active_id:
-            for instance in instances:
-                if str(instance.get("id", "")).strip() == active_id:
-                    selected_instance = dict(instance)
-                    break
-
-    if selected_instance is None:
-        for instance in instances:
-            selected_instance = dict(instance)
-            break
-
-    if selected_instance is None:
-        selected_instance = {"id": "", "name": "AUTO-MAS", "tasks": []}
-
-    if "tasks" not in selected_instance:
-        selected_instance["tasks"] = []
-    if not str(selected_instance.get("id", "")).strip():
-        selected_instance["id"] = "AUTO-MAS"
-
-    selected_instance["name"] = "AUTO-MAS"
-    data["instances"] = [selected_instance]
-    data["lastActiveInstanceId"] = selected_instance["id"]
-
-    data.setdefault("settings", {})
-
-    _dump_config(config_path, data)
-    return data, selected_instance
-
-
-def _replace_config_dir(source_dir: Path, target_dir: Path) -> None:
-    shutil.rmtree(target_dir, ignore_errors=True)
-    shutil.copytree(source_dir, target_dir)
 
 
 class ScriptConfigTask(TaskExecuteBase):
@@ -147,18 +88,44 @@ async def set_maaend(self):
         await self.maaend_process_manager.kill()
         await System.kill_process(self.maaend_exe_path)
 
-        if self.config_file_path.exists():
-            config_path = self.config_file_path / CONFIG_FILE_NAME
-            if config_path.exists():
-                _keep_single_instance(config_path)
-            _replace_config_dir(self.config_file_path, self.maaend_set_path)
-
-        maaend_set, maaend_instance = _keep_single_instance(
-            self.maaend_set_path / CONFIG_FILE_NAME
+        # 初始化任务实例
+        maaend_set = json.loads(
+            (self.maaend_set_path / "mxu-MaaEnd.json").read_text(encoding="utf-8")
+        )
+        maaend_instances = maaend_set["instances"]
+
+        selected_instance = None
+
+        # 创建任务项单例
+        for instance in maaend_instances:
+            if instance.get("id") == "automas":
+                selected_instance = deepcopy(instance)
+                break
+        else:
+            if maaend_set.get("lastActiveInstanceId"):
+                for instance in maaend_instances:
+                    if instance.get("id") == maaend_set.get("lastActiveInstanceId"):
+                        selected_instance = deepcopy(instance)
+                        break
+                else:
+                    selected_instance = (
+                        deepcopy(maaend_instances[0])
+                        if len(maaend_instances) > 0
+                        else {"id": "automas", "name": "AUTO-MAS", "tasks": []}
+                    )
+
+        selected_instance = dict(
+            selected_instance or {"id": "automas", "name": "AUTO-MAS", "tasks": []}
         )
+        if "tasks" not in selected_instance:
+            selected_instance["tasks"] = []
+        selected_instance["id"] = "automas"
+        selected_instance["name"] = "AUTO-MAS"
+        maaend_instances = [selected_instance]
+        maaend_set["lastActiveInstanceId"] = "automas"
 
         # 不直接运行任务
-        maaend_set["settings"]["autoStartInstanceId"] = maaend_instance["id"]
+        maaend_set["settings"]["autoStartInstanceId"] = "automas"
         maaend_set["settings"]["autoRunOnLaunch"] = False
 
         (self.maaend_set_path / "mxu-MaaEnd.json").write_text(
@@ -173,7 +140,6 @@ async def final_task(self):
         await self.maaend_process_manager.kill()
         await System.kill_process(self.maaend_exe_path)
 
-        _keep_single_instance(self.maaend_set_path / CONFIG_FILE_NAME)
         shutil.rmtree(self.config_file_path, ignore_errors=True)
         shutil.copytree(self.maaend_set_path, self.config_file_path)
 

app/task/MaaEnd/manager.py

@@ -81,13 +81,23 @@ async def check(self) -> str:
         ):
             return "未完成游戏配置, 请检查脚本配置中的游戏设置！"
 
+        if not (
+            Path(
+                Config.ScriptConfig[uuid.UUID(self.script_info.script_id)].get(
+                    "Info", "Path"
+                )
+            )
+            / "config/mxu-MaaEnd.json"
+        ).exists():
+            return "MaaEnd 配置文件不存在, 请检查 MaaEnd 路径设置或先启动 MaaEnd 完成配置文件生成！"
+
         if (
             self.task_info.mode != "ScriptConfig"
             and not (
                 Path.cwd() / f"data/{self.script_info.script_id}/Default/ConfigFile"
             ).exists()
         ):
-            return "MaaEnd default config is missing, please configure MaaEnd first."
+            return "未完成 MaaEnd 全局设置, 请先设置 MaaEnd！"
         return "Pass"
 
     async def prepare(self):

requirements.txt

@@ -24,4 +24,5 @@ numpy==2.3.4
 opencv-python==4.10.0.84
 maafw==5.8.1
 mss==9.0.2
-fastapi-mcp==0.4.0
+fastapi-mcp==0.4.0
+json5==0.14.0