feat: 修改脚本配置API
This commit is contained in:
@@ -1,30 +0,0 @@
|
|||||||
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
|
|
||||||
# Copyright © 2024-2025 DLmaster361
|
|
||||||
|
|
||||||
# This file is part of AUTO_MAA.
|
|
||||||
|
|
||||||
# AUTO_MAA is free software: you can redistribute it and/or modify
|
|
||||||
# it under the terms of the GNU General Public License as published
|
|
||||||
# by the Free Software Foundation, either version 3 of the License,
|
|
||||||
# or (at your option) any later version.
|
|
||||||
|
|
||||||
# AUTO_MAA is distributed in the hope that it will be useful,
|
|
||||||
# but WITHOUT ANY WARRANTY; without even the implied warranty
|
|
||||||
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See
|
|
||||||
# the GNU General Public License for more details.
|
|
||||||
|
|
||||||
# You should have received a copy of the GNU General Public License
|
|
||||||
# along with AUTO_MAA. If not, see <https://www.gnu.org/licenses/>.
|
|
||||||
|
|
||||||
# Contact: DLmaster_361@163.com
|
|
||||||
|
|
||||||
|
|
||||||
__version__ = "5.0.0"
|
|
||||||
__author__ = "DLmaster361 <DLmaster_361@163.com>"
|
|
||||||
__license__ = "GPL-3.0 license"
|
|
||||||
|
|
||||||
from .api import app
|
|
||||||
from .core import Config
|
|
||||||
from .utils import get_logger
|
|
||||||
|
|
||||||
__all__ = ["app", "Config", "get_logger"]
|
|
||||||
@@ -22,6 +22,6 @@ __version__ = "5.0.0"
|
|||||||
__author__ = "DLmaster361 <DLmaster_361@163.com>"
|
__author__ = "DLmaster361 <DLmaster_361@163.com>"
|
||||||
__license__ = "GPL-3.0 license"
|
__license__ = "GPL-3.0 license"
|
||||||
|
|
||||||
from .api import app
|
from .scripts import router as scripts_router
|
||||||
|
|
||||||
__all__ = ["app"]
|
__all__ = ["scripts_router"]
|
||||||
|
|||||||
472
app/api/api.py
472
app/api/api.py
@@ -10,21 +10,6 @@ from fastapi.middleware.cors import CORSMiddleware
|
|||||||
from app.core import Config
|
from app.core import Config
|
||||||
from app.utils import get_logger
|
from app.utils import get_logger
|
||||||
|
|
||||||
logger = get_logger("API 模块")
|
|
||||||
|
|
||||||
app = FastAPI(
|
|
||||||
title="AUTO_MAA",
|
|
||||||
description="API for managing automation scripts, plans, and tasks",
|
|
||||||
version="1.0.0",
|
|
||||||
)
|
|
||||||
|
|
||||||
app.add_middleware(
|
|
||||||
CORSMiddleware,
|
|
||||||
allow_origins=["*"], # 允许所有域名跨域访问
|
|
||||||
allow_credentials=True,
|
|
||||||
allow_methods=["*"], # 允许所有请求方法,如 GET、POST、PUT、DELETE
|
|
||||||
allow_headers=["*"], # 允许所有请求头
|
|
||||||
)
|
|
||||||
|
|
||||||
# 此文件由ai生成 返回值非最终版本
|
# 此文件由ai生成 返回值非最终版本
|
||||||
|
|
||||||
@@ -37,43 +22,6 @@ app.add_middleware(
|
|||||||
# Script Models
|
# Script Models
|
||||||
|
|
||||||
|
|
||||||
class BaseOut(BaseModel):
|
|
||||||
code: int = Field(default=200, description="状态码")
|
|
||||||
status: str = Field(default="success", description="操作状态")
|
|
||||||
message: str = Field(default="操作成功", description="操作消息")
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptCreateIn(BaseModel):
|
|
||||||
type: Literal["MAA", "General"]
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptCreateOut(BaseOut):
|
|
||||||
scriptId: str = Field(..., description="新创建的脚本ID")
|
|
||||||
data: Dict[str, Any] = Field(..., description="脚本配置数据")
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptGetIn(BaseModel):
|
|
||||||
scriptId: Optional[str] = Field(None, description="脚本ID,仅在模式为Single时需要")
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptGetOut(BaseOut):
|
|
||||||
index: List[Dict[str, str]] = Field(..., description="脚本索引列表")
|
|
||||||
data: Dict[str, Any] = Field(..., description="脚本列表或单个脚本数据")
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptUpdateIn(BaseModel):
|
|
||||||
scriptId: str = Field(..., description="脚本ID")
|
|
||||||
data: Dict[str, Any] = Field(..., description="脚本更新数据")
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptUpdateOut(BaseOut):
|
|
||||||
message: str = Field(default="脚本更新成功", description="操作消息")
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptDeleteIn(BaseModel):
|
|
||||||
scriptId: str = Field(..., description="脚本ID")
|
|
||||||
|
|
||||||
|
|
||||||
class ScriptUser(BaseModel):
|
class ScriptUser(BaseModel):
|
||||||
userId: str
|
userId: str
|
||||||
config: Dict[str, Any] = {}
|
config: Dict[str, Any] = {}
|
||||||
@@ -161,289 +109,235 @@ class SettingsUpdate(BaseModel):
|
|||||||
# return {"status": "success", "data": {}}
|
# return {"status": "success", "data": {}}
|
||||||
|
|
||||||
|
|
||||||
@app.post(
|
# @app.post("/api/scripts/{scriptId}/users", summary="为脚本添加用户")
|
||||||
"/api/add/scripts",
|
# async def add_script_user(
|
||||||
summary="添加脚本",
|
# scriptId: str = Path(..., description="脚本ID"), user: ScriptUser = Body(...)
|
||||||
response_model=ScriptCreateOut,
|
|
||||||
status_code=200,
|
|
||||||
)
|
|
||||||
async def add_script(script: ScriptCreateIn = Body(...)) -> ScriptCreateOut:
|
|
||||||
"""添加脚本"""
|
|
||||||
|
|
||||||
uid, config = await Config.add_script(script.type)
|
|
||||||
return ScriptCreateOut(scriptId=str(uid), data=await config.toDict())
|
|
||||||
|
|
||||||
|
|
||||||
@app.post(
|
|
||||||
"/api/get/scripts", summary="查询脚本", response_model=ScriptGetOut, status_code=200
|
|
||||||
)
|
|
||||||
async def get_scripts(script: ScriptGetIn = Body(...)) -> ScriptGetOut:
|
|
||||||
"""查询脚本"""
|
|
||||||
try:
|
|
||||||
index, data = await Config.get_script(script.scriptId)
|
|
||||||
except Exception as e:
|
|
||||||
return ScriptGetOut(code=500, status="error", message=str(e), index=[], data={})
|
|
||||||
return ScriptGetOut(index=index, data=data)
|
|
||||||
|
|
||||||
|
|
||||||
# @app.post("/api/update/scripts/{scriptId}", summary="更新脚本")
|
|
||||||
# async def update_script(
|
|
||||||
# scriptId: str = Path(..., description="脚本ID"),
|
|
||||||
# update_data: ScriptUpdate = Body(...),
|
|
||||||
# ):
|
# ):
|
||||||
# """
|
# """
|
||||||
# 更新脚本
|
# 为脚本添加用户
|
||||||
# """
|
# """
|
||||||
# # 实现更新脚本的逻辑
|
# # 实现为脚本添加用户的逻辑
|
||||||
# return {"status": "success"}
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post(
|
# @app.get("/api/scripts/{scriptId}/users", summary="查询脚本的所有下属用户")
|
||||||
"/api/delete/scripts",
|
# async def get_script_users(scriptId: str = Path(..., description="脚本ID")):
|
||||||
summary="删除脚本",
|
# """
|
||||||
response_model=BaseOut,
|
# 查询脚本的所有下属用户
|
||||||
status_code=200,
|
# """
|
||||||
)
|
# # 实现查询脚本的所有下属用户的逻辑
|
||||||
async def delete_script(
|
# return {"status": "success", "data": []}
|
||||||
script: ScriptDeleteIn = Body(..., description="脚本ID")
|
|
||||||
) -> BaseOut:
|
|
||||||
"""删除脚本"""
|
|
||||||
try:
|
|
||||||
await Config.del_script(script.scriptId)
|
|
||||||
except Exception as e:
|
|
||||||
return BaseOut(code=500, status="error", message=str(e))
|
|
||||||
return BaseOut()
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/scripts/{scriptId}/users", summary="为脚本添加用户")
|
# @app.get("/api/scripts/{scriptId}/users/{userId}", summary="查询脚本下的单个下属用户")
|
||||||
async def add_script_user(
|
# async def get_script_user(
|
||||||
scriptId: str = Path(..., description="脚本ID"), user: ScriptUser = Body(...)
|
# scriptId: str = Path(..., description="脚本ID"),
|
||||||
):
|
# userId: str = Path(..., description="用户ID"),
|
||||||
"""
|
# ):
|
||||||
为脚本添加用户
|
# """
|
||||||
"""
|
# 查询脚本下的单个下属用户
|
||||||
# 实现为脚本添加用户的逻辑
|
# """
|
||||||
return {"status": "success"}
|
# # 实现查询脚本下的单个下属用户的逻辑
|
||||||
|
# return {"status": "success", "data": {}}
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/scripts/{scriptId}/users", summary="查询脚本的所有下属用户")
|
# @app.put("/api/scripts/{scriptId}/users/{userId}", summary="更新脚本下属用户的关联信息")
|
||||||
async def get_script_users(scriptId: str = Path(..., description="脚本ID")):
|
# async def update_script_user(
|
||||||
"""
|
# scriptId: str = Path(..., description="脚本ID"),
|
||||||
查询脚本的所有下属用户
|
# userId: str = Path(..., description="用户ID"),
|
||||||
"""
|
# config: Dict[str, Any] = Body(...),
|
||||||
# 实现查询脚本的所有下属用户的逻辑
|
# ):
|
||||||
return {"status": "success", "data": []}
|
# """
|
||||||
|
# 更新脚本下属用户的关联信息
|
||||||
|
# """
|
||||||
|
# # 实现更新脚本下属用户的关联信息的逻辑
|
||||||
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.get("/api/scripts/{scriptId}/users/{userId}", summary="查询脚本下的单个下属用户")
|
# @app.delete("/api/scripts/{scriptId}/users/{userId}", summary="从脚本移除用户")
|
||||||
async def get_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"),
|
||||||
):
|
# ):
|
||||||
"""
|
# """
|
||||||
查询脚本下的单个下属用户
|
# 从脚本移除用户
|
||||||
"""
|
# """
|
||||||
# 实现查询脚本下的单个下属用户的逻辑
|
# # 实现从脚本移除用户的逻辑
|
||||||
return {"status": "success", "data": {}}
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.put("/api/scripts/{scriptId}/users/{userId}", summary="更新脚本下属用户的关联信息")
|
# @app.post("/api/add/plans", summary="创建计划")
|
||||||
async def update_script_user(
|
# async def add_plan(plan: PlanCreate = Body(...)):
|
||||||
scriptId: str = Path(..., description="脚本ID"),
|
# """
|
||||||
userId: str = Path(..., description="用户ID"),
|
# 创建计划
|
||||||
config: Dict[str, Any] = Body(...),
|
# {
|
||||||
):
|
# "name": "计划 1",
|
||||||
"""
|
# "mode": "全局", // 或 "周计划"
|
||||||
更新脚本下属用户的关联信息
|
# "details": {
|
||||||
"""
|
# "周一": {
|
||||||
# 实现更新脚本下属用户的关联信息的逻辑
|
# "吃理智药": 0,
|
||||||
return {"status": "success"}
|
# "连战次数": "AUTO",
|
||||||
|
# "关卡选择": "当前/上次",
|
||||||
|
# "备选-1": "当前/上次",
|
||||||
|
# "备选-2": "当前/上次",
|
||||||
|
# "备选-3": "当前/上次",
|
||||||
|
# "剩余理智": "不使用"
|
||||||
|
# },
|
||||||
|
# // 其他天数...
|
||||||
|
# }
|
||||||
|
# }
|
||||||
|
# """
|
||||||
|
# # 实现创建计划的逻辑
|
||||||
|
# return {"status": "success", "planId": "new_plan_id"}
|
||||||
|
|
||||||
|
|
||||||
@app.delete("/api/scripts/{scriptId}/users/{userId}", summary="从脚本移除用户")
|
# @app.post("/api/get/plans", summary="查询所有计划")
|
||||||
async def remove_script_user(
|
# async def get_plans():
|
||||||
scriptId: str = Path(..., description="脚本ID"),
|
# """
|
||||||
userId: str = Path(..., description="用户ID"),
|
# 查询所有计划
|
||||||
):
|
# """
|
||||||
"""
|
# # 实现查询所有计划的逻辑
|
||||||
从脚本移除用户
|
# return {"status": "success", "data": []}
|
||||||
"""
|
|
||||||
# 实现从脚本移除用户的逻辑
|
|
||||||
return {"status": "success"}
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/add/plans", summary="创建计划")
|
# @app.post("/api/get/plans/{planId}", summary="查询单个计划")
|
||||||
async def add_plan(plan: PlanCreate = Body(...)):
|
# async def get_plan(planId: str = Path(..., description="计划ID")):
|
||||||
"""
|
# """
|
||||||
创建计划
|
# 查询单个计划
|
||||||
{
|
# """
|
||||||
"name": "计划 1",
|
# # 实现查询单个计划的逻辑
|
||||||
"mode": "全局", // 或 "周计划"
|
# return {"status": "success", "data": {}}
|
||||||
"details": {
|
|
||||||
"周一": {
|
|
||||||
"吃理智药": 0,
|
|
||||||
"连战次数": "AUTO",
|
|
||||||
"关卡选择": "当前/上次",
|
|
||||||
"备选-1": "当前/上次",
|
|
||||||
"备选-2": "当前/上次",
|
|
||||||
"备选-3": "当前/上次",
|
|
||||||
"剩余理智": "不使用"
|
|
||||||
},
|
|
||||||
// 其他天数...
|
|
||||||
}
|
|
||||||
}
|
|
||||||
"""
|
|
||||||
# 实现创建计划的逻辑
|
|
||||||
return {"status": "success", "planId": "new_plan_id"}
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/get/plans", summary="查询所有计划")
|
# @app.post("/api/update/plans/{planId}", summary="更新计划")
|
||||||
async def get_plans():
|
# async def update_plan(
|
||||||
"""
|
# planId: str = Path(..., description="计划ID"), update_data: PlanUpdate = Body(...)
|
||||||
查询所有计划
|
# ):
|
||||||
"""
|
# """
|
||||||
# 实现查询所有计划的逻辑
|
# 更新计划
|
||||||
return {"status": "success", "data": []}
|
# """
|
||||||
|
# # 实现更新计划的逻辑
|
||||||
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/get/plans/{planId}", summary="查询单个计划")
|
# @app.post("/api/delete/plans/{planId}", summary="删除计划")
|
||||||
async def get_plan(planId: str = Path(..., description="计划ID")):
|
# async def delete_plan(planId: str = Path(..., description="计划ID")):
|
||||||
"""
|
# """
|
||||||
查询单个计划
|
# 删除计划
|
||||||
"""
|
# """
|
||||||
# 实现查询单个计划的逻辑
|
# # 实现删除计划的逻辑
|
||||||
return {"status": "success", "data": {}}
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/update/plans/{planId}", summary="更新计划")
|
# @app.post("/api/update/plans/{planId}/mode", summary="切换计划模式")
|
||||||
async def update_plan(
|
# async def update_plan_mode(
|
||||||
planId: str = Path(..., description="计划ID"), update_data: PlanUpdate = Body(...)
|
# planId: str = Path(..., description="计划ID"), mode_data: PlanModeUpdate = Body(...)
|
||||||
):
|
# ):
|
||||||
"""
|
# """
|
||||||
更新计划
|
# 切换计划模式
|
||||||
"""
|
# {
|
||||||
# 实现更新计划的逻辑
|
# "mode": "周计划"
|
||||||
return {"status": "success"}
|
# }
|
||||||
|
# """
|
||||||
|
# # 实现切换计划模式的逻辑
|
||||||
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/delete/plans/{planId}", summary="删除计划")
|
# @app.post("/api/add/queues", summary="创建调度队列")
|
||||||
async def delete_plan(planId: str = Path(..., description="计划ID")):
|
# async def add_queue(queue: QueueCreate = Body(...)):
|
||||||
"""
|
# """
|
||||||
删除计划
|
# 创建调度队列
|
||||||
"""
|
# """
|
||||||
# 实现删除计划的逻辑
|
# # 实现创建调度队列的逻辑
|
||||||
return {"status": "success"}
|
# return {"status": "success", "queueId": "new_queue_id"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/update/plans/{planId}/mode", summary="切换计划模式")
|
# @app.post("/api/get/queues", summary="查询所有调度队列")
|
||||||
async def update_plan_mode(
|
# async def get_queues():
|
||||||
planId: str = Path(..., description="计划ID"), mode_data: PlanModeUpdate = Body(...)
|
# """
|
||||||
):
|
# 查询所有调度队列
|
||||||
"""
|
# """
|
||||||
切换计划模式
|
# # 实现查询所有调度队列的逻辑
|
||||||
{
|
# return {"status": "success", "data": []}
|
||||||
"mode": "周计划"
|
|
||||||
}
|
|
||||||
"""
|
|
||||||
# 实现切换计划模式的逻辑
|
|
||||||
return {"status": "success"}
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/add/queues", summary="创建调度队列")
|
# @app.post("/api/get/queues/{queueId}", summary="查询单个调度队列详情")
|
||||||
async def add_queue(queue: QueueCreate = Body(...)):
|
# async def get_queue(queueId: str = Path(..., description="调度队列ID")):
|
||||||
"""
|
# """
|
||||||
创建调度队列
|
# 查询单个调度队列详情
|
||||||
"""
|
# """
|
||||||
# 实现创建调度队列的逻辑
|
# # 实现查询单个调度队列详情的逻辑
|
||||||
return {"status": "success", "queueId": "new_queue_id"}
|
# return {"status": "success", "data": {}}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/get/queues", summary="查询所有调度队列")
|
# @app.post("/api/update/queues/{queueId}", summary="更新调度队列")
|
||||||
async def get_queues():
|
# async def update_queue(
|
||||||
"""
|
# queueId: str = Path(..., description="调度队列ID"),
|
||||||
查询所有调度队列
|
# update_data: QueueUpdate = Body(...),
|
||||||
"""
|
# ):
|
||||||
# 实现查询所有调度队列的逻辑
|
# """
|
||||||
return {"status": "success", "data": []}
|
# 更新调度队列
|
||||||
|
# """
|
||||||
|
# # 实现更新调度队列的逻辑
|
||||||
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/get/queues/{queueId}", summary="查询单个调度队列详情")
|
# @app.post("/api/delete/queues/{queueId}", summary="删除调度队列")
|
||||||
async def get_queue(queueId: str = Path(..., description="调度队列ID")):
|
# async def delete_queue(queueId: str = Path(..., description="调度队列ID")):
|
||||||
"""
|
# """
|
||||||
查询单个调度队列详情
|
# 删除调度队列
|
||||||
"""
|
# """
|
||||||
# 实现查询单个调度队列详情的逻辑
|
# # 实现删除调度队列的逻辑
|
||||||
return {"status": "success", "data": {}}
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/update/queues/{queueId}", summary="更新调度队列")
|
# @app.post("/api/add/tasks", summary="添加任务")
|
||||||
async def update_queue(
|
# async def add_task(task: TaskCreate = Body(...)):
|
||||||
queueId: str = Path(..., description="调度队列ID"),
|
# """
|
||||||
update_data: QueueUpdate = Body(...),
|
# 添加任务
|
||||||
):
|
# """
|
||||||
"""
|
# # 实现添加任务的逻辑
|
||||||
更新调度队列
|
# return {"status": "success", "taskId": "new_task_id"}
|
||||||
"""
|
|
||||||
# 实现更新调度队列的逻辑
|
|
||||||
return {"status": "success"}
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/delete/queues/{queueId}", summary="删除调度队列")
|
# @app.post("/api/tasks/{taskId}/start", summary="开始任务")
|
||||||
async def delete_queue(queueId: str = Path(..., description="调度队列ID")):
|
# async def start_task(taskId: str = Path(..., description="任务ID")):
|
||||||
"""
|
# """
|
||||||
删除调度队列
|
# 开始任务
|
||||||
"""
|
# """
|
||||||
# 实现删除调度队列的逻辑
|
# # 实现开始任务的逻辑
|
||||||
return {"status": "success"}
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/add/tasks", summary="添加任务")
|
# @app.post("/api/get/history", summary="查询历史记录")
|
||||||
async def add_task(task: TaskCreate = Body(...)):
|
# async def get_history():
|
||||||
"""
|
# """
|
||||||
添加任务
|
# 查询历史记录
|
||||||
"""
|
# """
|
||||||
# 实现添加任务的逻辑
|
# # 实现查询历史记录的逻辑
|
||||||
return {"status": "success", "taskId": "new_task_id"}
|
# return {"status": "success", "data": []}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/tasks/{taskId}/start", summary="开始任务")
|
# @app.post("/api/update/settings", summary="更新部分设置")
|
||||||
async def start_task(taskId: str = Path(..., description="任务ID")):
|
# async def update_settings(settings: SettingsUpdate = Body(...)):
|
||||||
"""
|
# """
|
||||||
开始任务
|
# 更新部分设置
|
||||||
"""
|
# """
|
||||||
# 实现开始任务的逻辑
|
# # 实现更新部分设置的逻辑
|
||||||
return {"status": "success"}
|
# return {"status": "success"}
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/get/history", summary="查询历史记录")
|
# # ======================
|
||||||
async def get_history():
|
# # Error Handlers
|
||||||
"""
|
# # ======================
|
||||||
查询历史记录
|
|
||||||
"""
|
|
||||||
# 实现查询历史记录的逻辑
|
|
||||||
return {"status": "success", "data": []}
|
|
||||||
|
|
||||||
|
|
||||||
@app.post("/api/update/settings", summary="更新部分设置")
|
# @app.exception_handler(HTTPException)
|
||||||
async def update_settings(settings: SettingsUpdate = Body(...)):
|
# async def http_exception_handler(request, exc):
|
||||||
"""
|
# return {"status": "error", "code": exc.status_code, "message": exc.detail}
|
||||||
更新部分设置
|
|
||||||
"""
|
|
||||||
# 实现更新部分设置的逻辑
|
|
||||||
return {"status": "success"}
|
|
||||||
|
|
||||||
|
|
||||||
# ======================
|
# if __name__ == "__main__":
|
||||||
# Error Handlers
|
# import uvicorn
|
||||||
# ======================
|
|
||||||
|
|
||||||
|
# uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||||
@app.exception_handler(HTTPException)
|
|
||||||
async def http_exception_handler(request, exc):
|
|
||||||
return {"status": "error", "code": exc.status_code, "message": exc.detail}
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
import uvicorn
|
|
||||||
|
|
||||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
|
||||||
|
|||||||
70
app/api/scripts.py
Normal file
70
app/api/scripts.py
Normal file
@@ -0,0 +1,70 @@
|
|||||||
|
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
|
||||||
|
# Copyright © 2024-2025 DLmaster361
|
||||||
|
|
||||||
|
# This file is part of AUTO_MAA.
|
||||||
|
|
||||||
|
# AUTO_MAA is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU General Public License as published
|
||||||
|
# by the Free Software Foundation, either version 3 of the License,
|
||||||
|
# or (at your option) any later version.
|
||||||
|
|
||||||
|
# AUTO_MAA is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty
|
||||||
|
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See
|
||||||
|
# the GNU General Public License for more details.
|
||||||
|
|
||||||
|
# You should have received a copy of the GNU General Public License
|
||||||
|
# along with AUTO_MAA. If not, see <https://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
# Contact: DLmaster_361@163.com
|
||||||
|
|
||||||
|
|
||||||
|
from fastapi import APIRouter, Body
|
||||||
|
|
||||||
|
from core import Config
|
||||||
|
from models.schema import *
|
||||||
|
|
||||||
|
router = APIRouter(prefix="/api/scripts", tags=["脚本管理"])
|
||||||
|
|
||||||
|
|
||||||
|
@router.post(
|
||||||
|
"/add", summary="添加脚本", response_model=ScriptCreateOut, status_code=200
|
||||||
|
)
|
||||||
|
async def add_script(script: ScriptCreateIn = Body(...)) -> ScriptCreateOut:
|
||||||
|
"""添加脚本"""
|
||||||
|
|
||||||
|
uid, config = await Config.add_script(script.type)
|
||||||
|
return ScriptCreateOut(scriptId=str(uid), data=await config.toDict())
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/get", summary="查询脚本", response_model=ScriptGetOut, status_code=200)
|
||||||
|
async def get_scripts(script: ScriptGetIn = Body(...)) -> ScriptGetOut:
|
||||||
|
"""查询脚本"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
index, data = await Config.get_script(script.scriptId)
|
||||||
|
except Exception as e:
|
||||||
|
return ScriptGetOut(code=500, status="error", message=str(e), index=[], data={})
|
||||||
|
return ScriptGetOut(index=index, data=data)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/update", summary="更新脚本", response_model=BaseOut, status_code=200)
|
||||||
|
async def update_script(script: ScriptUpdateIn = Body(...)) -> BaseOut:
|
||||||
|
"""更新脚本"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
await Config.update_script(script.scriptId, script.data)
|
||||||
|
except Exception as e:
|
||||||
|
return BaseOut(code=500, status="error", message=str(e))
|
||||||
|
return BaseOut()
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/delete", summary="删除脚本", response_model=BaseOut, status_code=200)
|
||||||
|
async def delete_script(script: ScriptDeleteIn = Body(...)) -> BaseOut:
|
||||||
|
"""删除脚本"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
await Config.del_script(script.scriptId)
|
||||||
|
except Exception as e:
|
||||||
|
return BaseOut(code=500, status="error", message=str(e))
|
||||||
|
return BaseOut()
|
||||||
@@ -33,8 +33,8 @@ from pathlib import Path
|
|||||||
|
|
||||||
from typing import Union, Dict, List, Literal, Optional, Any, Tuple, Callable, TypeVar
|
from typing import Union, Dict, List, Literal, Optional, Any, Tuple, Callable, TypeVar
|
||||||
|
|
||||||
from app.utils import get_logger
|
from utils import get_logger
|
||||||
from app.models.ConfigBase import *
|
from models.ConfigBase import *
|
||||||
|
|
||||||
|
|
||||||
class GlobalConfig(ConfigBase):
|
class GlobalConfig(ConfigBase):
|
||||||
@@ -628,6 +628,8 @@ class AppConfig(GlobalConfig):
|
|||||||
) -> tuple[uuid.UUID, ConfigBase]:
|
) -> tuple[uuid.UUID, ConfigBase]:
|
||||||
"""添加脚本配置"""
|
"""添加脚本配置"""
|
||||||
|
|
||||||
|
self.logger.info(f"添加脚本配置:{script}")
|
||||||
|
|
||||||
class_book = {"MAA": MaaConfig, "General": GeneralConfig}
|
class_book = {"MAA": MaaConfig, "General": GeneralConfig}
|
||||||
|
|
||||||
return await self.ScriptConfig.add(class_book[script])
|
return await self.ScriptConfig.add(class_book[script])
|
||||||
@@ -635,6 +637,8 @@ class AppConfig(GlobalConfig):
|
|||||||
async def get_script(self, script_id: Optional[str]) -> tuple[list, dict]:
|
async def get_script(self, script_id: Optional[str]) -> tuple[list, dict]:
|
||||||
"""获取脚本配置"""
|
"""获取脚本配置"""
|
||||||
|
|
||||||
|
self.logger.info(f"获取脚本配置:{script_id}")
|
||||||
|
|
||||||
if script_id is None:
|
if script_id is None:
|
||||||
data = await self.ScriptConfig.toDict()
|
data = await self.ScriptConfig.toDict()
|
||||||
else:
|
else:
|
||||||
@@ -644,6 +648,24 @@ class AppConfig(GlobalConfig):
|
|||||||
|
|
||||||
return list(index), data
|
return list(index), data
|
||||||
|
|
||||||
|
async def update_script(
|
||||||
|
self, script_id: str, data: Dict[str, Dict[str, Any]]
|
||||||
|
) -> None:
|
||||||
|
"""更新脚本配置"""
|
||||||
|
|
||||||
|
self.logger.info(f"更新脚本配置:{script_id}")
|
||||||
|
|
||||||
|
uid = uuid.UUID(script_id)
|
||||||
|
|
||||||
|
for group, items in data.items():
|
||||||
|
for name, value in items.items():
|
||||||
|
self.logger.debug(
|
||||||
|
f"更新脚本配置:{script_id} - {group}.{name} = {value}"
|
||||||
|
)
|
||||||
|
await self.ScriptConfig[uid].set(group, name, value)
|
||||||
|
|
||||||
|
await self.ScriptConfig.save()
|
||||||
|
|
||||||
async def del_script(self, script_id: str) -> None:
|
async def del_script(self, script_id: str) -> None:
|
||||||
"""删除脚本配置"""
|
"""删除脚本配置"""
|
||||||
|
|
||||||
|
|||||||
@@ -22,12 +22,37 @@
|
|||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
import ctypes
|
import ctypes
|
||||||
|
import uvicorn
|
||||||
|
from fastapi import FastAPI
|
||||||
|
from fastapi.middleware.cors import CORSMiddleware
|
||||||
|
|
||||||
|
|
||||||
|
from api import scripts_router
|
||||||
|
from core import Config
|
||||||
|
from utils import get_logger
|
||||||
|
|
||||||
from app.utils import get_logger
|
|
||||||
|
|
||||||
logger = get_logger("主程序")
|
logger = get_logger("主程序")
|
||||||
|
|
||||||
|
|
||||||
|
app = FastAPI(
|
||||||
|
title="AUTO_MAA",
|
||||||
|
description="API for managing automation scripts, plans, and tasks",
|
||||||
|
version="1.0.0",
|
||||||
|
)
|
||||||
|
|
||||||
|
app.add_middleware(
|
||||||
|
CORSMiddleware,
|
||||||
|
allow_origins=["*"], # 允许所有域名跨域访问
|
||||||
|
allow_credentials=True,
|
||||||
|
allow_methods=["*"], # 允许所有请求方法,如 GET、POST、PUT、DELETE
|
||||||
|
allow_headers=["*"], # 允许所有请求头
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
app.include_router(scripts_router)
|
||||||
|
|
||||||
|
|
||||||
def is_admin() -> bool:
|
def is_admin() -> bool:
|
||||||
"""检查当前程序是否以管理员身份运行"""
|
"""检查当前程序是否以管理员身份运行"""
|
||||||
try:
|
try:
|
||||||
@@ -41,9 +66,6 @@ def main():
|
|||||||
|
|
||||||
if is_admin():
|
if is_admin():
|
||||||
|
|
||||||
import uvicorn
|
|
||||||
from app.api import app
|
|
||||||
|
|
||||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||||
|
|
||||||
else:
|
else:
|
||||||
@@ -23,5 +23,6 @@ __author__ = "DLmaster361 <DLmaster_361@163.com>"
|
|||||||
__license__ = "GPL-3.0 license"
|
__license__ = "GPL-3.0 license"
|
||||||
|
|
||||||
from .ConfigBase import *
|
from .ConfigBase import *
|
||||||
|
from .schema import *
|
||||||
|
|
||||||
__all__ = ["ConfigBase"]
|
__all__ = ["ConfigBase", "schema"]
|
||||||
|
|||||||
56
app/models/schema.py
Normal file
56
app/models/schema.py
Normal file
@@ -0,0 +1,56 @@
|
|||||||
|
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
|
||||||
|
# Copyright © 2024-2025 DLmaster361
|
||||||
|
|
||||||
|
# This file is part of AUTO_MAA.
|
||||||
|
|
||||||
|
# AUTO_MAA is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU General Public License as published
|
||||||
|
# by the Free Software Foundation, either version 3 of the License,
|
||||||
|
# or (at your option) any later version.
|
||||||
|
|
||||||
|
# AUTO_MAA is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty
|
||||||
|
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See
|
||||||
|
# the GNU General Public License for more details.
|
||||||
|
|
||||||
|
# You should have received a copy of the GNU General Public License
|
||||||
|
# along with AUTO_MAA. If not, see <https://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
# Contact: DLmaster_361@163.com
|
||||||
|
|
||||||
|
|
||||||
|
from pydantic import BaseModel, Field
|
||||||
|
from typing import Dict, Any, List, Optional, Literal
|
||||||
|
|
||||||
|
|
||||||
|
class BaseOut(BaseModel):
|
||||||
|
code: int = Field(default=200, description="状态码")
|
||||||
|
status: str = Field(default="success", description="操作状态")
|
||||||
|
message: str = Field(default="操作成功", description="操作消息")
|
||||||
|
|
||||||
|
|
||||||
|
class ScriptCreateIn(BaseModel):
|
||||||
|
type: Literal["MAA", "General"]
|
||||||
|
|
||||||
|
|
||||||
|
class ScriptCreateOut(BaseOut):
|
||||||
|
scriptId: str = Field(..., description="新创建的脚本ID")
|
||||||
|
data: Dict[str, Any] = Field(..., description="脚本配置数据")
|
||||||
|
|
||||||
|
|
||||||
|
class ScriptGetIn(BaseModel):
|
||||||
|
scriptId: Optional[str] = Field(None, description="脚本ID,仅在模式为Single时需要")
|
||||||
|
|
||||||
|
|
||||||
|
class ScriptGetOut(BaseOut):
|
||||||
|
index: List[Dict[str, str]] = Field(..., description="脚本索引列表")
|
||||||
|
data: Dict[str, Any] = Field(..., description="脚本列表或单个脚本数据")
|
||||||
|
|
||||||
|
|
||||||
|
class ScriptUpdateIn(BaseModel):
|
||||||
|
scriptId: str = Field(..., description="脚本ID")
|
||||||
|
data: Dict[str, Dict[str, Any]] = Field(..., description="脚本更新数据")
|
||||||
|
|
||||||
|
|
||||||
|
class ScriptDeleteIn(BaseModel):
|
||||||
|
scriptId: str = Field(..., description="脚本ID")
|
||||||
Reference in New Issue
Block a user