diff --git a/res/version.json b/res/version.json index f07f6cdb8..bbcb636ac 100644 --- a/res/version.json +++ b/res/version.json @@ -3,6 +3,7 @@ "version_info": { "v5.5.0-beta.3": { "新增功能": [ + "专项适配 提供可复制的 General 能力基线模板，覆盖后端任务、前端编辑页、注册清单与最小回归测试", "MFW 项目支持在脚本运行前或运行后自动更新，可在项目配置中选择时机；升级后已有脚本默认为「运行前更新」，不需要可在项目配置中改为「不更新」 by [@qiyinxi](https://github.com/qiyinxi)", "MAA专项 新增绿票商店开关：开启后每月单独启动一次 MAA 自动购买绿票商店，用户配置页可查看本月状态并手动重置 by [@qiyinxi](https://github.com/qiyinxi)", "调度队列 新增循环队列：队列里的每个任务可单独设定固定时间或间隔重复运行，在调度台以「循环运行」启动后会按各自的周期一直跑下去，并显示接下来要运行的任务与时间 by [@qiyinxi](https://github.com/qiyinxi)", diff --git a/templates/specialized/README.md b/templates/specialized/README.md new file mode 100644 index 000000000..1a6ef1c99 --- /dev/null +++ b/templates/specialized/README.md @@ -0,0 +1,78 @@ +# 专项适配模板 + +这是一套从 `General` 能力线整理出的专项适配骨架。它不是独立运行的脚本，也不依赖专用模板生成器；复制后按注册清单接入 AUTO-MAS 即可。接入新 schema 后，仍按项目流程生成 OpenAPI client。 + +## 占位符 + +- `Xxx`：Python/Vue 的 PascalCase 专项名，例如 `MaaDemo`。 +- `xxx`：路径、测试和路由使用的 kebab-case 或 snake_case 名，例如 `maa_demo`。 +- `专项显示名称`：用户界面和日志中的中文名称。 + +全局替换时请同时检查大小写和中文显示名。`Xxx` 不是最终的 `ScriptType`，最终类型名必须与注册表、路由、schema 和任务调度完全一致。 + +## 目录说明 + +```text +templates/specialized/ +├─ README.md +├─ backend/ +│ ├─ task/Xxx/ +│ │ ├─ __init__.py +│ │ ├─ AutoProxy.py +│ │ ├─ manager.py +│ │ └─ ScriptConfig.py +│ ├─ config.py.template +│ ├─ schema.py.template +│ └─ registration-checklist.md +├─ frontend/ +│ ├─ XxxScriptEdit.vue +│ ├─ XxxUserEdit.vue +│ └─ XxxUserEdit/ +│ ├─ BasicInfoSection.vue +│ └─ NotifyConfigSection.vue +└─ tests/ + └─ test_xxx_autoproxy.py +``` + +后端任务文件保留了 General 的启动、进程追踪、配置交换、游戏/模拟器启动、重试/超时、前后置脚本、日志判态、用户统计、历史记录、通知和 `final_task` / `on_crash` 生命周期。专项差异集中在 `TODO(specialized)` 处；不要把 TODO 留在可运行路径上。 + +## 五步使用流程 + +1. 复制 `backend/task/Xxx` 到 `app/task/新专项名`，并全局替换 `Xxx`、`xxx` 和显示名称。 +2. 复制 `backend/config.py.template`、`backend/schema.py.template` 的片段，填写真实专项字段和验证器。 +3. 复制两个前端编辑页及 `XxxUserEdit/`，保留基本信息、通知和通用配置会话，加入专项表单。 +4. 按 `backend/registration-checklist.md` 补齐 `ScriptType`、`BOOK`、API/core/task 调度、路由、Hub 和前端类型分支。 +5. 将 `tests/test_xxx_autoproxy.py` 移入 `tests/task/test_xxx_autoproxy.py`，替换夹具并运行最小专项测试。 + +前端通常放置为：`XxxScriptEdit.vue` → `frontend/src/views/EditView/Script/`，`XxxUserEdit.vue` → `frontend/src/views/EditView/User/`，`XxxUserEdit/` → `frontend/src/views/XxxUserEdit/`；页面中的相对 import 已按该目录关系书写。 + +## 设计边界 + +- `schema.py` 只描述 API 数据；文件、进程、日志和配置交换留在 task/core。 +- `config.py` 中的所有 `ConfigItem` 必须在 `super().__init__()` 前声明，并配有注释。 +- 用户配置来源默认沿用 General 的“用户独立 / 脚本直控”两态。若上游真实存在脚本共享或其他 owner，先在专项设计中确认，再同步 config、UI、AutoProxy 和 ScriptConfig；不要为了界面统一臆造第三种模式。 +- 运行前备份脚本直控配置，运行中按用户配置原子替换，成功、失败、取消、超时和异常均恢复原配置。需要把运行结果写回用户配置时，必须先更新用户副本，再恢复直控配置。 +- 日志监控使用 `LogMonitor(time_stamp_range, time_format, check_log)` 三参构造；每一轮创建 `LogRecord`，只写 `log_record.status`，不要给 `UserItem.result` 赋值。 +- 生成的 OpenAPI 文件不得手改。schema 接入后由开发者从 `frontend/` 运行 `yarn openapi`。 + +## 必填专项决策 + +复制前先写下四个答案： + +1. 脚本根目录、主程序和目标进程的真实哨兵文件是什么？自动发现和手动选择必须复用同一组哨兵。 +2. 自动任务的启动参数如何构造？没有稳定 CLI 时，改为写入上游约定的运行配置，不要猜参数。 +3. 哪些配置由上游脚本拥有，哪些由 MAS 用户副本拥有？配置会话、AutoProxy 和恢复逻辑必须一致。 +4. 哪些日志明确代表成功、失败、运行中和提前退出？把失败/回退路径写入测试。 + +## 验证 + +```powershell +# 后端：在替换占位符并完成注册后 +python -m pytest tests/task/test_xxx_autoproxy.py -q + +# 前端：接入 schema 后由开发者执行生成器，再检查类型和 lint +yarn openapi +yarn lint +``` + +模板本身只提供最小纯逻辑回归测试；未完成上游契约、注册和夹具替换前，不要把模板测试当作专项已适配的证明。 diff --git a/templates/specialized/backend/config.py.template b/templates/specialized/backend/config.py.template new file mode 100644 index 000000000..52737657d --- /dev/null +++ b/templates/specialized/backend/config.py.template @@ -0,0 +1,248 @@ +"""专项配置片段。 + +将本文件中的两个 ConfigBase 子类复制到 ``app/models/config.py``，并复用该文件 +已有的 ConfigItem、Validator、MultipleConfig、Webhook 和 UTC4 imports。模板只放 +真实运行时会被 AutoProxy/ScriptConfig 消费的字段；专项字段必须同时出现在 schema、 +前端表单和任务逻辑中。 +""" + + +class XxxUserConfig(ConfigBase): + """专项用户配置：从 General 用户配置开始，再加入专项字段。""" + + def __init__(self) -> None: + ## Info ------------------------------------------------------------ + ## 用户名称 + self.Info_Name = ConfigItem("Info", "Name", "新用户", UserNameValidator()) + ## 是否启用 + self.Info_Status = ConfigItem("Info", "Status", True, BoolValidator()) + ## 剩余天数，-1 表示无限 + self.Info_RemainedDay = ConfigItem( + "Info", "RemainedDay", -1, RangeValidator(-1, 9999) + ) + ## 是否使用用户独立脚本配置；关闭时直接使用脚本原配置 + self.Info_IfUseMasConfig = ConfigItem( + "Info", "IfUseMasConfig", True, BoolValidator() + ) + ## 是否在任务前执行自定义脚本 + self.Info_IfScriptBeforeTask = ConfigItem( + "Info", "IfScriptBeforeTask", False, BoolValidator() + ) + ## 任务前脚本路径 + self.Info_ScriptBeforeTask = ConfigItem( + "Info", "ScriptBeforeTask", "", FileValidator() + ) + ## 是否在任务后执行自定义脚本 + self.Info_IfScriptAfterTask = ConfigItem( + "Info", "IfScriptAfterTask", False, BoolValidator() + ) + ## 任务后脚本路径 + self.Info_ScriptAfterTask = ConfigItem( + "Info", "ScriptAfterTask", "", FileValidator() + ) + ## 用户备注 + self.Info_Notes = ConfigItem("Info", "Notes", "无") + ## 用户列表展示标签 + self.Info_Tag = ConfigItem( + "Info", "Tag", "[ ]", VirtualConfigValidator(self.getTags) + ) + + ## Task ------------------------------------------------------------ + # TODO(specialized): 增加专项任务字段，并在 AutoProxy/前端真正消费。 + # self.Task_Example = ConfigItem( + # "Task", "Example", False, BoolValidator() + # ) + + ## Data ------------------------------------------------------------ + ## 上次代理日期 + self.Data_LastProxyDate = ConfigItem( + "Data", "LastProxyDate", "2000-01-01", DateTimeValidator("%Y-%m-%d") + ) + ## 当日代理次数 + self.Data_ProxyTimes = ConfigItem( + "Data", "ProxyTimes", 0, RangeValidator(0, 9999) + ) + + ## Notify ---------------------------------------------------------- + ## 是否启用用户通知 + self.Notify_Enabled = ConfigItem("Notify", "Enabled", False, BoolValidator()) + ## 是否发送统计信息 + self.Notify_IfSendStatistic = ConfigItem( + "Notify", "IfSendStatistic", False, BoolValidator() + ) + ## 是否发送邮件 + self.Notify_IfSendMail = ConfigItem( + "Notify", "IfSendMail", False, BoolValidator() + ) + ## 邮件收件地址 + self.Notify_ToAddress = ConfigItem("Notify", "ToAddress", "") + ## 是否发送 Server 酱 + self.Notify_IfServerChan = ConfigItem( + "Notify", "IfServerChan", False, BoolValidator() + ) + ## Server 酱密钥 + self.Notify_ServerChanKey = ConfigItem("Notify", "ServerChanKey", "") + ## 自定义 Webhook + self.Notify_CustomWebhooks = MultipleConfig([Webhook]) + + super().__init__() + + def getTags(self) -> str: + """生成用户列表需要的简短状态标签。""" + tags = [] + if ( + datetime.strptime(self.get("Data", "LastProxyDate"), "%Y-%m-%d").date() + == datetime.now(tz=UTC4).date() + ): + tags.append( + {"text": f"任务：已代理{self.get('Data', 'ProxyTimes')}次", "color": "green"} + ) + else: + tags.append({"text": "任务：未代理", "color": "orange"}) + + remained_day = self.get("Info", "RemainedDay") + if remained_day == -1: + color = "gold" + elif remained_day == 0: + color = "red" + elif remained_day <= 3: + color = "orange" + elif remained_day <= 7: + color = "yellow" + elif remained_day <= 30: + color = "blue" + else: + color = "green" + tags.append( + { + "text": ( + f"剩余天数：{remained_day}天" + if remained_day >= 0 + else "剩余天数：无期限" + ), + "color": color, + } + ) + + notes = self.get("Info", "Notes") + tags.append( + {"text": f"备注：{notes}" if len(notes) <= 20 else f"备注：{notes[:20]}...", "color": "pink"} + ) + return json.dumps(tags, ensure_ascii=False) + + +class XxxConfig(ConfigBase): + """专项脚本配置：保留 General 运行基线。""" + + related_config: dict[str, MultipleConfig] = {} + + def __init__(self) -> None: + ## Info ------------------------------------------------------------ + ## 脚本显示名称 + self.Info_Name = ConfigItem("Info", "Name", "新专项脚本") + ## 脚本根目录 + self.Info_RootPath = ConfigItem("Info", "RootPath", "", FileValidator()) + + ## Script ---------------------------------------------------------- + ## 脚本主程序 + self.Script_ScriptPath = ConfigItem( + "Script", "ScriptPath", "", FileValidator() + ) + ## 通用启动参数；专项参数在 AutoProxy 中构造 + self.Script_Arguments = ConfigItem( + "Script", "Arguments", "", AdvancedArgumentValidator() + ) + ## 是否追踪目标进程 + self.Script_IfTrackProcess = ConfigItem( + "Script", "IfTrackProcess", False, BoolValidator() + ) + ## 目标进程名称 + self.Script_TrackProcessName = ConfigItem("Script", "TrackProcessName", "") + ## 目标进程可执行文件 + self.Script_TrackProcessExe = ConfigItem("Script", "TrackProcessExe", "") + ## 目标进程命令行 + self.Script_TrackProcessCmdline = ConfigItem( + "Script", "TrackProcessCmdline", "", ArgumentValidator() + ) + ## 脚本配置文件或目录 + self.Script_ConfigPath = ConfigItem( + "Script", "ConfigPath", "", FileValidator() + ) + ## 配置路径类型 + self.Script_ConfigPathMode = ConfigItem( + "Script", "ConfigPathMode", "File", OptionsValidator(["File", "Folder"]) + ) + ## 脚本配置回写时机 + self.Script_UpdateConfigMode = ConfigItem( + "Script", + "UpdateConfigMode", + "Never", + OptionsValidator(["Never", "Success", "Failure", "Always"]), + ) + ## 日志文件路径 + self.Script_LogPath = ConfigItem("Script", "LogPath", "", FileValidator()) + ## 动态日志文件名格式；固定文件名可留空 + self.Script_LogPathFormat = ConfigItem("Script", "LogPathFormat", "%Y-%m-%d") + ## 日志时间戳在日志行中的起止位置 + self.Script_LogTimeStart = ConfigItem( + "Script", "LogTimeStart", 1, RangeValidator(1, 9999) + ) + self.Script_LogTimeEnd = ConfigItem( + "Script", "LogTimeEnd", 1, RangeValidator(1, 9999) + ) + ## 日志时间戳格式 + self.Script_LogTimeFormat = ConfigItem( + "Script", "LogTimeFormat", "%Y-%m-%d %H:%M:%S" + ) + ## 成功日志关键词，多个关键词用 | 分隔 + self.Script_SuccessLog = ConfigItem("Script", "SuccessLog", "") + ## 失败日志关键词，多个关键词用 | 分隔 + self.Script_ErrorLog = ConfigItem("Script", "ErrorLog", "") + + ## Game ------------------------------------------------------------ + ## 是否由 MAS 启动游戏或模拟器 + self.Game_Enabled = ConfigItem("Game", "Enabled", False, BoolValidator()) + ## Emulator / Client / URL + self.Game_Type = ConfigItem( + "Game", "Type", "Emulator", OptionsValidator(["Emulator", "Client", "URL"]) + ) + ## PC 游戏或启动器路径 + self.Game_Path = ConfigItem("Game", "Path", "", FileValidator()) + ## URL 协议 + self.Game_URL = ConfigItem("Game", "URL", "") + ## URL/客户端进程名称 + self.Game_ProcessName = ConfigItem("Game", "ProcessName", "") + ## 游戏启动参数 + self.Game_Arguments = ConfigItem("Game", "Arguments", "", ArgumentValidator()) + ## 启动等待时间（秒） + self.Game_WaitTime = ConfigItem("Game", "WaitTime", 0, RangeValidator(0, 9999)) + ## 是否强制关闭游戏 + self.Game_IfForceClose = ConfigItem( + "Game", "IfForceClose", False, BoolValidator() + ) + ## 模拟器实例关系 + self.Game_EmulatorId = ConfigItem( + "Game", "EmulatorId", "-", MultipleUIDValidator("-", self.related_config, "EmulatorConfig") + ) + self.Game_EmulatorIndex = ConfigItem("Game", "EmulatorIndex", "-") + + ## Task ------------------------------------------------------------ + # TODO(specialized): 增加脚本实际需要的专项级配置。 + + ## Run ------------------------------------------------------------- + ## 单用户每日代理次数上限，0 表示不限制 + self.Run_ProxyTimesLimit = ConfigItem( + "Run", "ProxyTimesLimit", 0, RangeValidator(0, 9999) + ) + ## 单用户尝试次数 + self.Run_RunTimesLimit = ConfigItem( + "Run", "RunTimesLimit", 3, RangeValidator(1, 9999) + ) + ## 无新日志的超时分钟数 + self.Run_RunTimeLimit = ConfigItem( + "Run", "RunTimeLimit", 10, RangeValidator(1, 9999) + ) + + self.UserData = MultipleConfig([XxxUserConfig]) + + super().__init__() diff --git a/templates/specialized/backend/registration-checklist.md b/templates/specialized/backend/registration-checklist.md new file mode 100644 index 000000000..1651f0463 --- /dev/null +++ b/templates/specialized/backend/registration-checklist.md @@ -0,0 +1,52 @@ +# 专项注册清单 + +复制模板后逐项勾选。没有真实消费者的字段、按钮或模式不要注册。 + +## 1. 配置与 schema + +- [ ] 将 `XxxConfig`、`XxxUserConfig` 复制进 `app/models/config.py`，所有 `ConfigItem` 位于 `super().__init__()` 前。 +- [ ] 将 `Xxx*` schema 复制进 `app/models/schema.py`，补齐真实专项字段、`Literal` 和描述。 +- [ ] `Config.ScriptConfig` / `GlobalConfig` / 相关 `MultipleConfig` 允许新类型。 +- [ ] `app/models/config.py` 的类映射、序列化和默认配置分支包含新类型。 +- [ ] `app/utils/constants.py` 的 `TYPE_BOOK["XxxConfig"]` 有用户可见文案。 + +## 2. API 与核心调度 + +- [ ] `app/api/scripts.py` 的 `SCRIPT_BOOK` 增加 `XxxConfig`。 +- [ ] `app/api/scripts.py` 的 `USER_BOOK` 增加 `XxxConfig: XxxUserConfig`。 +- [ ] 任何专项 API 只做请求校验/响应整形；文件交换、任务循环和日志判态留在 core/task。 +- [ ] `app/core/config.py` 的加载、创建、删除、用户增删和类型 union 分支包含新类型。 +- [ ] `app/core/task_manager.py` 导入 `XxxManager`，在脚本类型 dispatch 中注册。 +- [ ] 若接入计划表/任务队列，单独补 `PLAN_BOOK`、consumer、队列类型和对应前端表面；不要复制无关专项能力。 + +## 3. 任务模块 + +- [ ] 将 `task/Xxx/` 复制到 `app/task/Xxx/`，并全局替换类名、导入和 logger。 +- [ ] `manager.py` 的 `METHOD_BOOK` 至少包含 `AutoProxy` 与 `ScriptConfig`，并核对任务模式是否真实支持。 +- [ ] `AutoProxy.py` 的 `check()` 使用用户可操作的失败提示。 +- [ ] `AutoProxy.py` 的 `LogMonitor` 使用时间范围、时间格式、回调三参构造。 +- [ ] 每轮从 `log_record[start_time] = LogRecord()` 开始，只写 `log_record.status`。 +- [ ] `final_task` 与 `on_crash` 都停止监控、停止/清理进程、恢复配置、释放锁，并在需要时写历史记录。 +- [ ] 进程追踪至少有一个非空 `ProcessInfo` 字段；不要把空追踪条件交给运行时猜测。 +- [ ] 配置目录/文件复制使用临时路径后替换；逐步清理失败分别记录日志。 +- [ ] 多用户任务中单个用户检查失败只标记当前用户并继续后续用户。 + +## 4. 前端 Hub、路由和类型 + +- [ ] `frontend/src/types/script.ts` 增加 `ScriptType`、脚本/用户结构和默认值。 +- [ ] `frontend/src/composables/useScriptApi.ts` 增加脚本类型映射、默认配置和 `XxxUserConfig -> users[]` 分支。 +- [ ] `frontend/src/router/index.ts` 增加脚本编辑、用户新增、用户编辑路由；路径使用 lowercase kebab-case。 +- [ ] `frontend/src/views/Scripts.vue` 的编辑、添加用户、编辑用户、创建/复制脚本分支全部补齐。 +- [ ] `frontend/src/components/ScriptTable.vue` 增加图标、类型文案和专项操作（若确有专项动作）。 +- [ ] `frontend/src/views/EditView/Script/XxxScriptEdit.vue` 与 `EditView/User/XxxUserEdit.vue` 接入真实 API。 +- [ ] 新增用户流程先 `addUser`，再 `router.replace` 到带 `userId` 的编辑路由。 +- [ ] 若使用 ScriptConfig 遮罩，启动、完成、错误、取消、超时、卸载都停止任务并清理 WebSocket 订阅。 +- [ ] 自动发现与手动选择复用同一组哨兵文件；保存失败恢复旧值并显示原因。 + +## 5. 生成代码与验证 + +- [ ] 后端 schema 接入并重启后，确认 `openapi.json` 文本包含 `XxxConfig` / `XxxUserConfig`。 +- [ ] 在 `frontend/` 运行 `yarn openapi`；禁止手改 `frontend/src/api/**`。 +- [ ] 将测试移动到 `tests/task/test_xxx_autoproxy.py`，替换占位夹具，先运行该最小文件。 +- [ ] 前端改动至少运行 `yarn lint`；路由/类型/构建改动再运行相关 build/type 命令。 +- [ ] 版本记录在 `res/version.json` 下一个未发布版本中，说明新增专项模板或用户可见能力。 diff --git a/templates/specialized/backend/schema.py.template b/templates/specialized/backend/schema.py.template new file mode 100644 index 000000000..4325d00fa --- /dev/null +++ b/templates/specialized/backend/schema.py.template @@ -0,0 +1,123 @@ +"""专项 schema 片段。 + +将这些模型复制到 ``app/models/schema.py``，并在文件底部所有相关 union、index +和 API response model 中加入 XxxConfig/XxxUserConfig。schema 只描述数据契约， +不要在这里加入路径检查、进程启动或日志判态。 +""" + +from typing import Literal, Optional + +from pydantic import BaseModel, Field + + +class XxxUserConfig_Info(BaseModel): + Name: Optional[str] = Field(default=None, description="用户名") + Status: Optional[bool] = Field(default=None, description="是否启用") + RemainedDay: Optional[int] = Field(default=None, description="剩余天数，-1 表示无限") + IfUseMasConfig: Optional[bool] = Field( + default=None, description="是否使用用户独立脚本配置" + ) + IfScriptBeforeTask: Optional[bool] = Field( + default=None, description="是否在任务前执行脚本" + ) + ScriptBeforeTask: Optional[str] = Field(default=None, description="任务前脚本路径") + IfScriptAfterTask: Optional[bool] = Field( + default=None, description="是否在任务后执行脚本" + ) + ScriptAfterTask: Optional[str] = Field(default=None, description="任务后脚本路径") + Notes: Optional[str] = Field(default=None, description="备注") + Tag: Optional[str] = Field(default=None, description="用户标签 JSON 字符串") + + # TODO(specialized): 加入专项用户字段，并同步前端 BasicInfoSection。 + + +class XxxUserConfig_Data(BaseModel): + LastProxyDate: Optional[str] = Field(default=None, description="上次代理日期") + ProxyTimes: Optional[int] = Field(default=None, description="代理次数") + + +class XxxUserConfig_Notify(BaseModel): + Enabled: Optional[bool] = Field(default=None, description="是否启用通知") + IfSendStatistic: Optional[bool] = Field( + default=None, description="是否发送统计信息" + ) + IfSendMail: Optional[bool] = Field(default=None, description="是否发送邮件") + ToAddress: Optional[str] = Field(default=None, description="邮件收件地址") + IfServerChan: Optional[bool] = Field( + default=None, description="是否使用 Server 酱" + ) + ServerChanKey: Optional[str] = Field(default=None, description="Server 酱密钥") + + +class XxxUserConfig(BaseModel): + Info: Optional[XxxUserConfig_Info] = Field(default=None, description="用户信息") + Data: Optional[XxxUserConfig_Data] = Field(default=None, description="用户数据") + Notify: Optional[XxxUserConfig_Notify] = Field( + default=None, description="用户通知" + ) + + +class XxxConfig_Info(BaseModel): + Name: Optional[str] = Field(default=None, description="脚本名称") + RootPath: Optional[str] = Field(default=None, description="脚本根目录") + + +class XxxConfig_Script(BaseModel): + ScriptPath: Optional[str] = Field(default=None, description="脚本主程序路径") + Arguments: Optional[str] = Field(default=None, description="通用启动参数") + IfTrackProcess: Optional[bool] = Field( + default=None, description="是否追踪目标进程" + ) + TrackProcessName: Optional[str] = Field(default=None, description="目标进程名称") + TrackProcessExe: Optional[str] = Field(default=None, description="目标进程路径") + TrackProcessCmdline: Optional[str] = Field( + default=None, description="目标进程命令行" + ) + ConfigPath: Optional[str] = Field(default=None, description="配置文件或目录") + ConfigPathMode: Optional[Literal["File", "Folder"]] = Field( + default=None, description="配置路径类型" + ) + UpdateConfigMode: Optional[Literal["Never", "Success", "Failure", "Always"]] = ( + Field(default=None, description="用户配置回写时机") + ) + LogPath: Optional[str] = Field(default=None, description="日志文件路径") + LogPathFormat: Optional[str] = Field(default=None, description="日志文件名格式") + LogTimeStart: Optional[int] = Field(default=None, description="日志时间戳起点") + LogTimeEnd: Optional[int] = Field(default=None, description="日志时间戳终点") + LogTimeFormat: Optional[str] = Field(default=None, description="日志时间戳格式") + SuccessLog: Optional[str] = Field(default=None, description="成功日志关键词") + ErrorLog: Optional[str] = Field(default=None, description="失败日志关键词") + + +class XxxConfig_Game(BaseModel): + Enabled: Optional[bool] = Field(default=None, description="是否由 MAS 启动游戏") + Type: Optional[Literal["Emulator", "Client", "URL"]] = Field( + default=None, description="游戏启动类型" + ) + Path: Optional[str] = Field(default=None, description="游戏或启动器路径") + URL: Optional[str] = Field(default=None, description="游戏 URL 协议") + ProcessName: Optional[str] = Field(default=None, description="游戏进程名称") + Arguments: Optional[str] = Field(default=None, description="游戏启动参数") + WaitTime: Optional[int] = Field(default=None, description="游戏启动等待秒数") + IfForceClose: Optional[bool] = Field(default=None, description="是否强制关闭游戏") + EmulatorId: Optional[str] = Field(default=None, description="模拟器 ID") + EmulatorIndex: Optional[str] = Field(default=None, description="模拟器实例索引") + + +class XxxConfig_Run(BaseModel): + ProxyTimesLimit: Optional[int] = Field(default=None, description="每日代理次数上限") + RunTimesLimit: Optional[int] = Field(default=None, description="单用户重试次数") + RunTimeLimit: Optional[int] = Field(default=None, description="日志无变化超时分钟") + + +class XxxConfig_Task(BaseModel): + # TODO(specialized): 加入真实专项任务字段，并确保每个字段都有运行时消费者。 + pass + + +class XxxConfig(BaseModel): + Info: Optional[XxxConfig_Info] = Field(default=None, description="脚本基础信息") + Script: Optional[XxxConfig_Script] = Field(default=None, description="脚本配置") + Game: Optional[XxxConfig_Game] = Field(default=None, description="游戏配置") + Task: Optional[XxxConfig_Task] = Field(default=None, description="专项任务配置") + Run: Optional[XxxConfig_Run] = Field(default=None, description="运行配置") diff --git a/templates/specialized/backend/task/Xxx/AutoProxy.py b/templates/specialized/backend/task/Xxx/AutoProxy.py new file mode 100644 index 000000000..52fda0870 --- /dev/null +++ b/templates/specialized/backend/task/Xxx/AutoProxy.py @@ -0,0 +1,715 @@ +"""Xxx 专项自动代理任务。 + +这是从 ``app/task/general/AutoProxy.py`` 收敛出的可复制基线。复制后先完成 +``TODO(specialized)``，再接入真实 ScriptType；不要把这里的默认日志关键词或 +启动参数直接当成上游脚本契约。 +""" + +from __future__ import annotations + +import asyncio +import re +import shlex +import shutil +import uuid +from contextlib import suppress +from datetime import datetime, timedelta +from pathlib import Path + +from app.core import Config +from app.models.ConfigBase import MultipleConfig +from app.models.config import XxxConfig, XxxUserConfig +from app.models.emulator import DeviceBase +from app.models.task import LogRecord, ScriptItem, TaskExecuteBase +from app.services import Notify, System +from app.utils import ( + LogMonitor, + ProcessInfo, + ProcessManager, + get_logger, + is_process_running, + strptime, +) +from app.utils.constants import UTC4 +from app.task.general.tools import execute_script_task, push_notification + + +logger = get_logger("专项显示名称自动代理") + +_PREFIX_SENTINEL = "******" +_STRPTIME_DIRECTIVES: dict[str, str] = { + "%Y": r"\d{4}", + "%y": r"\d{2}", + "%m": r"\d{1,2}", + "%d": r"\d{1,2}", + "%H": r"\d{1,2}", + "%I": r"\d{1,2}", + "%M": r"\d{1,2}", + "%S": r"\d{1,2}", + "%f": r"\d+", + "%j": r"\d{1,3}", + "%U": r"\d{1,2}", + "%W": r"\d{1,2}", + "%w": r"\d", + "%A": r"\w+", + "%a": r"\w+", + "%B": r"\w+", + "%b": r"\w+", + "%p": r"[APap][Mm]", + "%%": r"%", +} + + +def _format_to_prefix_regex(fmt: str) -> re.Pattern[str]: + """将 strptime 格式转换成日志文件名前缀正则。""" + + parts: list[str] = [] + index = 0 + while index < len(fmt): + if fmt[index] == "%" and index + 1 < len(fmt): + directive = fmt[index : index + 2] + if directive in _STRPTIME_DIRECTIVES: + parts.append(_STRPTIME_DIRECTIVES[directive]) + index += 2 + continue + parts.append(re.escape(fmt[index])) + index += 1 + return re.compile("^" + "".join(parts)) + + +def _split_script_arguments(raw: str, script_path: Path) -> tuple[list[Path], list[list[str]]]: + """解析 General 的 ``path%args|path%args`` 参数格式。""" + + executable_paths: list[Path] = [] + arguments: list[list[str]] = [] + for item in (part.strip() for part in str(raw).split("|") if part.strip()): + parts = [part.strip() for part in item.split("%", 1) if part.strip()] + executable_paths.append( + (script_path / parts[0] if len(parts) > 1 else script_path).resolve() + ) + arguments.append(shlex.split(parts[-1], posix=False)) + return executable_paths, arguments + + +class AutoProxyTask(TaskExecuteBase): + """专项自动代理：一名用户串行执行多次，直到成功或达到重试上限。""" + + def __init__( + self, + script_info: ScriptItem, + script_config: XxxConfig, + user_config: MultipleConfig[XxxUserConfig], + game_manager: ProcessManager | DeviceBase | None, + ) -> None: + super().__init__() + if script_info.task_info is None: + raise RuntimeError("ScriptItem 未绑定到 TaskItem") + + self.task_info = script_info.task_info + self.script_info = script_info + self.script_config = script_config + self.user_config = user_config + self.game_manager = game_manager + self.cur_user_item = self.script_info.user_list[self.script_info.current_index] + self.cur_user_uid = uuid.UUID(self.cur_user_item.user_id) + self.cur_user_config = self.user_config[self.cur_user_uid] + self.use_mas_config = bool( + self.cur_user_config.get("Info", "IfUseMasConfig") + ) + + # 跨回调、final_task 和 on_crash 使用的状态全部显式初始化。 + self.check_result = "-" + self.run_book = False + self.wait_event: asyncio.Event | None = None + self.general_process_manager: ProcessManager | None = None + self.general_log_monitor: LogMonitor | None = None + self.script_exe_path: Path | None = None + self.script_path: Path | None = None + self.script_arguments: list[str] = [] + self.script_set_arguments: list[str] = [] + self.script_target_process_info: ProcessInfo | None = None + self.script_config_path: Path | None = None + self.script_log_path: Path | None = None + self.log_format = "" + self.log_use_prefix = False + self.game_path: Path | None = None + self.game_url = "" + self.game_process_name = "" + self.log_time_range = (0, 1) + self.success_log: list[str] = [] + self.error_log: list[str] = [] + self.curdate = "" + self.user_start_time = datetime.now() + self.log_start_time = datetime.now() + self.cur_user_log: LogRecord | None = None + + # 直控配置快照：任务结束、取消、超时和异常都从这里恢复。 + self.temp_path: Path | None = None + self.external_config_exists = False + self.external_config_snapshot_ready = False + + async def check(self) -> str: + """检查用户状态和专项运行前置条件。""" + + proxy_limit = self.script_config.get("Run", "ProxyTimesLimit") + if proxy_limit != 0 and self.cur_user_config.get("Data", "ProxyTimes") >= proxy_limit: + self.cur_user_item.status = "跳过" + return "今日代理次数已达上限, 跳过该用户" + + if self.use_mas_config and not self._user_config_source_path().exists(): + self.cur_user_item.status = "异常" + return "请先在用户配置页完成「专项显示名称配置」步骤" + + # TODO(specialized): 修改脚本路径校验 + script_path = Path(self.script_config.get("Script", "ScriptPath")) + if not script_path.exists(): + self.cur_user_item.status = "异常" + return "请设置脚本路径" + if not self.script_config.get("Info", "RootPath"): + self.cur_user_item.status = "异常" + return "请设置脚本根目录" + return "Pass" + + def _user_config_path(self) -> Path: + return ( + Path.cwd() + / "data" + / self.script_info.script_id + / str(self.cur_user_uid) + / "ConfigFile" + ) + + def _user_config_source_path(self) -> Path: + """返回当前用户副本中与脚本配置模式对应的源路径。""" + + user_path = self._user_config_path() + if self.script_config.get("Script", "ConfigPathMode") == "Folder": + return user_path + config_path = Path(self.script_config.get("Script", "ConfigPath")) + return user_path / config_path.name + + def _remove_path(self, path: Path) -> None: + if path.is_dir(): + shutil.rmtree(path) + elif path.exists(): + path.unlink() + + def _copy_path_atomic(self, source: Path, destination: Path, mode: str) -> None: + """以临时路径替换配置，避免中断留下半份配置。""" + + if mode == "Folder": + temporary = destination.with_name(destination.name + ".tmp") + shutil.rmtree(temporary, ignore_errors=True) + if source.exists(): + temporary.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(source, temporary, dirs_exist_ok=True) + self._remove_path(destination) + if temporary.exists(): + temporary.rename(destination) + return + + temporary = destination.with_name(destination.name + ".tmp") + if temporary.exists(): + temporary.unlink() + if source.exists(): + temporary.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, temporary) + temporary.replace(destination) + else: + self._remove_path(destination) + + def _snapshot_external_config(self) -> None: + """备份任务开始前的脚本直控配置。""" + + if self.script_config_path is None or self.temp_path is None: + return + shutil.rmtree(self.temp_path, ignore_errors=True) + self.external_config_exists = self.script_config_path.exists() + self.temp_path.mkdir(parents=True, exist_ok=True) + if self.external_config_exists: + mode = self.script_config.get("Script", "ConfigPathMode") + if mode == "Folder": + shutil.copytree(self.script_config_path, self.temp_path, dirs_exist_ok=True) + else: + shutil.copy2(self.script_config_path, self.temp_path / "config.temp") + self.external_config_snapshot_ready = True + + def _restore_external_config(self) -> None: + """恢复任务开始前的脚本直控配置。""" + + if ( + not self.external_config_snapshot_ready + or self.script_config_path is None + or self.temp_path is None + ): + return + self._remove_path(self.script_config_path) + if not self.external_config_exists: + return + mode = self.script_config.get("Script", "ConfigPathMode") + if mode == "Folder": + self._copy_path_atomic(self.temp_path, self.script_config_path, "Folder") + else: + self._copy_path_atomic( + self.temp_path / "config.temp", self.script_config_path, "File" + ) + + def _cleanup_external_config_snapshot(self) -> None: + if self.temp_path is not None: + shutil.rmtree(self.temp_path, ignore_errors=True) + self.external_config_snapshot_ready = False + + def _build_specialized_arguments(self) -> list[str]: + # TODO(specialized): 构造专项启动参数 + return [] + + async def prepare(self) -> None: + """加载运行参数、进程追踪信息、日志判态和配置快照。""" + + self.wait_event = asyncio.Event() + self.general_process_manager = ProcessManager() + self.user_start_time = datetime.now() + self.script_path = Path(self.script_config.get("Script", "ScriptPath")) + argument_paths, argument_lists = _split_script_arguments( + self.script_config.get("Script", "Arguments"), self.script_path + ) + self.script_exe_path = argument_paths[0] if argument_paths else self.script_path + self.script_arguments = ( + argument_lists[0] if argument_lists else [] + ) + self._build_specialized_arguments() + self.script_set_arguments = argument_lists[1] if len(argument_lists) > 1 else [] + + if self.script_config.get("Script", "IfTrackProcess"): + self.script_target_process_info = ProcessInfo( + name=self.script_config.get("Script", "TrackProcessName") or None, + exe=self.script_config.get("Script", "TrackProcessExe") or None, + cmdline=shlex.split( + self.script_config.get("Script", "TrackProcessCmdline"), + posix=False, + ) + or None, + ) + + self.script_config_path = Path(self.script_config.get("Script", "ConfigPath")) + self.temp_path = ( + Path.cwd() + / "data" + / self.script_info.script_id + / "Temp" + / str(self.cur_user_uid) + ) + self.script_log_path = Path(self.script_config.get("Script", "LogPath")) + self.log_format = self.script_config.get("Script", "LogPathFormat") or "" + self.log_use_prefix = self.log_format.endswith(_PREFIX_SENTINEL) + if self.log_use_prefix: + prefix_re = _format_to_prefix_regex( + self.log_format[: -len(_PREFIX_SENTINEL)] + ) + if not prefix_re.match(self.script_log_path.stem): + logger.warning( + f"LogPathFormat 与 LogPath 不匹配: {self.log_format} vs {self.script_log_path}" + ) + elif self.log_format: + with suppress(ValueError): + datetime.strptime(self.script_log_path.stem, self.log_format) + self.log_format += self.script_log_path.suffix + else: + self.log_format = self.script_log_path.name + + self.game_path = Path(self.script_config.get("Game", "Path")) + self.game_url = self.script_config.get("Game", "URL") + self.game_process_name = self.script_config.get("Game", "ProcessName") + self.log_time_range = ( + self.script_config.get("Script", "LogTimeStart") - 1, + self.script_config.get("Script", "LogTimeEnd"), + ) + self.success_log = [ + item.strip() + for item in self.script_config.get("Script", "SuccessLog").split("|") + if item.strip() + ] + self.error_log = [ + item.strip() + for item in self.script_config.get("Script", "ErrorLog").split("|") + if item.strip() + ] + self.general_log_monitor = LogMonitor( + self.log_time_range, + self.script_config.get("Script", "LogTimeFormat"), + self.check_log, + ) + self._snapshot_external_config() + + async def _start_game(self) -> None: + if self.game_manager is None: + return + try: + if isinstance(self.game_manager, ProcessManager): + if self.script_config.get("Game", "Type") == "URL": + if self.game_process_name and is_process_running(self.game_process_name): + logger.info(f"游戏进程已运行，跳过重复启动: {self.game_process_name}") + else: + await self.game_manager.open_protocol( + self.game_url, + ProcessInfo(name=self.game_process_name or None), + ) + else: + game_process_name = self.game_path.name if self.game_path else "" + if game_process_name and is_process_running(game_process_name): + logger.info(f"游戏进程已运行，跳过重复启动: {game_process_name}") + else: + await self.game_manager.open_process( + self.game_path, + *str(self.script_config.get("Game", "Arguments")).split(" "), + ) + await asyncio.sleep(self.script_config.get("Game", "WaitTime")) + elif isinstance(self.game_manager, DeviceBase): + await self.game_manager.open( + self.script_config.get("Game", "EmulatorIndex") + ) + except Exception as error: + await self.handle_pre_script_error("游戏/模拟器启动失败", error) + raise + + async def main_task(self) -> None: + """执行前置脚本、游戏、专项程序、日志监控和后置脚本。""" + + self.curdate = datetime.now(tz=UTC4).strftime("%Y-%m-%d") + if self.cur_user_config.get("Data", "LastProxyDate") != self.curdate: + await self.cur_user_config.set("Data", "LastProxyDate", self.curdate) + await self.cur_user_config.set("Data", "ProxyTimes", 0) + + self.check_result = await self.check() + if self.check_result != "Pass": + if self.cur_user_item.status == "异常": + await Config.send_websocket_message( + id=self.task_info.task_id, + type="Info", + data={"Error": self.check_result}, + ) + return + + await self.prepare() + self.cur_user_item.status = "运行" + run_limit = self.script_config.get("Run", "RunTimesLimit") + + for attempt in range(run_limit): + if self.run_book: + break + self.log_start_time = datetime.now() + self.cur_user_item.log_record[self.log_start_time] = self.cur_user_log = ( + LogRecord() + ) + logger.info(f"用户 {self.cur_user_item.name} - 尝试次数: {attempt + 1}/{run_limit}") + + if self.cur_user_config.get("Info", "IfScriptBeforeTask"): + await execute_script_task( + Path(self.cur_user_config.get("Info", "ScriptBeforeTask")), + "脚本前任务", + ) + + try: + await self._start_game() + await self.set_script_config() + if ( + self.general_process_manager is None + or self.script_exe_path is None + or self.wait_event is None + or self.general_log_monitor is None + ): + raise RuntimeError("专项任务未完成初始化") + self.wait_event.clear() + process_started_at = datetime.now() + await self.general_process_manager.open_process( + self.script_exe_path, + *self.script_arguments, + target_process=self.script_target_process_info, + ) + self.script_info.log = "正在等待脚本日志文件生成" + log_path = await self._wait_for_log_file(process_started_at) + if log_path is None: + await self.handle_pre_script_error("未找到日志文件") + continue + self.script_log_path = log_path + await self.general_log_monitor.start_monitor_file( + self.script_log_path, self.log_start_time + ) + await self.wait_event.wait() + await self.general_log_monitor.stop() + except Exception as error: + await self.handle_pre_script_error("专项任务启动失败", error) + continue + + if self.cur_user_log.status == "Success!": + self.run_book = True + self.script_info.log = "检测到专项任务完成，正在等待相关程序结束" + await self.kill_managed_process() + await asyncio.sleep(1) + if self.script_config.get("Script", "UpdateConfigMode") in ( + "Success", + "Always", + ): + await self.update_config() + else: + logger.warning( + f"用户 {self.cur_user_uid} - 代理任务异常: {self.cur_user_log.status}" + ) + self.script_info.log = f"{self.cur_user_log.status}\n正在中止相关程序" + await self.kill_managed_process() + await Notify.push_plyer( + "用户自动代理出现异常！", + f"用户 {self.cur_user_item.name} 的自动代理出现一次异常", + f"{self.cur_user_item.name}的自动代理出现异常", + 3, + ) + if self.script_config.get("Script", "UpdateConfigMode") in ( + "Failure", + "Always", + ): + await self.update_config() + + if self.cur_user_config.get("Info", "IfScriptAfterTask"): + await execute_script_task( + Path(self.cur_user_config.get("Info", "ScriptAfterTask")), + "脚本后任务", + ) + + async def _wait_for_log_file(self, started_at: datetime) -> Path | None: + """等待固定日志或按日期/序号生成的日志文件。""" + + if self.script_log_path is None: + return None + target_suffix: int | None = None + prefix_re = ( + _format_to_prefix_regex(self.log_format[: -len(_PREFIX_SENTINEL)]) + if self.log_use_prefix + else None + ) + for _ in range(60): + if self.script_log_path.exists() and not self.log_use_prefix: + return self.script_log_path + if prefix_re is not None and self.script_log_path.parent.exists(): + current_suffix = 0 + current_file: Path | None = None + for candidate in self.script_log_path.parent.iterdir(): + if not candidate.is_file(): + continue + match = prefix_re.match(candidate.name) + if not match: + continue + with suppress(ValueError): + file_time = strptime( + match.group(0), self.log_format[: -len(_PREFIX_SENTINEL)], started_at + ) + if file_time.date() != started_at.date(): + continue + suffix_match = re.search( + r"(\d+)\s*$", candidate.name[match.end() :].rsplit(".", 1)[0] + ) + suffix = int(suffix_match.group(1)) if suffix_match else 0 + if suffix > current_suffix: + current_suffix = suffix + current_file = candidate + if target_suffix is None: + target_suffix = current_suffix + 1 + if current_file is not None and current_suffix >= target_suffix: + return current_file + await asyncio.sleep(1) + return None + + async def handle_pre_script_error( + self, error_message: str, error: Exception | None = None + ) -> None: + message = error_message if error is None else f"{error_message}: {error}" + logger.warning(f"用户 {self.cur_user_uid} - {message}") + self.script_info.log = message + if self.cur_user_log is not None: + self.cur_user_log.content = [f"{message}, 无日志记录"] + self.cur_user_log.status = error_message + await Config.send_websocket_message( + id=self.task_info.task_id, + type="Info", + data={"Error": message}, + ) + await self.kill_managed_process() + + async def set_script_config(self) -> None: + """将用户副本导入脚本配置路径。""" + + if self.script_config_path is None: + return + await System.kill_process(self.script_exe_path) + if not self.use_mas_config: + logger.info("脚本直控配置：跳过写入用户配置") + return + # TODO(specialized): 写入专项配置 + self._copy_path_atomic( + self._user_config_source_path(), + self.script_config_path, + self.script_config.get("Script", "ConfigPathMode"), + ) + + async def update_config(self) -> None: + """按脚本设置把运行后的配置写回用户副本。""" + + if not self.use_mas_config or self.script_config_path is None: + return + self._copy_path_atomic( + self.script_config_path, + self._user_config_source_path(), + self.script_config.get("Script", "ConfigPathMode"), + ) + logger.success("专项脚本配置已更新") + + async def kill_managed_process(self) -> None: + """分别清理专项程序和游戏/模拟器，单步失败不阻塞后续清理。""" + + if self.general_process_manager is not None: + try: + await self.general_process_manager.kill() + except Exception as error: + logger.warning(f"中止专项进程管理器失败: {error}") + if self.script_exe_path is not None: + try: + await System.kill_process(self.script_exe_path) + except Exception as error: + logger.warning(f"中止专项主进程失败: {error}") + if self.game_manager is None: + return + try: + if isinstance(self.game_manager, ProcessManager): + await self.game_manager.kill() + if ( + self.script_config.get("Game", "Type") == "Client" + and self.script_config.get("Game", "IfForceClose") + and self.game_path is not None + ): + await System.kill_process(self.game_path) + elif isinstance(self.game_manager, DeviceBase): + await self.game_manager.close( + self.script_config.get("Game", "EmulatorIndex") + ) + except Exception as error: + logger.warning(f"关闭游戏/模拟器失败: {error}") + + async def check_log(self, log_content: list[str], latest_time: datetime) -> None: + """根据日志文本、时间戳和进程状态更新本轮结果。""" + + if self.cur_user_log is None or self.wait_event is None: + return + log = "".join(log_content) + self.cur_user_log.content = log_content + self.script_info.log = log + + # TODO(specialized): 定义成功与失败条件 + if any(marker in log for marker in self.success_log): + self.cur_user_log.status = "Success!" + elif datetime.now() - latest_time > timedelta( + minutes=self.script_config.get("Run", "RunTimeLimit") + ): + self.cur_user_log.status = "脚本进程超时" + elif any(marker in log for marker in self.error_log): + self.cur_user_log.status = "异常日志" + elif self.general_process_manager and await self.general_process_manager.is_running(): + self.cur_user_log.status = "专项脚本正常运行中" + elif self.success_log: + self.cur_user_log.status = "脚本在完成任务前退出" + else: + self.cur_user_log.status = "Success!" + + if self.cur_user_log.status != "专项脚本正常运行中": + self.wait_event.set() + + async def _stop_log_monitor(self) -> None: + if self.general_log_monitor is not None: + with suppress(Exception): + await self.general_log_monitor.stop() + + async def _save_history(self) -> None: + user_logs: list[Path] = [] + for start_time, log_item in self.cur_user_item.log_record.items(): + if log_item.status == "专项脚本正常运行中": + log_item.status = "任务被用户手动中止" + if not log_item.content: + log_item.content = ["未捕获到任何日志内容"] + log_item.status = "未捕获到日志" + log_time = start_time.replace( + tzinfo=datetime.now().astimezone().tzinfo + ).astimezone(UTC4) + history_path = Config.build_history_log_path( + script_name=self.script_info.name, + user_name=self.cur_user_item.name, + log_time=log_time, + ) + await Config.save_general_log(history_path, log_item.content, log_item.status) + user_logs.append(history_path.with_suffix(".json")) + + statistics = await Config.merge_statistic_info(user_logs) + statistics["user_info"] = self.cur_user_item.name + statistics["start_time"] = self.user_start_time.strftime("%Y-%m-%d %H:%M:%S") + statistics["end_time"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + statistics["user_result"] = ( + "代理任务全部完成" if self.run_book else self.cur_user_item.result + ) + try: + await push_notification( + "统计信息", + f"{datetime.now().strftime('%m-%d')} |{'√' if self.run_book else 'X'}| " + f"{self.cur_user_item.name} 的自动代理统计报告", + statistics, + self.cur_user_config, + ) + except Exception as error: + logger.warning(f"推送通知时出现异常: {error}") + + async def final_task(self) -> None: + """停止运行、写历史记录、恢复配置并更新用户状态。""" + + await self._stop_log_monitor() + await self.kill_managed_process() + if self.check_result == "Pass": + await self._save_history() + # 回写用户副本后恢复原配置；直控配置始终不被任务结果污染。 + self._restore_external_config() + self._cleanup_external_config_snapshot() + + if self.run_book: + if ( + self.cur_user_config.get("Data", "ProxyTimes") == 0 + and self.cur_user_config.get("Info", "RemainedDay") != -1 + ): + await self.cur_user_config.set( + "Info", + "RemainedDay", + self.cur_user_config.get("Info", "RemainedDay") - 1, + ) + await self.cur_user_config.set( + "Data", + "ProxyTimes", + self.cur_user_config.get("Data", "ProxyTimes") + 1, + ) + self.cur_user_item.status = "完成" + await Notify.push_plyer( + "成功完成一个自动代理任务！", + f"已完成用户 {self.cur_user_item.name} 的自动代理任务", + f"已完成 {self.cur_user_item.name} 的自动代理任务", + 3, + ) + elif self.check_result == "Pass": + self.cur_user_item.status = "异常" + + async def on_crash(self, error: Exception) -> None: + """异常路径必须可重复执行，并向调度台报告 Error。""" + + self.cur_user_item.status = "异常" + logger.opt(exception=True).warning(f"专项自动代理任务出现异常: {error}") + await self._stop_log_monitor() + await self.kill_managed_process() + self._restore_external_config() + self._cleanup_external_config_snapshot() + with suppress(Exception): + await Config.send_websocket_message( + id=self.task_info.task_id, + type="Info", + data={"Error": f"专项自动代理任务出现异常: {error}"}, + ) diff --git a/templates/specialized/backend/task/Xxx/ScriptConfig.py b/templates/specialized/backend/task/Xxx/ScriptConfig.py new file mode 100644 index 000000000..9ebc65d37 --- /dev/null +++ b/templates/specialized/backend/task/Xxx/ScriptConfig.py @@ -0,0 +1,195 @@ +"""Xxx 原生配置会话任务。""" + +from __future__ import annotations + +import asyncio +import shutil +import uuid +from contextlib import suppress +from pathlib import Path + +from app.core import Config +from app.models.ConfigBase import MultipleConfig +from app.models.config import XxxConfig, XxxUserConfig +from app.models.emulator import DeviceBase +from app.models.task import ScriptItem, TaskExecuteBase +from app.services import System +from app.utils import ProcessManager, get_logger +from .AutoProxy import _split_script_arguments + + +logger = get_logger("专项显示名称配置") + + +class ScriptConfigTask(TaskExecuteBase): + """启动上游配置界面，停止任务时把配置保存回用户副本。""" + + def __init__( + self, + script_info: ScriptItem, + script_config: XxxConfig, + user_config: MultipleConfig[XxxUserConfig], + game_manager: ProcessManager | DeviceBase | None, + ) -> None: + super().__init__() + if script_info.task_info is None: + raise RuntimeError("ScriptItem 未绑定到 TaskItem") + + self.task_info = script_info.task_info + self.script_info = script_info + self.script_config = script_config + self.user_config = user_config + self.game_manager = game_manager + self.cur_user_item = self.script_info.user_list[self.script_info.current_index] + self.use_mas_config = True + if self.cur_user_item.user_id != "Default": + self.use_mas_config = bool( + self.user_config[uuid.UUID(self.cur_user_item.user_id)].get( + "Info", "IfUseMasConfig" + ) + ) + + self.general_process_manager: ProcessManager | None = None + self.wait_event: asyncio.Event | None = None + self.script_path: Path | None = None + self.script_set_exe_path: Path | None = None + self.script_set_arguments: list[str] = [] + self.script_config_path: Path | None = None + self.configuration_started = False + + def _user_config_path(self) -> Path: + if self.cur_user_item.user_id == "Default": + return Path.cwd() / "data" / self.script_info.script_id / "Default" / "ConfigFile" + return ( + Path.cwd() + / "data" + / self.script_info.script_id + / self.cur_user_item.user_id + / "ConfigFile" + ) + + def _user_config_source_path(self) -> Path: + user_path = self._user_config_path() + if self.script_config.get("Script", "ConfigPathMode") == "Folder": + return user_path + return user_path / Path(self.script_config.get("Script", "ConfigPath")).name + + def _remove_path(self, path: Path) -> None: + if path.is_dir(): + shutil.rmtree(path) + elif path.exists(): + path.unlink() + + def _copy_path_atomic(self, source: Path, destination: Path, mode: str) -> None: + if mode == "Folder": + temporary = destination.with_name(destination.name + ".tmp") + shutil.rmtree(temporary, ignore_errors=True) + if source.exists(): + temporary.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(source, temporary, dirs_exist_ok=True) + self._remove_path(destination) + if temporary.exists(): + temporary.rename(destination) + return + + temporary = destination.with_name(destination.name + ".tmp") + if temporary.exists(): + temporary.unlink() + if source.exists(): + temporary.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, temporary) + temporary.replace(destination) + + async def prepare(self) -> None: + self.general_process_manager = ProcessManager() + self.wait_event = asyncio.Event() + self.script_path = Path(self.script_config.get("Script", "ScriptPath")) + argument_paths, argument_lists = _split_script_arguments( + self.script_config.get("Script", "Arguments"), self.script_path + ) + self.script_set_exe_path = ( + argument_paths[1] if len(argument_paths) > 1 else self.script_path + ) + self.script_set_arguments = argument_lists[1] if len(argument_lists) > 1 else [] + self.script_config_path = Path(self.script_config.get("Script", "ConfigPath")) + + async def set_script_config(self) -> None: + """配置会话启动前导入用户副本。""" + + if self.script_config_path is None: + return + await System.kill_process(self.script_set_exe_path) + if not self.use_mas_config: + logger.info("脚本直控配置：跳过导入用户配置") + return + source = self._user_config_source_path() + if not source.exists(): + logger.info("用户副本尚未创建，沿用脚本当前配置并在会话结束时保存") + return + self._copy_path_atomic( + source, + self.script_config_path, + self.script_config.get("Script", "ConfigPathMode"), + ) + + async def main_task(self) -> None: + await self.prepare() + await self.set_script_config() + if ( + self.general_process_manager is None + or self.script_set_exe_path is None + or self.wait_event is None + ): + raise RuntimeError("专项配置会话未完成初始化") + logger.info( + f"启动专项配置会话: {self.script_set_exe_path}, 参数: {self.script_set_arguments}" + ) + await self.general_process_manager.open_process( + self.script_set_exe_path, + *self.script_set_arguments, + ) + self.configuration_started = True + await self.wait_event.wait() + + async def final_task(self) -> None: + """停止原生配置进程并按模式保存用户副本。""" + + if self.general_process_manager is not None: + try: + await self.general_process_manager.kill() + except Exception as error: + logger.warning(f"停止专项配置进程失败: {error}") + if self.script_set_exe_path is not None: + try: + await System.kill_process(self.script_set_exe_path) + except Exception as error: + logger.warning(f"清理专项配置进程失败: {error}") + + if ( + not self.configuration_started + or not self.use_mas_config + or self.script_config_path is None + ): + logger.info("脚本直控配置：跳过保存用户副本") + return + self._copy_path_atomic( + self.script_config_path, + self._user_config_source_path(), + self.script_config.get("Script", "ConfigPathMode"), + ) + logger.success("专项配置已保存到用户副本") + + async def on_crash(self, error: Exception) -> None: + self.cur_user_item.status = "异常" + logger.opt(exception=True).warning(f"专项配置会话出现异常: {error}") + if self.script_set_exe_path is not None: + try: + await System.kill_process(self.script_set_exe_path) + except Exception as cleanup_error: + logger.warning(f"清理专项配置进程失败: {cleanup_error}") + with suppress(Exception): + await Config.send_websocket_message( + id=self.task_info.task_id, + type="Info", + data={"Error": f"专项配置会话出现异常: {error}"}, + ) diff --git a/templates/specialized/backend/task/Xxx/__init__.py b/templates/specialized/backend/task/Xxx/__init__.py new file mode 100644 index 000000000..e4b6d96c8 --- /dev/null +++ b/templates/specialized/backend/task/Xxx/__init__.py @@ -0,0 +1,5 @@ +"""Xxx 专项任务入口。""" + +from .manager import XxxManager + +__all__ = ["XxxManager"] diff --git a/templates/specialized/backend/task/Xxx/manager.py b/templates/specialized/backend/task/Xxx/manager.py new file mode 100644 index 000000000..5f26589e8 --- /dev/null +++ b/templates/specialized/backend/task/Xxx/manager.py @@ -0,0 +1,286 @@ +"""Xxx 专项调度器。""" + +from __future__ import annotations + +import shutil +import uuid +from contextlib import suppress +from datetime import datetime +from pathlib import Path + +from app.core import Config, EmulatorManager +from app.models.ConfigBase import MultipleConfig +from app.models.config import XxxConfig, XxxUserConfig +from app.models.emulator import DeviceBase +from app.models.task import ScriptItem, TaskExecuteBase, UserItem +from app.services import Notify +from app.utils import ProcessManager, get_logger +from .AutoProxy import AutoProxyTask +from .ScriptConfig import ScriptConfigTask + + +logger = get_logger("专项显示名称调度器") + +METHOD_BOOK: dict[str, type[AutoProxyTask | ScriptConfigTask]] = { + "AutoProxy": AutoProxyTask, + "ScriptConfig": ScriptConfigTask, +} + + +class XxxManager(TaskExecuteBase): + """协调脚本锁、用户列表、任务子类和脚本直控配置快照。""" + + def __init__(self, script_info: ScriptItem) -> None: + super().__init__() + if script_info.task_info is None: + raise RuntimeError("ScriptItem 未绑定到 TaskItem") + + self.task_info = script_info.task_info + self.script_info = script_info + self.check_result = "-" + self.script_config: XxxConfig | None = None + self.user_config: MultipleConfig[XxxUserConfig] | None = None + self.script_config_path: Path | None = None + self.temp_path: Path | None = None + self.external_config_exists = False + self.external_config_snapshot_ready = False + self.emulator_manager: DeviceBase | None = None + self.game_process_manager: ProcessManager | None = None + self.begin_time = "" + + async def check(self) -> str: + """检查任务模式、脚本类型和游戏配置。""" + + if self.task_info.mode not in METHOD_BOOK: + return "当前专项不支持该任务模式" + + script_config = Config.ScriptConfig[uuid.UUID(self.script_info.script_id)] + if not isinstance(script_config, XxxConfig): + return "脚本配置类型错误，请重新选择专项脚本" + + if ( + script_config.get("Script", "IfTrackProcess") + and not script_config.get("Script", "TrackProcessName") + and not script_config.get("Script", "TrackProcessExe") + and not script_config.get("Script", "TrackProcessCmdline") + ): + return "请至少填写一项目标进程信息" + + if not script_config.get("Game", "Enabled"): + return "Pass" + game_type = script_config.get("Game", "Type") + if game_type == "Emulator" and ( + script_config.get("Game", "EmulatorId") == "-" + or script_config.get("Game", "EmulatorIndex") in ("", "-") + ): + return "请完成模拟器配置" + if game_type == "Client" and not Path(script_config.get("Game", "Path")).exists(): + return "请设置游戏或启动器路径" + if game_type == "URL" and not ( + script_config.get("Game", "URL") + and script_config.get("Game", "ProcessName") + ): + return "请填写游戏 URL 和进程名称" + return "Pass" + + def _remove_path(self, path: Path) -> None: + if path.is_dir(): + shutil.rmtree(path) + elif path.exists(): + path.unlink() + + def _copy_path_atomic(self, source: Path, destination: Path, mode: str) -> None: + """恢复直控配置时先写临时路径，再替换目标。""" + + if mode == "Folder": + temporary = destination.with_name(destination.name + ".tmp") + shutil.rmtree(temporary, ignore_errors=True) + temporary.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree(source, temporary, dirs_exist_ok=True) + self._remove_path(destination) + temporary.rename(destination) + return + + temporary = destination.with_name(destination.name + ".tmp") + if temporary.exists(): + temporary.unlink() + temporary.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, temporary) + temporary.replace(destination) + + def _snapshot_external_config(self) -> None: + """保存脚本直控配置，隔离多个用户之间的配置变更。""" + + if self.script_config is None or self.script_config_path is None: + return + self.temp_path = Path.cwd() / "data" / self.script_info.script_id / "Temp" / "manager" + shutil.rmtree(self.temp_path, ignore_errors=True) + self.external_config_exists = self.script_config_path.exists() + self.temp_path.mkdir(parents=True, exist_ok=True) + if self.external_config_exists: + if self.script_config.get("Script", "ConfigPathMode") == "Folder": + shutil.copytree(self.script_config_path, self.temp_path, dirs_exist_ok=True) + else: + shutil.copy2(self.script_config_path, self.temp_path / "config.temp") + self.external_config_snapshot_ready = True + + def _restore_external_config(self) -> None: + if ( + not self.external_config_snapshot_ready + or self.script_config is None + or self.script_config_path is None + or self.temp_path is None + ): + return + self._remove_path(self.script_config_path) + if not self.external_config_exists: + return + if self.script_config.get("Script", "ConfigPathMode") == "Folder": + self._copy_path_atomic(self.temp_path, self.script_config_path, "Folder") + else: + self._copy_path_atomic( + self.temp_path / "config.temp", self.script_config_path, "File" + ) + + def _cleanup_external_config_snapshot(self) -> None: + if self.temp_path is not None: + shutil.rmtree(self.temp_path, ignore_errors=True) + self.external_config_snapshot_ready = False + + def _user_uses_mas_config(self) -> bool: + if self.user_config is None: + return True + user_id = self.script_info.user_list[self.script_info.current_index].user_id + if user_id == "Default": + return True + return bool(self.user_config[uuid.UUID(user_id)].get("Info", "IfUseMasConfig")) + + async def prepare(self) -> None: + """锁定脚本并构建本次任务的用户列表。""" + + script_uid = uuid.UUID(self.script_info.script_id) + await Config.ScriptConfig[script_uid].lock() + self.script_config = Config.ScriptConfig[script_uid] + self.user_config = MultipleConfig([XxxUserConfig]) + await self.user_config.load(await self.script_config.UserData.toDict()) + self.script_config_path = Path(self.script_config.get("Script", "ConfigPath")) + + if self.script_config.get("Game", "Enabled"): + game_type = self.script_config.get("Game", "Type") + if game_type == "Emulator": + self.emulator_manager = await EmulatorManager.get_emulator_instance( + self.script_config.get("Game", "EmulatorId") + ) + elif game_type in ("Client", "URL"): + self.game_process_manager = ProcessManager() + + if self.task_info.mode == "ScriptConfig": + self.script_info.user_list = [ + UserItem( + user_id=self.task_info.user_id or "Default", + name="", + status="等待", + ) + ] + else: + self.script_info.user_list = [ + UserItem(user_id=str(uid), name=config.get("Info", "Name"), status="等待") + for uid, config in self.user_config.items() + if config.get("Info", "Status") and config.get("Info", "RemainedDay") != 0 + ] + self._snapshot_external_config() + logger.info(f"专项用户列表加载完成: {len(self.script_info.user_list)}") + + async def main_task(self) -> None: + self.check_result = await self.check() + if self.check_result != "Pass": + self.script_info.status = "异常" + await Config.send_websocket_message( + id=self.task_info.task_id, + type="Info", + data={"Error": self.check_result}, + ) + return + + self.begin_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + await self.prepare() + if self.script_config is None or self.user_config is None: + raise RuntimeError("专项配置未初始化") + + for self.script_info.current_index in range(len(self.script_info.user_list)): + use_mas_config = self._user_uses_mas_config() + user_id = self.script_info.user_list[self.script_info.current_index].user_id + logger.info( + f"用户 {user_id} 配置来源: {'MAS 独立配置' if use_mas_config else '脚本直控配置'}" + ) + if not use_mas_config: + self._restore_external_config() + + task = METHOD_BOOK[self.task_info.mode]( + self.script_info, + self.script_config, + self.user_config, + ( + self.emulator_manager + if self.script_config.get("Game", "Type") == "Emulator" + else self.game_process_manager + ) + if self.script_config.get("Game", "Enabled") + else None, + ) + try: + await self.spawn(task) + finally: + if not use_mas_config: + self._snapshot_external_config() + + async def final_task(self) -> None: + """解锁、写回用户数据并聚合脚本状态。""" + + if self.check_result != "Pass": + self.script_info.status = "异常" + return + + self._restore_external_config() + self._cleanup_external_config_snapshot() + script_uid = uuid.UUID(self.script_info.script_id) + script_config = Config.ScriptConfig[script_uid] + if script_config.is_locked: + await script_config.unlock() + + # unlock-then-write：ConfigBase 在锁定状态下拒绝 load。 + if self.task_info.mode == "AutoProxy" and self.user_config is not None: + await script_config.UserData.load(await self.user_config.toDict()) + await Config.ScriptConfig.save() + + has_error = any(user.status == "异常" for user in self.script_info.user_list) + has_success = any(user.status == "完成" for user in self.script_info.user_list) + self.script_info.status = "异常" if has_error else "完成" + await Notify.push_plyer( + "专项自动代理任务已结束", + f"已完成用户数: {sum(user.status == '完成' for user in self.script_info.user_list)}，" + f"异常用户数: {sum(user.status == '异常' for user in self.script_info.user_list)}", + self.script_info.result, + 10 if has_success else 3, + ) + + async def on_crash(self, error: Exception) -> None: + self.script_info.status = "异常" + logger.opt(exception=True).warning(f"专项调度任务出现异常: {error}") + try: + self._restore_external_config() + self._cleanup_external_config_snapshot() + except Exception as restore_error: + logger.warning(f"恢复专项脚本配置失败: {restore_error}") + try: + script_config = Config.ScriptConfig[uuid.UUID(self.script_info.script_id)] + if script_config.is_locked: + await script_config.unlock() + except Exception as unlock_error: + logger.warning(f"解锁专项脚本配置失败: {unlock_error}") + with suppress(Exception): + await Config.send_websocket_message( + id=self.task_info.task_id, + type="Info", + data={"Error": f"专项调度任务出现异常: {error}"}, + ) diff --git a/templates/specialized/frontend/XxxScriptEdit.vue b/templates/specialized/frontend/XxxScriptEdit.vue new file mode 100644 index 000000000..93f3204e4 --- /dev/null +++ b/templates/specialized/frontend/XxxScriptEdit.vue @@ -0,0 +1,536 @@ + + + + + diff --git a/templates/specialized/frontend/XxxUserEdit.vue b/templates/specialized/frontend/XxxUserEdit.vue new file mode 100644 index 000000000..e37ac4158 --- /dev/null +++ b/templates/specialized/frontend/XxxUserEdit.vue @@ -0,0 +1,392 @@ + + + + + diff --git a/templates/specialized/frontend/XxxUserEdit/BasicInfoSection.vue b/templates/specialized/frontend/XxxUserEdit/BasicInfoSection.vue new file mode 100644 index 000000000..6b7dfab4a --- /dev/null +++ b/templates/specialized/frontend/XxxUserEdit/BasicInfoSection.vue @@ -0,0 +1,172 @@ + + + + + diff --git a/templates/specialized/frontend/XxxUserEdit/NotifyConfigSection.vue b/templates/specialized/frontend/XxxUserEdit/NotifyConfigSection.vue new file mode 100644 index 000000000..2ca4fe27d --- /dev/null +++ b/templates/specialized/frontend/XxxUserEdit/NotifyConfigSection.vue @@ -0,0 +1,170 @@ + + + + + diff --git a/templates/specialized/tests/test_xxx_autoproxy.py b/templates/specialized/tests/test_xxx_autoproxy.py new file mode 100644 index 000000000..0c6a2af19 --- /dev/null +++ b/templates/specialized/tests/test_xxx_autoproxy.py @@ -0,0 +1,104 @@ +"""Xxx AutoProxy 最小回归测试。 + +复制到 ``tests/task/test_xxx_autoproxy.py`` 后，将 import 路径和夹具替换为真实 +专项配置。测试不启动外部脚本，只固定参数解析、日志文件名前缀和配置恢复边界。 +""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +try: + from app.task.Xxx.AutoProxy import ( + AutoProxyTask, + _format_to_prefix_regex, + _split_script_arguments, + ) +except ImportError: + # 模板尚未复制进 app/task 时允许仓库全量 pytest 收集并跳过本文件。 + AutoProxyTask = None + + +class _ScriptConfigStub: + def __init__(self, mode: str) -> None: + self.mode = mode + + def get(self, section: str, key: str): + if (section, key) == ("Script", "ConfigPathMode"): + return self.mode + raise AssertionError(f"unexpected config lookup: {section}.{key}") + + +def _build_task(root: Path, mode: str) -> AutoProxyTask: + task = object.__new__(AutoProxyTask) + task.script_config = _ScriptConfigStub(mode) + task.script_config_path = root / "script-config" + task.temp_path = root / "temp" + task.external_config_exists = False + task.external_config_snapshot_ready = False + return task + + +@unittest.skipUnless(AutoProxyTask is not None, "请先将模板复制并注册为 Xxx 专项") +class XxxAutoProxyTest(unittest.TestCase): + def test_split_script_arguments_keeps_executable_and_arguments(self) -> None: + paths, arguments = _split_script_arguments( + "runner.exe%--headless --task 1|config.exe%--settings", + Path("C:/Xxx"), + ) + self.assertEqual(paths[0], Path("C:/Xxx/runner.exe").resolve()) + self.assertEqual(arguments[0], ["--headless", "--task", "1"]) + self.assertEqual(arguments[1], ["--settings"]) + + def test_log_prefix_pattern_supports_strptime_directives(self) -> None: + pattern = _format_to_prefix_regex("%Y-%m-%d") + self.assertIsNotNone(pattern.match("2026-08-21-1.log")) + self.assertIsNone(pattern.match("not-a-date.log")) + + def test_folder_snapshot_restores_original_config(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + task = _build_task(root, "Folder") + task.script_config_path.mkdir() + (task.script_config_path / "keep.json").write_text("direct", encoding="utf-8") + + task._snapshot_external_config() + (task.script_config_path / "keep.json").unlink() + (task.script_config_path / "managed.json").write_text("managed", encoding="utf-8") + task._restore_external_config() + + self.assertEqual( + (task.script_config_path / "keep.json").read_text(encoding="utf-8"), + "direct", + ) + self.assertFalse((task.script_config_path / "managed.json").exists()) + + def test_file_snapshot_restores_original_config(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + task = _build_task(root, "File") + task.script_config_path.write_text("direct", encoding="utf-8") + + task._snapshot_external_config() + task.script_config_path.write_text("managed", encoding="utf-8") + task._restore_external_config() + + self.assertEqual(task.script_config_path.read_text(encoding="utf-8"), "direct") + + def test_missing_config_remains_missing_after_restore(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + task = _build_task(root, "Folder") + task._snapshot_external_config() + task.script_config_path.mkdir() + (task.script_config_path / "managed.json").write_text("managed", encoding="utf-8") + + task._restore_external_config() + + self.assertFalse(task.script_config_path.exists()) + + +if __name__ == "__main__": + unittest.main()