feat: 添加信息获取API与主业务定时器

This commit is contained in:
DLmaster361
2025-08-05 16:09:48 +08:00
parent 4a9c9ab1f3
commit d61b90baa4
15 changed files with 375 additions and 31 deletions

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.
@@ -22,9 +23,16 @@ __version__ = "5.0.0"
__author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from .info import router as info_router
from .scripts import router as scripts_router
from .plan import router as plan_router
from .queue import router as queue_router
from .setting import router as setting_router
__all__ = ["scripts_router", "plan_router", "queue_router", "setting_router"]
__all__ = [
"info_router",
"scripts_router",
"plan_router",
"queue_router",
"setting_router",
]

66
app/api/info.py Normal file
View File

@@ -0,0 +1,66 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# 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/info", tags=["信息获取"])
@router.post(
"/stage", summary="获取关卡号信息", response_model=InfoOut, status_code=200
)
async def get_stage_info() -> InfoOut:
try:
if_get_maa_stage, data = await Config.get_stage()
except Exception as e:
return InfoOut(code=500, status="error", message=str(e), data={})
return InfoOut(
status="success" if if_get_maa_stage else "warning",
message="获取关卡号信息成功" if if_get_maa_stage else "未能获取活动关卡号信息",
data=data,
)
@router.post("/notice", summary="获取通知信息", response_model=InfoOut, status_code=200)
async def get_notice_info() -> InfoOut:
try:
data = await Config.get_server_info("notice")
except Exception as e:
return InfoOut(code=500, status="error", message=str(e), data={})
return InfoOut(data=data)
@router.post(
"/apps_info", summary="获取可下载应用信息", response_model=InfoOut, status_code=200
)
async def get_apps_info() -> InfoOut:
try:
data = await Config.get_server_info("apps_info")
except Exception as e:
return InfoOut(code=500, status="error", message=str(e), data={})
return InfoOut(data=data)

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.
@@ -23,5 +24,6 @@ __author__ = "DLmaster361 <DLmaster_361@163.com>"
__license__ = "GPL-3.0 license"
from .config import Config
from .timer import MainTimer
__all__ = ["Config"]
__all__ = ["Config", "MainTimer"]

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.
@@ -19,13 +20,14 @@
# Contact: DLmaster_361@163.com
import argparse
import sqlite3
import json
import sys
import shutil
import re
import base64
import requests
import truststore
import calendar
from datetime import datetime, timedelta, date
from pathlib import Path
@@ -548,34 +550,50 @@ class AppConfig(GlobalConfig):
VERSION = "4.5.0.1"
CLASS_BOOK = {
"Maa": MaaConfig,
"MaaPlan": MaaPlanConfig,
"General": GeneralConfig,
}
STAGE_DAILY_INFO = [
{"value": "-", "text": "当前/上次", "days": [1, 2, 3, 4, 5, 6, 7]},
{"value": "1-7", "text": "1-7", "days": [1, 2, 3, 4, 5, 6, 7]},
{"value": "R8-11", "text": "R8-11", "days": [1, 2, 3, 4, 5, 6, 7]},
{
"value": "12-17-HARD",
"text": "12-17-HARD",
"days": [1, 2, 3, 4, 5, 6, 7],
},
{"value": "CE-6", "text": "龙门币-6/5", "days": [2, 4, 6, 7]},
{"value": "AP-5", "text": "红票-5", "days": [1, 4, 6, 7]},
{"value": "CA-5", "text": "技能-5", "days": [2, 3, 5, 7]},
{"value": "LS-6", "text": "经验-6/5", "days": [1, 2, 3, 4, 5, 6, 7]},
{"value": "SK-5", "text": "碳-5", "days": [1, 3, 5, 6]},
{"value": "PR-A-1", "text": "奶/盾芯片", "days": [1, 4, 5, 7]},
{"value": "PR-A-2", "text": "奶/盾芯片组", "days": [1, 4, 5, 7]},
{"value": "PR-B-1", "text": "术/狙芯片", "days": [1, 2, 5, 6]},
{"value": "PR-B-2", "text": "术/狙芯片组", "days": [1, 2, 5, 6]},
{"value": "PR-C-1", "text": "先/辅芯片", "days": [3, 4, 6, 7]},
{"value": "PR-C-2", "text": "先/辅芯片组", "days": [3, 4, 6, 7]},
{"value": "PR-D-1", "text": "近/特芯片", "days": [2, 3, 6, 7]},
{"value": "PR-D-2", "text": "近/特芯片组", "days": [2, 3, 6, 7]},
]
def __init__(self) -> None:
super().__init__(if_save_multi_config=False)
self.root_path = Path.cwd()
self.log_path = self.root_path / "debug/app.log"
# self.database_path = self.root_path / "data/data.db"
self.database_path = self.root_path / "data/data.db"
self.config_path = self.root_path / "config"
# self.key_path = self.root_path / "data/key"
self.key_path = self.root_path / "data/key"
# self.main_window = None
# self.PASSWORD = ""
# self.running_list = []
# self.silence_dict: Dict[Path, datetime] = {}
# self.info_bar_list = []
# self.stage_dict = {
# "ALL": {"value": [], "text": []},
# "Monday": {"value": [], "text": []},
# "Tuesday": {"value": [], "text": []},
# "Wednesday": {"value": [], "text": []},
# "Thursday": {"value": [], "text": []},
# "Friday": {"value": [], "text": []},
# "Saturday": {"value": [], "text": []},
# "Sunday": {"value": [], "text": []},
# }
# self.power_sign = "NoAction"
# self.if_ignore_silence = False
# self.if_database_opened = False
self.running_list = []
self.silence_dict: Dict[Path, datetime] = {}
self.power_sign = "NoAction"
self.if_ignore_silence = False
self.logger = get_logger("配置管理")
self.logger.info("")
@@ -593,8 +611,11 @@ class AppConfig(GlobalConfig):
self.PlanConfig = MultipleConfig([MaaPlanConfig])
self.QueueConfig = MultipleConfig([QueueConfig])
truststore.inject_into_ssl()
async def init_config(self) -> None:
"""初始化配置管理"""
await self.connect(self.config_path / "Config.json")
await self.ScriptConfig.connect(self.config_path / "ScriptConfig.json")
await self.PlanConfig.connect(self.config_path / "PlanConfig.json")
@@ -610,9 +631,7 @@ class AppConfig(GlobalConfig):
self.logger.info(f"添加脚本配置:{script}")
class_book = {"MAA": MaaConfig, "General": GeneralConfig}
return await self.ScriptConfig.add(class_book[script])
return await self.ScriptConfig.add(self.CLASS_BOOK[script])
async def get_script(self, script_id: Optional[str]) -> tuple[list, dict]:
"""获取脚本配置"""
@@ -838,9 +857,7 @@ class AppConfig(GlobalConfig):
self.logger.info(f"添加计划表:{script}")
class_book = {"MaaPlan": MaaPlanConfig}
return await self.PlanConfig.add(class_book[script])
return await self.PlanConfig.add(self.CLASS_BOOK[script])
async def get_plan(self, plan_id: Optional[str]) -> tuple[list, dict]:
"""获取计划表配置"""
@@ -976,5 +993,86 @@ class AppConfig(GlobalConfig):
dt = dt - timedelta(days=1)
return dt.date()
async def get_stage(self) -> tuple[bool, Dict[str, Dict[str, list]]]:
"""从MAA服务器更新活动关卡信息"""
self.logger.info("开始获取活动关卡信息")
response = requests.get(
"https://api.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
timeout=10,
proxies={
"http": self.get("Update", "ProxyAddress"),
"https": self.get("Update", "ProxyAddress"),
},
)
if response.status_code == 200:
stage_infos = response.json()["Official"]["sideStoryStage"]
if_get_maa_stage = True
else:
self.logger.warning(f"无法从MAA服务器获取活动关卡信息:{response.text}")
if_get_maa_stage = False
stage_infos = []
ss_stage_dict = {"value": [], "text": []}
for stage_info in stage_infos:
if (
datetime.strptime(
stage_info["Activity"]["UtcStartTime"], "%Y/%m/%d %H:%M:%S"
)
< datetime.now()
< datetime.strptime(
stage_info["Activity"]["UtcExpireTime"], "%Y/%m/%d %H:%M:%S"
)
):
ss_stage_dict["value"].append(stage_info["Value"])
ss_stage_dict["text"].append(stage_info["Value"])
stage_dict = {}
for day in range(0, 8):
today_stage_dict = {"value": [], "text": []}
for stage_info in self.STAGE_DAILY_INFO:
if day in stage_info["days"] or day == 0:
today_stage_dict["value"].append(stage_info["value"])
today_stage_dict["text"].append(stage_info["text"])
stage_dict[calendar.day_name[day - 1] if day > 0 else "ALL"] = {
"value": ss_stage_dict["value"] + today_stage_dict["value"],
"text": ss_stage_dict["text"] + today_stage_dict["text"],
}
return if_get_maa_stage, stage_dict
async def get_server_info(self, type: str) -> Dict[str, Any]:
"""获取公告信息"""
self.logger.info(f"开始从 AUTO_MAA 服务器获取 {type} 信息")
response = requests.get(
url=f"http://221.236.27.82:10197/d/AUTO_MAA/Server/{type}.json",
timeout=10,
proxies={
"http": self.get("Update", "ProxyAddress"),
"https": self.get("Update", "ProxyAddress"),
},
)
if response.status_code == 200:
return response.json()
else:
self.logger.warning(
f"无法从 AUTO_MAA 服务器获取 {type} 信息:{response.text}"
)
raise ConnectionError(
"Cannot connect to the notice server. Please check your network connection or try again later."
)
Config = AppConfig()

120
app/core/timer.py Normal file
View File

@@ -0,0 +1,120 @@
# 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
import asyncio
import keyboard
from datetime import datetime
from utils import get_logger
from .config import Config
class _MainTimer:
def __init__(self):
super().__init__()
self.logger = get_logger("主业务定时器")
async def second_task(self):
"""每秒定期任务"""
self.logger.info("每秒定期任务启动")
while True:
await self.set_silence()
await self.check_power()
await asyncio.sleep(1)
async def set_silence(self):
"""静默模式通过模拟老板键来隐藏模拟器窗口"""
if (
not Config.if_ignore_silence
and await Config.get("Function", "IfSilence")
and await Config.get("Function", "BossKey") != ""
):
pass
# windows = System.get_window_info()
# emulator_windows = []
# for window in windows:
# for emulator_path, endtime in Config.silence_dict.items():
# if (
# datetime.now() < endtime
# and str(emulator_path) in window
# and window[0] != "新通知" # 此处排除雷电名为新通知的窗口
# ):
# emulator_windows.append(window)
# if emulator_windows:
# logger.info(
# f"检测到模拟器窗口:{emulator_windows}", module="主业务定时器"
# )
# try:
# keyboard.press_and_release(
# "+".join(
# _.strip().lower()
# for _ in Config.get(Config.function_BossKey).split("+")
# )
# )
# logger.info(
# f"模拟按键:{Config.get(Config.function_BossKey)}",
# module="主业务定时器",
# )
# except Exception as e:
# logger.exception(f"模拟按键时出错:{e}", module="主业务定时器")
async def check_power(self):
"""检查电源操作"""
# if Config.power_sign != "NoAction" and not Config.running_list:
# logger.info(f"触发电源操作:{Config.power_sign}", module="主业务定时器")
# from app.ui import ProgressRingMessageBox
# mode_book = {
# "KillSelf": "退出软件",
# "Sleep": "睡眠",
# "Hibernate": "休眠",
# "Shutdown": "关机",
# "ShutdownForce": "关机(强制)",
# }
# choice = ProgressRingMessageBox(
# Config.main_window, f"{mode_book[Config.power_sign]}倒计时"
# )
# if choice.exec():
# logger.info(
# f"确认执行电源操作:{Config.power_sign}", module="主业务定时器"
# )
# System.set_power(Config.power_sign)
# Config.set_power_sign("NoAction")
# else:
# logger.info(f"取消电源操作:{Config.power_sign}", module="主业务定时器")
# Config.set_power_sign("NoAction")
MainTimer = _MainTimer()

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.
@@ -41,6 +42,7 @@ def main():
if is_admin():
import asyncio
import uvicorn
from fastapi import FastAPI
from contextlib import asynccontextmanager
@@ -48,17 +50,30 @@ def main():
@asynccontextmanager
async def lifespan(app: FastAPI):
from core import Config
from core import Config, MainTimer
await Config.init_config()
main_timer = asyncio.create_task(MainTimer.second_task())
yield
main_timer.cancel()
try:
await main_timer
except asyncio.CancelledError:
logger.info("主业务定时器已关闭")
logger.info("AUTO_MAA 后端程序关闭")
logger.info("----------------END----------------")
from fastapi.middleware.cors import CORSMiddleware
from api import scripts_router, plan_router, queue_router, setting_router
from api import (
info_router,
scripts_router,
plan_router,
queue_router,
setting_router,
)
app = FastAPI(
title="AUTO_MAA",
@@ -75,6 +90,7 @@ def main():
allow_headers=["*"], # 允许所有请求头
)
app.include_router(info_router)
app.include_router(scripts_router)
app.include_router(plan_router)
app.include_router(queue_router)

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.
@@ -292,7 +293,7 @@ class ConfigBase:
return data
async def get(self, group: str, name: str) -> Any:
def get(self, group: str, name: str) -> Any:
"""获取配置项的值"""
if not hasattr(self, f"{group}_{name}"):

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.
@@ -29,6 +30,10 @@ class OutBase(BaseModel):
message: str = Field(default="操作成功", description="操作消息")
class InfoOut(OutBase):
data: Dict[str, Any] = Field(..., description="收到的服务器数据")
class ScriptCreateIn(BaseModel):
type: Literal["MAA", "General"]

View File

@@ -1,5 +1,6 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# This file is part of AUTO_MAA.

View File

@@ -1,3 +1,25 @@
# AUTO_MAA:A MAA Multi Account Management and Automation Tool
# Copyright © 2024-2025 DLmaster361
# Copyright © 2025 MoeSnowyFox
# 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 loguru import logger as _logger
import sys
from pathlib import Path