automas_admin/AUTO-MAS-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 适配添加脚本功能

Browse Source
This commit is contained in:
DLmaster361
2025-08-04 01:12:06 +08:00
parent a4891131fc
commit b576440612
4 changed files with 82 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@@ -16,3 +16,7 @@ logs/
*.swp *.swp
*.swo *.swo
.DS_Store .DS_Store
# Debug files
debug/
config/

122
app/api/api.py
View File

@@ -1,37 +1,49 @@
from fastapi import FastAPI, HTTPException, Path, Body from fastapi import FastAPI, HTTPException, Path, Body
from pydantic import BaseModel from pydantic import BaseModel, Field
from typing import Dict, Any, List, Optional from typing import Dict, Any, List, Optional, Literal
from datetime import datetime from datetime import datetime
from app.core import Config, logger
app = FastAPI( app = FastAPI(
title="auto_maa", title="AUTO_MAA",
description="API for managing automation scripts, plans, and tasks", description="API for managing automation scripts, plans, and tasks",
version="1.0.0" version="1.0.0",
) )
#此文件由ai生成 返回值非最终版本 # 此文件由ai生成 返回值非最终版本
# ====================== # ======================
# Data Models # Data Models
# ====================== # ======================
# Script Models # Script Models
class ScriptCreate(BaseModel): class ScriptCreateIn(BaseModel):
name: str type: Literal["MAA", "General"]
type: str # "MAA" or "通用"
content: str
description: Optional[str] = None class ScriptCreateOut(BaseModel):
code: int = Field(default=200, description="状态码")
status: str = Field(default="success", description="操作状态")
message: str = Field(default="脚本添加成功", description="操作消息")
scriptId: str = Field(..., description="新创建的脚本ID")
data: Dict[str, Any] = Field(..., description="脚本配置数据")
class ScriptUpdate(BaseModel): class ScriptUpdate(BaseModel):
name: Optional[str] = None name: Optional[str] = None
content: Optional[str] = None content: Optional[str] = None
description: Optional[str] = None description: Optional[str] = None
class ScriptUser(BaseModel): class ScriptUser(BaseModel):
userId: str userId: str
config: Dict[str, Any] = {} config: Dict[str, Any] = {}
# Plan Models # Plan Models
class PlanDayConfig(BaseModel): class PlanDayConfig(BaseModel):
吃理智药: int 吃理智药: int
@@ -42,6 +54,7 @@ class PlanDayConfig(BaseModel):
备选_3: str 备选_3: str
剩余理智: str 剩余理智: str
class PlanDetails(BaseModel): class PlanDetails(BaseModel):
周一: PlanDayConfig 周一: PlanDayConfig
周二: PlanDayConfig 周二: PlanDayConfig
@@ -51,19 +64,23 @@ class PlanDetails(BaseModel):
周六: PlanDayConfig 周六: PlanDayConfig
周日: PlanDayConfig 周日: PlanDayConfig
class PlanCreate(BaseModel): class PlanCreate(BaseModel):
name: str name: str
mode: str # "全局" or "周计划" mode: str # "全局" or "周计划"
details: PlanDetails details: PlanDetails
class PlanUpdate(BaseModel): class PlanUpdate(BaseModel):
name: Optional[str] = None name: Optional[str] = None
mode: Optional[str] = None mode: Optional[str] = None
details: Optional[PlanDetails] = None details: Optional[PlanDetails] = None
class PlanModeUpdate(BaseModel): class PlanModeUpdate(BaseModel):
mode: str # "全局" or "周计划" mode: str # "全局" or "周计划"
# Queue Models # Queue Models
class QueueCreate(BaseModel): class QueueCreate(BaseModel):
name: str name: str
@@ -71,12 +88,14 @@ class QueueCreate(BaseModel):
schedule: str schedule: str
description: Optional[str] = None description: Optional[str] = None
class QueueUpdate(BaseModel): class QueueUpdate(BaseModel):
name: Optional[str] = None name: Optional[str] = None
scripts: Optional[List[str]] = None scripts: Optional[List[str]] = None
schedule: Optional[str] = None schedule: Optional[str] = None
description: Optional[str] = None description: Optional[str] = None
# Task Models # Task Models
class TaskCreate(BaseModel): class TaskCreate(BaseModel):
name: str name: str
@@ -86,15 +105,18 @@ class TaskCreate(BaseModel):
priority: int = 0 priority: int = 0
parameters: Dict[str, Any] = {} parameters: Dict[str, Any] = {}
# Settings Models # Settings Models
class SettingsUpdate(BaseModel): class SettingsUpdate(BaseModel):
key: str key: str
value: Any value: Any
# ====================== # ======================
# API Endpoints # API Endpoints
# ====================== # ======================
@app.get("/api/activity/latest", summary="获取最新活动内容") @app.get("/api/activity/latest", summary="获取最新活动内容")
async def get_latest_activity(): async def get_latest_activity():
""" """
@@ -104,14 +126,17 @@ async def get_latest_activity():
return {"status": "success", "data": {}} return {"status": "success", "data": {}}
@app.post("/api/add/scripts", summary="添加脚本") @app.post(
async def add_script(script: ScriptCreate = Body(...)): "/api/add/scripts",
""" summary="添加脚本",
添加脚本 response_model=ScriptCreateOut,
这里后端需要支持两种脚本，MAA和通用 status_code=200,
""" )
# 实现添加脚本的逻辑 async def add_script(script: ScriptCreateIn = Body(...)) -> ScriptCreateOut:
return {"status": "success", "scriptId": "new_script_id"} """添加脚本"""
uid, config = await Config.add_script(script.type)
return ScriptCreateOut(scriptId=str(uid), data=config.toDict())
@app.post("/api/get/scripts", summary="查询脚本") @app.post("/api/get/scripts", summary="查询脚本")
@@ -124,9 +149,7 @@ async def get_scripts():
@app.post("/api/get/scripts/{scriptId}", summary="查询单个脚本") @app.post("/api/get/scripts/{scriptId}", summary="查询单个脚本")
async def get_script( async def get_script(scriptId: str = Path(..., description="脚本ID")):
scriptId: str = Path(..., description="脚本ID")
):
""" """
查询单个脚本 查询单个脚本
""" """
@@ -137,7 +160,7 @@ async def get_script(
@app.post("/api/update/scripts/{scriptId}", summary="更新脚本") @app.post("/api/update/scripts/{scriptId}", summary="更新脚本")
async def update_script( async def update_script(
scriptId: str = Path(..., description="脚本ID"), scriptId: str = Path(..., description="脚本ID"),
update_data: ScriptUpdate = Body(...) update_data: ScriptUpdate = Body(...),
): ):
""" """
更新脚本 更新脚本
@@ -147,9 +170,7 @@ async def update_script(
@app.post("/api/delete/scripts/{scriptId}", summary="删除脚本") @app.post("/api/delete/scripts/{scriptId}", summary="删除脚本")
async def delete_script( async def delete_script(scriptId: str = Path(..., description="脚本ID")):
scriptId: str = Path(..., description="脚本ID")
):
""" """
删除脚本 删除脚本
""" """
@@ -159,8 +180,7 @@ async def delete_script(
@app.post("/api/scripts/{scriptId}/users", summary="为脚本添加用户") @app.post("/api/scripts/{scriptId}/users", summary="为脚本添加用户")
async def add_script_user( async def add_script_user(
scriptId: str = Path(..., description="脚本ID"), scriptId: str = Path(..., description="脚本ID"), user: ScriptUser = Body(...)
user: ScriptUser = Body(...)
): ):
""" """
为脚本添加用户 为脚本添加用户
@@ -170,9 +190,7 @@ async def add_script_user(
@app.get("/api/scripts/{scriptId}/users", summary="查询脚本的所有下属用户") @app.get("/api/scripts/{scriptId}/users", summary="查询脚本的所有下属用户")
async def get_script_users( async def get_script_users(scriptId: str = Path(..., description="脚本ID")):
scriptId: str = Path(..., description="脚本ID")
):
""" """
查询脚本的所有下属用户 查询脚本的所有下属用户
""" """
@@ -183,7 +201,7 @@ async def get_script_users(
@app.get("/api/scripts/{scriptId}/users/{userId}", summary="查询脚本下的单个下属用户") @app.get("/api/scripts/{scriptId}/users/{userId}", summary="查询脚本下的单个下属用户")
async def get_script_user( async def get_script_user(
scriptId: str = Path(..., description="脚本ID"), scriptId: str = Path(..., description="脚本ID"),
userId: str = Path(..., description="用户ID") userId: str = Path(..., description="用户ID"),
): ):
""" """
查询脚本下的单个下属用户 查询脚本下的单个下属用户
@@ -196,7 +214,7 @@ async def get_script_user(
async def update_script_user( async def update_script_user(
scriptId: str = Path(..., description="脚本ID"), scriptId: str = Path(..., description="脚本ID"),
userId: str = Path(..., description="用户ID"), userId: str = Path(..., description="用户ID"),
config: Dict[str, Any] = Body(...) config: Dict[str, Any] = Body(...),
): ):
""" """
更新脚本下属用户的关联信息 更新脚本下属用户的关联信息
@@ -208,7 +226,7 @@ async def update_script_user(
@app.delete("/api/scripts/{scriptId}/users/{userId}", summary="从脚本移除用户") @app.delete("/api/scripts/{scriptId}/users/{userId}", summary="从脚本移除用户")
async def remove_script_user( async def remove_script_user(
scriptId: str = Path(..., description="脚本ID"), scriptId: str = Path(..., description="脚本ID"),
userId: str = Path(..., description="用户ID") userId: str = Path(..., description="用户ID"),
): ):
""" """
从脚本移除用户 从脚本移除用户
@@ -252,9 +270,7 @@ async def get_plans():
@app.post("/api/get/plans/{planId}", summary="查询单个计划") @app.post("/api/get/plans/{planId}", summary="查询单个计划")
async def get_plan( async def get_plan(planId: str = Path(..., description="计划ID")):
planId: str = Path(..., description="计划ID")
):
""" """
查询单个计划 查询单个计划
""" """
@@ -264,8 +280,7 @@ async def get_plan(
@app.post("/api/update/plans/{planId}", summary="更新计划") @app.post("/api/update/plans/{planId}", summary="更新计划")
async def update_plan( async def update_plan(
planId: str = Path(..., description="计划ID"), planId: str = Path(..., description="计划ID"), update_data: PlanUpdate = Body(...)
update_data: PlanUpdate = Body(...)
): ):
""" """
更新计划 更新计划
@@ -275,9 +290,7 @@ async def update_plan(
@app.post("/api/delete/plans/{planId}", summary="删除计划") @app.post("/api/delete/plans/{planId}", summary="删除计划")
async def delete_plan( async def delete_plan(planId: str = Path(..., description="计划ID")):
planId: str = Path(..., description="计划ID")
):
""" """
删除计划 删除计划
""" """
@@ -287,8 +300,7 @@ async def delete_plan(
@app.post("/api/update/plans/{planId}/mode", summary="切换计划模式") @app.post("/api/update/plans/{planId}/mode", summary="切换计划模式")
async def update_plan_mode( async def update_plan_mode(
planId: str = Path(..., description="计划ID"), planId: str = Path(..., description="计划ID"), mode_data: PlanModeUpdate = Body(...)
mode_data: PlanModeUpdate = Body(...)
): ):
""" """
切换计划模式 切换计划模式
@@ -319,9 +331,7 @@ async def get_queues():
@app.post("/api/get/queues/{queueId}", summary="查询单个调度队列详情") @app.post("/api/get/queues/{queueId}", summary="查询单个调度队列详情")
async def get_queue( async def get_queue(queueId: str = Path(..., description="调度队列ID")):
queueId: str = Path(..., description="调度队列ID")
):
""" """
查询单个调度队列详情 查询单个调度队列详情
""" """
@@ -332,7 +342,7 @@ async def get_queue(
@app.post("/api/update/queues/{queueId}", summary="更新调度队列") @app.post("/api/update/queues/{queueId}", summary="更新调度队列")
async def update_queue( async def update_queue(
queueId: str = Path(..., description="调度队列ID"), queueId: str = Path(..., description="调度队列ID"),
update_data: QueueUpdate = Body(...) update_data: QueueUpdate = Body(...),
): ):
""" """
更新调度队列 更新调度队列
@@ -342,9 +352,7 @@ async def update_queue(
@app.post("/api/delete/queues/{queueId}", summary="删除调度队列") @app.post("/api/delete/queues/{queueId}", summary="删除调度队列")
async def delete_queue( async def delete_queue(queueId: str = Path(..., description="调度队列ID")):
queueId: str = Path(..., description="调度队列ID")
):
""" """
删除调度队列 删除调度队列
""" """
@@ -362,9 +370,7 @@ async def add_task(task: TaskCreate = Body(...)):
@app.post("/api/tasks/{taskId}/start", summary="开始任务") @app.post("/api/tasks/{taskId}/start", summary="开始任务")
async def start_task( async def start_task(taskId: str = Path(..., description="任务ID")):
taskId: str = Path(..., description="任务ID")
):
""" """
开始任务 开始任务
""" """
@@ -394,15 +400,13 @@ async def update_settings(settings: SettingsUpdate = Body(...)):
# Error Handlers # Error Handlers
# ====================== # ======================
@app.exception_handler(HTTPException) @app.exception_handler(HTTPException)
async def http_exception_handler(request, exc): async def http_exception_handler(request, exc):
return { return {"status": "error", "code": exc.status_code, "message": exc.detail}
"status": "error",
"code": exc.status_code,
"message": exc.detail
}
if __name__ == "__main__": if __name__ == "__main__":
import uvicorn import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
uvicorn.run(app, host="0.0.0.0", port=8000)

11
app/core/config.py
View File

@@ -32,7 +32,7 @@ from collections import defaultdict
from pathlib import Path from pathlib import Path
from typing import Union, Dict, List from typing import Union, Dict, List, Literal
from .logger import logger from .logger import logger
from app.models.ConfigBase import * from app.models.ConfigBase import *
@@ -641,6 +641,15 @@ class AppConfig(GlobalConfig):
logger.info(f"根目录： {self.root_path}", module="配置管理") logger.info(f"根目录： {self.root_path}", module="配置管理")
logger.info("===================================", module="配置管理") logger.info("===================================", module="配置管理")
async def add_script(
self, script: Literal["MAA", "General"]
) -> tuple[uuid.UUID, ConfigBase]:
"""添加脚本配置"""
class_book = {"MAA": MaaConfig, "General": GeneralConfig}
return self.ScriptConfig.add(class_book[script])
# def check_data(self) -> None: # def check_data(self) -> None:
# """检查用户数据文件并处理数据文件版本更新""" # """检查用户数据文件并处理数据文件版本更新"""

10
app/models/ConfigBase.py
View File

@@ -253,7 +253,7 @@ class ConfigBase:
if self.file: if self.file:
self.save() self.save()
def toDict(self): def toDict(self) -> Dict[str, Any]:
"""将配置项转换为字典""" """将配置项转换为字典"""
data = {} data = {}
@@ -482,7 +482,7 @@ class MultipleConfig:
with self.file.open("w", encoding="utf-8") as f: with self.file.open("w", encoding="utf-8") as f:
json.dump(self.toDict(), f, ensure_ascii=False, indent=4) json.dump(self.toDict(), f, ensure_ascii=False, indent=4)
def add(self, config_type: type) -> Any: def add(self, config_type: type) -> tuple[uuid.UUID, ConfigBase]:
""" """
添加一个新的配置项 添加一个新的配置项
@@ -493,8 +493,8 @@ class MultipleConfig:
Returns Returns
------- -------
Any tuple[uuid.UUID, ConfigBase]
新创建的配置项实例 新创建的配置项的唯一标识符和实例
""" """
if config_type not in self.sub_config_type: if config_type not in self.sub_config_type:
@@ -507,7 +507,7 @@ class MultipleConfig:
if self.file: if self.file:
self.save() self.save()
return self.data[uid] return uid, self.data[uid]
def remove(self, uid: uuid.UUID): def remove(self, uid: uuid.UUID):
""" """