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-17 10:53:00 +08:00
parent 1fdac22bea
commit 8e2c6bb642
7 changed files with 94 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/api/dispatch.py
View File

@@ -25,6 +25,7 @@ import asyncio
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Body, Path
from app.core import TaskManager, Broadcast
from app.services import System
from app.models.schema import *
router = APIRouter(prefix="/api/dispatch", tags=["任务调度"])
@@ -59,6 +60,18 @@ async def stop_task(task: DispatchIn = Body(...)) -> OutBase:
return OutBase()
@router.post("/power", summary="电源操作", response_model=OutBase, status_code=200)
async def power_task(task: PowerIn = Body(...)) -> OutBase:
try:
await System.set_power(task.signal)
except Exception as e:
return OutBase(
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"
)
return OutBase()
@router.websocket("/ws/{websocketId}")
async def websocket_endpoint(
websocket: WebSocket,

27
app/api/setting.py
View File

@@ -20,9 +20,13 @@
# Contact: DLmaster_361@163.com
import os
from pathlib import Path
import shutil
from fastapi import APIRouter, Body
from app.core import Config
from app.services import System
from app.models.schema import *
router = APIRouter(prefix="/api/setting", tags=["全局设置"])
@@ -49,7 +53,28 @@ async def update_script(script: SettingUpdateIn = Body(...)) -> OutBase:
"""更新配置"""
try:
await Config.update_setting(script.data.model_dump(exclude_unset=True))
data = script.data.model_dump(exclude_unset=True)
await Config.update_setting(data)
if data.get("Start", {}).get("IfSelfStart", None) is not None:
await System.set_SelfStart()
if data.get("Function", None) is not None:
function = data["Function"]
if function.get("IfAllowSleep", None) is not None:
await System.set_Sleep()
if function.get("IfSkipMumuSplashAds", None) is not None:
MuMu_splash_ads_path = (
Path(os.getenv("APPDATA") or "")
/ "Netease/MuMuPlayer-12.0/data/startupImage"
)
if Config.get("Function", "IfSkipMumuSplashAds"):
if MuMu_splash_ads_path.exists() and MuMu_splash_ads_path.is_dir():
shutil.rmtree(MuMu_splash_ads_path)
MuMu_splash_ads_path.touch()
else:
if MuMu_splash_ads_path.exists() and MuMu_splash_ads_path.is_file():
MuMu_splash_ads_path.unlink()
except Exception as e:
return OutBase(
code=500, status="error", message=f"{type(e).__name__}: {str(e)}"

1
app/core/config.py
View File

@@ -659,7 +659,6 @@ class AppConfig(GlobalConfig):
self.history_path.mkdir(parents=True, exist_ok=True)
self.silence_dict: Dict[Path, datetime] = {}
self.power_sign = "NoAction"
self.if_ignore_silence: List[uuid.UUID] = []
self.temp_task: List[asyncio.Task] = []

91
app/core/timer.py
View File

@@ -22,6 +22,7 @@ import asyncio
import keyboard
from datetime import datetime
from app.services import System
from app.utils import get_logger
from .config import Config
@@ -31,9 +32,6 @@ logger = get_logger("主业务定时器")
class _MainTimer:
def __init__(self):
super().__init__()
async def second_task(self):
"""每秒定期任务"""
logger.info("每秒定期任务启动")
@@ -41,7 +39,6 @@ class _MainTimer:
while True:
await self.set_silence()
await self.check_power()
await asyncio.sleep(1)
@@ -54,68 +51,36 @@ class _MainTimer:
and Config.get("Function", "BossKey") != ""
):
pass
windows = await System.get_window_info()
# 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)
# 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:
# 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")
logger.info(
f"检测到模拟器窗口：{emulator_windows}", module="主业务定时器"
)
try:
keyboard.press_and_release(
"+".join(
_.strip().lower()
for _ in Config.get("Function", "BossKey").split("+")
)
)
logger.info(
f"模拟按键：{Config.get('Function', 'BossKey')}",
module="主业务定时器",
)
except Exception as e:
logger.exception(f"模拟按键时出错：{e}", module="主业务定时器")
MainTimer = _MainTimer()

6
app/models/schema.py
View File

@@ -669,6 +669,12 @@ class TaskMessage(BaseModel):
data: Dict[str, Any] = Field(..., description="消息数据，具体内容根据type类型而定")
class PowerIn(BaseModel):
signal: Literal[
"NoAction", "Shutdown", "ShutdownForce", "Hibernate", "Sleep", "KillSelf"
] = Field(..., description="电源操作信号")
class HistorySearchIn(BaseModel):
mode: Literal["按日合并", "按周合并", "按月合并"] = Field(
..., description="合并模式"

37
app/services/system.py
View File

@@ -29,6 +29,7 @@ import tempfile
import getpass
from datetime import datetime
from pathlib import Path
from typing import Literal
from app.core import Config
from app.utils.logger import get_logger
@@ -41,12 +42,7 @@ class _SystemHandler:
ES_CONTINUOUS = 0x80000000
ES_SYSTEM_REQUIRED = 0x00000001
def __init__(self):
self.set_Sleep()
self.set_SelfStart()
def set_Sleep(self) -> None:
async def set_Sleep(self) -> None:
"""同步系统休眠状态"""
if Config.get("Function", "IfAllowSleep"):
@@ -58,12 +54,10 @@ class _SystemHandler:
# 恢复系统电源状态
ctypes.windll.kernel32.SetThreadExecutionState(self.ES_CONTINUOUS)
def set_SelfStart(self) -> None:
async def set_SelfStart(self) -> None:
"""同步开机自启"""
return None # 目前不支持开机自启
if Config.get("Function", "IfSelfStart") and not self.is_startup():
if Config.get("Start", "IfSelfStart") and not await self.is_startup():
# 创建任务计划
try:
@@ -114,7 +108,7 @@ class _SystemHandler:
</Settings>
<Actions Context="Author">
<Exec>
<Command>"{Config.app_path_sys}"</Command>
<Command>"{Path.cwd() / 'AUTO_MAA.exe'}"</Command>
</Exec>
</Actions>
</Task>"""
@@ -145,7 +139,7 @@ class _SystemHandler:
if result.returncode == 0:
logger.success(
f"程序自启动任务计划已创建: {Config.app_path_sys}",
f"程序自启动任务计划已创建: {Path.cwd() / 'AUTO_MAA.exe'}",
module="系统服务",
)
else:
@@ -164,7 +158,7 @@ class _SystemHandler:
except Exception as e:
logger.exception(f"程序自启动任务计划创建失败: {e}")
elif not Config.get(Config.start_IfSelfStart) and self.is_startup():
elif not Config.get("Start", "IfSelfStart") and await self.is_startup():
try:
@@ -187,11 +181,16 @@ class _SystemHandler:
except Exception as e:
logger.exception(f"程序自启动任务计划删除失败: {e}")
def set_power(self, mode) -> None:
async def set_power(
self,
mode: Literal[
"NoAction", "Shutdown", "ShutdownForce", "Hibernate", "Sleep", "KillSelf"
],
) -> None:
"""
执行系统电源操作
:param mode: 电源操作模式，支持 "NoAction", "Shutdown", "Hibernate", "Sleep", "KillSelf", "ShutdownForce"
:param mode: 电源操作
"""
if sys.platform.startswith("win"):
@@ -202,7 +201,7 @@ class _SystemHandler:
elif mode == "Shutdown":
self.kill_emulator_processes()
await self.kill_emulator_processes()
logger.info("执行关机操作")
subprocess.run(["shutdown", "/s", "/t", "0"])
@@ -253,7 +252,7 @@ class _SystemHandler:
logger.info("执行退出主程序操作")
sys.exit(0)
def kill_emulator_processes(self):
async def kill_emulator_processes(self):
"""这里暂时仅支持 MuMu 模拟器"""
logger.info("正在清除模拟器进程")
@@ -270,7 +269,7 @@ class _SystemHandler:
logger.success("模拟器进程清除完成")
def is_startup(self) -> bool:
async def is_startup(self) -> bool:
"""判断程序是否已经开机自启"""
try:
@@ -286,7 +285,7 @@ class _SystemHandler:
logger.exception(f"检查任务计划程序失败: {e}")
return False
def get_window_info(self) -> list:
async def get_window_info(self) -> list:
"""获取当前前台窗口信息"""
def callback(hwnd, window_info):

3
main.py
View File

@@ -56,11 +56,14 @@ def main():
async def lifespan(app: FastAPI):
from app.core import Config, MainTimer
from app.services import System
await Config.init_config()
await Config.get_stage()
await Config.clean_old_history()
main_timer = asyncio.create_task(MainTimer.second_task())
await System.set_Sleep()
await System.set_SelfStart()
yield