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-09-09 21:00:11 +08:00
parent 3490eff76e
commit e3c1143c0c
8 changed files with 76 additions and 395 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
app/api/info.py
View File

@@ -28,6 +28,34 @@ from app.models.schema import *
router = APIRouter(prefix="/api/info", tags=["信息获取"])
@router.post(
"/version",
summary="获取后端git版本信息",
response_model=VersionOut,
status_code=200,
)
async def get_git_version() -> VersionOut:
try:
is_latest, commit_hash, commit_time = await Config.get_git_version()
except Exception as e:
return VersionOut(
code=500,
status="error",
message=f"{type(e).__name__}: {str(e)}",
if_latest=False,
current_hash="",
current_time="",
current_version="",
)
return VersionOut(
if_latest=is_latest,
current_hash=commit_hash,
current_time=commit_time,
current_version=Config.version(),
)
@router.post(
"/combox/stage",
summary="获取关卡号下拉框信息",

25
app/core/config.py
View File

@@ -28,11 +28,11 @@ import sqlite3
import calendar
import requests
import truststore
from git import Repo
from pathlib import Path
from fastapi import WebSocket
from urllib.parse import quote
from collections import defaultdict
from datetime import datetime, timedelta, date, timezone
from datetime import datetime, timedelta, date
from typing import Literal, Optional
from app.models.ConfigBase import *
@@ -597,6 +597,7 @@ class AppConfig(GlobalConfig):
self.config_path.mkdir(parents=True, exist_ok=True)
self.history_path.mkdir(parents=True, exist_ok=True)
self.repo = Repo(Path.cwd())
self.server: Optional[uvicorn.Server] = None
self.websocket: Optional[WebSocket] = None
self.silence_dict: Dict[Path, datetime] = {}
@@ -908,6 +909,26 @@ class AppConfig(GlobalConfig):
else:
await Config.websocket.send_json(data)
async def get_git_version(self) -> tuple[bool, str, str]:
# 获取当前 commit
current_commit = self.repo.head.commit
# 获取 commit 哈希
commit_hash = current_commit.hexsha
# 获取 commit 时间
commit_time = datetime.fromtimestamp(current_commit.committed_date)
# 检查是否为最新 commit
# 获取远程分支的最新 commit
origin = self.repo.remotes.origin
origin.fetch() # 拉取最新信息
remote_commit = self.repo.commit(f"origin/{self.repo.active_branch.name}")
is_latest = bool(current_commit.hexsha == remote_commit.hexsha)
return is_latest, commit_hash, commit_time.strftime("%Y-%m-%d %H:%M:%S")
async def add_script(
self, script: Literal["MAA", "General"]
) -> tuple[uuid.UUID, ConfigBase]:

7
app/models/schema.py
View File

@@ -34,6 +34,13 @@ class InfoOut(OutBase):
data: Dict[str, Any] = Field(..., description="收到的服务器数据")
class VersionOut(OutBase):
if_latest: bool = Field(..., description="后端代码是否为最新")
current_hash: str = Field(..., description="后端代码当前哈希值")
current_time: str = Field(..., description="后端代码当前时间戳")
current_version: str = Field(..., description="后端当前版本号")
class NoticeOut(OutBase):
if_need_show: bool = Field(..., description="是否需要显示公告")
data: Dict[str, str] = Field(

26
app/services/update.py
View File

@@ -22,6 +22,7 @@
import re
import time
import json
import asyncio
import zipfile
import requests
import subprocess
@@ -209,7 +210,7 @@ class _UpdateHandler:
f"连接失败: {download_url}, 状态码: {response.status_code}, 剩余重试次数: {check_times}"
)
time.sleep(1)
await asyncio.sleep(1)
continue
logger.info(f"连接成功: {download_url}, 状态码: {response.status_code}")
@@ -225,6 +226,17 @@ class _UpdateHandler:
f.write(chunk)
downloaded_size += len(chunk)
# 更新指定线程的下载进度, 每秒更新一次
if time.time() - last_time >= 1.0:
speed = (
(downloaded_size - last_download_size)
/ (time.time() - last_time)
/ 1024
)
last_download_size = downloaded_size
last_time = time.time()
await Config.send_json(
WebSocketMessage(
id="Update",
@@ -237,16 +249,6 @@ class _UpdateHandler:
).model_dump()
)
# 更新指定线程的下载进度, 每秒更新一次
if time.time() - last_time >= 1.0:
speed = (
(downloaded_size - last_download_size)
/ (time.time() - last_time)
/ 1024
)
last_download_size = downloaded_size
last_time = time.time()
(Path.cwd() / "download.temp").rename(
Path.cwd() / f"UpdatePack_{self.remote_version}.zip"
)
@@ -276,7 +278,7 @@ class _UpdateHandler:
logger.info(
f"下载出错: {download_url}, 错误信息: {e}, 剩余重试次数: {check_times}"
)
time.sleep(1)
await asyncio.sleep(1)
else:

2
app/task/skland.py
View File

@@ -249,7 +249,7 @@ async def skland_sign_in(token) -> dict:
f"{character.get("nickName")}（{character.get("channelName")}）"
)
time.sleep(3)
await asyncio.sleep(3)
return result

1
pyproject.toml
View File

@@ -12,6 +12,7 @@ dependencies = [
"websockets==15.0.1",
"aiofiles==24.1.0",
"aiohttp==3.12.15",
"gitpython==3.1.45",
"plyer==2.1.0",
"psutil==7.0.0",
"jinja2==3.1.6",

1
requirements.txt
View File

@@ -5,6 +5,7 @@ uvicorn==0.35.0
websockets==15.0.1
aiofiles==24.1.0
aiohttp==3.12.15
gitpython==3.1.45
plyer==2.1.0
psutil==7.0.0
jinja2==3.1.6

379
updater.py
View File

@@ -1,379 +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
import re
import sys
import time
import json
import psutil
import base64
import zipfile
import requests
import argparse
import truststore
import subprocess
import win32crypt
from packaging import version
from pathlib import Path
from typing import List, Dict
current_dir = Path(__file__).resolve().parent
if str(current_dir) not in sys.path:
sys.path.insert(0, str(current_dir))
MIRROR_ERROR_INFO = {
1001: "获取版本信息的URL参数不正确",
7001: "填入的 CDK 已过期",
7002: "填入的 CDK 错误",
7003: "填入的 CDK 今日下载次数已达上限",
7004: "填入的 CDK 类型和待下载的资源不匹配",
7005: "填入的 CDK 已被封禁",
8001: "对应架构和系统下的资源不存在",
8002: "错误的系统参数",
8003: "错误的架构参数",
8004: "错误的更新通道参数",
1: "未知错误类型",
}
def dpapi_decrypt(note: str, entropy: None | bytes = None) -> str:
"""
使用Windows DPAPI解密数据
:param note: 数据密文
:type note: str
:param entropy: 随机熵
:type entropy: bytes
:return: 解密后的明文
:rtype: str
"""
if note == "":
return ""
decrypted = win32crypt.CryptUnprotectData(
base64.b64decode(note), entropy, None, None, 0
)
return decrypted[1].decode("utf-8")
def kill_process(path: Path) -> None:
"""
根据路径中止进程
:param path: 进程路径
"""
print(f"开始中止进程: {path}")
for pid in search_pids(path):
killprocess = subprocess.Popen(
f"taskkill /F /T /PID {pid}",
shell=True,
creationflags=subprocess.CREATE_NO_WINDOW,
)
killprocess.wait()
print(f"进程已中止: {path}")
def search_pids(path: Path) -> list:
"""
根据路径查找进程PID
:param path: 进程路径
:return: 匹配的进程PID列表
"""
print(f"开始查找进程 PID: {path}")
pids = []
for proc in psutil.process_iter(["pid", "exe"]):
try:
if proc.info["exe"] and proc.info["exe"].lower() == str(path).lower():
pids.append(proc.info["pid"])
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
# 进程可能在此期间已结束或无法访问, 忽略这些异常
pass
return pids
truststore.inject_into_ssl()
parser = argparse.ArgumentParser(
prog="AUTO-MAS更新器", description="为AUTO-MAS前端提供更新服务"
)
parser.add_argument(
"--version", "-v", type=str, required=False, default=None, help="前端程序版本号"
)
args = parser.parse_args()
if (Path.cwd() / "config/Config.json").exists():
config = json.loads(
(Path.cwd() / "config/Config.json").read_text(encoding="utf-8")
).get(
"Update",
{
"MirrorChyanCDK": "",
"ProxyAddress": "",
"Source": "GitHub",
"UpdateType": "stable",
},
)
else:
config = {
"MirrorChyanCDK": "",
"ProxyAddress": "",
"Source": "GitHub",
"UpdateType": "stable",
}
if (
config.get("Source", "GitHub") == "MirrorChyan"
and dpapi_decrypt(config.get("MirrorChyanCDK", "")) == ""
):
print("使用 MirrorChyan源但未填写 MirrorChyanCDK, 转用 GitHub 源")
config["Source"] = "GitHub"
config["MirrorChyanCDK"] = ""
print(f"当前配置: {config}")
download_source = config.get("Source", "GitHub")
proxies = {
"http": config.get("ProxyAddress", ""),
"https": config.get("ProxyAddress", ""),
}
if args.version:
current_version = args.version
else:
current_version = "v0.0.0"
print(f"当前版本: {current_version}")
response = requests.get(
f"https://mirrorchyan.com/api/resources/AUTO_MAA/latest?user_agent=AutoMaaGui&current_version={current_version}&cdk={dpapi_decrypt(config.get('MirrorChyanCDK', ''))}&channel={config.get('UpdateType', 'stable')}",
timeout=10,
proxies=proxies,
)
if response.status_code == 200:
version_info = response.json()
else:
try:
result = response.json()
if result["code"] != 0:
if result["code"] in MIRROR_ERROR_INFO:
print(f"获取版本信息时出错: {MIRROR_ERROR_INFO[result['code']]}")
else:
print(
"获取版本信息时出错: 意料之外的错误, 请及时联系项目组以获取来自 Mirror 酱的技术支持"
)
print(f" {result['msg']}")
sys.exit(1)
except Exception:
print(f"获取版本信息时出错: {response.text}")
sys.exit(1)
remote_version = version_info["data"]["version_name"]
if version.parse(remote_version) > version.parse(current_version):
# 版本更新信息
print(f"发现新版本: {remote_version}, 当前版本: {current_version}")
version_info_json: Dict[str, Dict[str, List[str]]] = json.loads(
re.sub(
r"^<!--\s*(.*?)\s*-->$",
r"\1",
version_info["data"]["release_note"].splitlines()[0],
)
)
update_version_info = {}
for v_i in [
info
for ver, info in version_info_json.items()
if version.parse(ver) > version.parse(current_version)
]:
for key, value in v_i.items():
if key not in update_version_info:
update_version_info[key] = []
update_version_info[key] += value
for key, value in update_version_info.items():
print(f"{key}: ")
for v in value:
print(f" - {v}")
if download_source == "GitHub":
download_url = f"https://github.com/DLmaster361/AUTO_MAA/releases/download/{remote_version}/AUTO_MAA_{remote_version}.zip"
elif download_source == "MirrorChyan":
if "url" in version_info["data"]:
with requests.get(
version_info["data"]["url"],
allow_redirects=True,
timeout=10,
stream=True,
proxies=proxies,
) as response:
if response.status_code == 200:
download_url = response.url
else:
print(f"MirrorChyan 未返回下载链接, 使用自建下载站")
download_url = f"https://download.auto-mas.top/d/AUTO_MAA/AUTO_MAA_{remote_version}.zip"
elif download_source == "AutoSite":
download_url = (
f"https://download.auto-mas.top/d/AUTO_MAA/AUTO_MAA_{remote_version}.zip"
)
else:
print(f"未知的下载源: {download_source}, 请检查配置文件")
sys.exit(1)
print(f"开始下载: {download_url}")
# 清理可能存在的临时文件
if (Path.cwd() / "download.temp").exists():
(Path.cwd() / "download.temp").unlink()
check_times = 3
while check_times != 0:
try:
start_time = time.time()
response = requests.get(
download_url, timeout=10, stream=True, proxies=proxies
)
if response.status_code not in [200, 206]:
if check_times != -1:
check_times -= 1
print(
f"连接失败: {download_url}, 状态码: {response.status_code}, 剩余重试次数: {check_times}",
)
time.sleep(1)
continue
print(f"连接成功: {download_url}, 状态码: {response.status_code}")
file_size = int(response.headers.get("content-length", 0))
downloaded_size = 0
last_download_size = 0
last_time = time.time()
with (Path.cwd() / "download.temp").open(mode="wb") as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
downloaded_size += len(chunk)
# 更新指定线程的下载进度, 每秒更新一次
if time.time() - last_time >= 1.0:
speed = (
(downloaded_size - last_download_size)
/ (time.time() - last_time)
/ 1024
)
last_download_size = downloaded_size
last_time = time.time()
if speed >= 1024:
print(
f"正在下载: AUTO-MAS 已下载: {downloaded_size / 1048576:.2f}/{file_size / 1048576:.2f} MB （{downloaded_size / file_size * 100:.2f}%） 下载速度: {speed / 1024:.2f} MB/s",
)
else:
print(
f"正在下载: AUTO-MAS 已下载: {downloaded_size / 1048576:.2f}/{file_size / 1048576:.2f} MB （{downloaded_size / file_size * 100:.2f}%） 下载速度: {speed:.2f} KB/s",
)
print(
f"下载完成: {download_url}, 实际下载大小: {downloaded_size} 字节, 耗时: {time.time() - start_time:.2f} 秒",
)
break
except Exception as e:
if check_times != -1:
check_times -= 1
print(
f"下载出错: {download_url}, 错误信息: {e}, 剩余重试次数: {check_times}",
)
time.sleep(1)
else:
if (Path.cwd() / "download.temp").exists():
(Path.cwd() / "download.temp").unlink()
print(f"下载失败: {download_url}")
sys.exit(1)
print(f"开始解压: {Path.cwd() / 'download.temp'} 到 {Path.cwd()}")
while True:
try:
with zipfile.ZipFile(Path.cwd() / "download.temp", "r") as zip_ref:
zip_ref.extractall(Path.cwd())
print(f"解压完成: {Path.cwd() / 'download.temp'} 到 {Path.cwd()}")
break
except PermissionError:
print(f"解压出错: AUTO_MAA正在运行, 正在尝试将其关闭")
kill_process(Path.cwd() / "AUTO_MAA.exe")
time.sleep(1)
print("正在删除临时文件")
if (Path.cwd() / "changes.json").exists():
(Path.cwd() / "changes.json").unlink()
if (Path.cwd() / "download.temp").exists():
(Path.cwd() / "download.temp").unlink()
print("正在启动AUTO_MAA")
subprocess.Popen(
[Path.cwd() / "AUTO_MAA.exe"],
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
| subprocess.DETACHED_PROCESS
| subprocess.CREATE_NO_WINDOW,
)
print("更新完成")
sys.exit(0)
else:
print(f"当前版本为最新版本: {current_version}")
sys.exit(0)