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

feat: AUTO_MAA 配置分享中心上线

Browse Source
This commit is contained in:
DLmaster361
2025-07-19 21:38:38 +08:00
parent 8d6071f794
commit eb1fade6f5
8 changed files with 380 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/core/config.py
View File

@@ -597,7 +597,7 @@ class GeneralConfig(LQConfig):
super().__init__()
self.Script_Name = ConfigItem("Script", "Name", "")
self.Script_RootPath = ConfigItem("Script", "RootPath", ".", FolderValidator())
self.Script_RootPath = ConfigItem("Script", "RootPath", ".", FileValidator())
self.Script_ScriptPath = ConfigItem(
"Script", "ScriptPath", ".", FileValidator()
)

121
app/core/network.py
View File

@@ -31,6 +31,7 @@ import time
import requests
import truststore
from pathlib import Path
from typing import Dict
from .logger import logger
@@ -42,7 +43,14 @@ class NetworkThread(QThread):
timeout = 10
backoff_factor = 0.1
def __init__(self, mode: str, url: str, path: Path = None) -> None:
def __init__(
self,
mode: str,
url: str,
path: Path = None,
files: Dict = None,
data: Dict = None,
) -> None:
super().__init__()
self.setObjectName(
@@ -54,6 +62,8 @@ class NetworkThread(QThread):
self.mode = mode
self.url = url
self.path = path
self.files = files
self.data = data
from .config import Config
@@ -78,6 +88,8 @@ class NetworkThread(QThread):
self.get_json(self.url)
elif self.mode == "get_file":
self.get_file(self.url, self.path)
elif self.mode == "upload_file":
self.upload_file(self.url, self.files, self.data)
def get_json(self, url: str) -> None:
"""
@@ -102,7 +114,7 @@ class NetworkThread(QThread):
self.response_json = None
self.error_message = str(e)
logger.exception(
f"子线程 {self.objectName()} 网络请求失败：{e}",
f"子线程 {self.objectName()} 网络请求失败：{e}，第{_+1}次尝试",
module="网络请求子线程",
)
time.sleep(self.backoff_factor)
@@ -122,14 +134,15 @@ class NetworkThread(QThread):
response = None
try:
response = requests.get(url, timeout=10, proxies=self.proxies)
response = requests.get(url, timeout=self.timeout, proxies=self.proxies)
if response.status_code == 200:
with open(path, "wb") as file:
file.write(response.content)
self.status_code = response.status_code
self.error_message = None
else:
self.status_code = response.status_code
self.error_message = "下载失败"
self.error_message = f"下载失败，状态码: {response.status_code}"
except Exception as e:
self.status_code = response.status_code if response else None
@@ -140,6 +153,54 @@ class NetworkThread(QThread):
self.loop.quit()
def upload_file(self, url: str, files: Dict, data: Dict = None) -> None:
"""
通过POST方法上传文件
:param url: 请求的URL
:param files: 文件字典，格式为 {'file': ('filename', file_obj, 'content_type')}
:param data: 表单数据字典
"""
logger.info(f"子线程 {self.objectName()} 开始上传文件", module="网络请求子线程")
response = None
for _ in range(self.max_retries):
try:
response = requests.post(
url,
files=files,
data=data,
timeout=self.timeout,
proxies=self.proxies,
)
self.status_code = response.status_code
print(response.text)
# 尝试解析JSON响应
try:
self.response_json = response.json()
except ValueError:
# 如果不是JSON格式，保存文本内容
self.response_json = {"text": response.text}
self.error_message = None
break
except Exception as e:
self.status_code = response.status_code if response else None
self.response_json = None
self.error_message = str(e)
logger.exception(
f"子线程 {self.objectName()} 文件上传失败：{e}，第{_+1}次尝试",
module="网络请求子线程",
)
time.sleep(self.backoff_factor)
self.loop.quit()
class _Network(QObject):
"""网络请求线程管理类"""
@@ -149,19 +210,28 @@ class _Network(QObject):
self.task_queue = []
def add_task(self, mode: str, url: str, path: Path = None) -> NetworkThread:
def add_task(
self,
mode: str,
url: str,
path: Path = None,
files: Dict = None,
data: Dict = None,
) -> NetworkThread:
"""
添加网络请求任务
:param mode: 请求模式，支持 "get", "get_file"
:param mode: 请求模式，支持 "get", "get_file", "upload_file"
:param url: 请求的URL
:param path: 下载文件的保存路径，仅在 mode 为 "get_file" 时有效
:param files: 上传文件字典，仅在 mode 为 "upload_file" 时有效
:param data: 表单数据字典，仅在 mode 为 "upload_file" 时有效
:return: 返回创建的 NetworkThread 实例
"""
logger.info(f"添加网络请求任务: {mode} {url} {path}", module="网络请求")
network_thread = NetworkThread(mode, url, path)
network_thread = NetworkThread(mode, url, path, files, data)
self.task_queue.append(network_thread)
@@ -169,6 +239,43 @@ class _Network(QObject):
return network_thread
def upload_config_file(
self, file_path: Path, username: str = "", description: str = ""
) -> NetworkThread:
"""
上传配置文件到分享服务器
:param file_path: 要上传的文件路径
:param username: 用户名（可选）
:param description: 文件描述（必填）
:return: 返回创建的 NetworkThread 实例
"""
if not file_path.exists():
raise FileNotFoundError(f"文件不存在: {file_path}")
if not description:
raise ValueError("文件描述不能为空")
# 准备上传的文件
with open(file_path, "rb") as f:
files = {"file": (file_path.name, f.read(), "application/json")}
# 准备表单数据
data = {"description": description}
if username:
data["username"] = username
url = "http://221.236.27.82:10023/api/upload/share"
logger.info(
f"准备上传配置文件: {file_path.name}，用户: {username or '匿名'}，描述: {description}",
extra={"module": "网络请求"},
)
return self.add_task("upload_file", url, files=files, data=data)
def get_result(self, network_thread: NetworkThread) -> dict:
"""
获取网络请求结果

9
app/ui/Widget.py
View File

@@ -255,7 +255,9 @@ class NoticeMessageBox(MessageBoxBase):
self.button_cancel.clicked.connect(self.cancelButton.click)
self.index.index_cards[0].clicked.emit()
def __update_text(self, text: str):
def __update_text(self, index: int, text: str):
self.currentIndex = index
html = markdown.markdown(text).replace("\n", "")
html = re.sub(
@@ -273,7 +275,7 @@ class NoticeMessageBox(MessageBoxBase):
class NoticeIndexCard(HeaderCardWidget):
index_changed = Signal(str)
index_changed = Signal(int, str)
def __init__(self, title: str, content: Dict[str, str], parent=None):
super().__init__(parent)
@@ -289,12 +291,13 @@ class NoticeMessageBox(MessageBoxBase):
self.index_cards.append(QuantifiedItemCard([index, ""]))
self.index_cards[-1].clicked.connect(
partial(self.index_changed.emit, text)
partial(self.index_changed.emit, len(self.index_cards), text)
)
self.Layout.addWidget(self.index_cards[-1])
if not content:
self.Layout.addWidget(QuantifiedItemCard(["暂无信息", ""]))
self.currentIndex = 0
self.Layout.addStretch(1)

2
app/ui/home.py
View File

@@ -204,7 +204,7 @@ class Home(QWidget):
# 从远程服务器获取最新主题图像信息
network = Network.add_task(
mode="get",
url="https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/theme_image.json",
url="http://221.236.27.82:10197/d/AUTO_MAA/Server/theme_image.json",
)
network.loop.exec()
network_result = Network.get_result(network)

248
app/ui/member_manager.py
View File

@@ -54,7 +54,7 @@ from PySide6.QtCore import Signal
from datetime import datetime
from functools import partial
from pathlib import Path
from typing import List, Union, Type
from typing import List, Dict, Union, Type
import shutil
import json
@@ -94,6 +94,7 @@ from .Widget import (
PushAndComboBoxSettingCard,
StatusSwitchSetting,
UserNoticeSettingCard,
NoticeMessageBox,
PivotArea,
)
@@ -393,7 +394,7 @@ class MemberManager(QWidget):
# 从远程服务器获取应用列表
network = Network.add_task(
mode="get",
url="https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/apps_info.json",
url="http://221.236.27.82:10197/d/AUTO_MAA/Server/apps_info.json",
)
network.loop.exec()
network_result = Network.get_result(network)
@@ -2763,14 +2764,32 @@ class MemberManager(QWidget):
content="选择一个保存路径，将当前配置信息导出到文件",
parent=self,
)
self.card_ImportFromWeb = PushSettingCard(
text="查看",
icon=FluentIcon.PAGE_RIGHT,
title="从「AUTO_MAA 配置分享中心」导入",
content="从「AUTO_MAA 配置分享中心」选择一个用户分享的通用配置模板，导入其中的配置信息",
parent=self,
)
self.card_UploadToWeb = PushSettingCard(
text="上传",
icon=FluentIcon.PAGE_RIGHT,
title="上传到「AUTO_MAA 配置分享中心」",
content="将当前通用配置分享到「AUTO_MAA 配置分享中心」，通过审核后可供其他用户下载使用",
parent=self,
)
self.card_ImportFromFile.clicked.connect(self.import_from_file)
self.card_ExportToFile.clicked.connect(self.export_to_file)
self.card_ImportFromWeb.clicked.connect(self.import_from_web)
self.card_UploadToWeb.clicked.connect(self.upload_to_web)
widget = QWidget()
Layout = QVBoxLayout(widget)
Layout.addWidget(self.card_ImportFromFile)
Layout.addWidget(self.card_ExportToFile)
Layout.addWidget(self.card_ImportFromWeb)
Layout.addWidget(self.card_UploadToWeb)
self.viewLayout.setContentsMargins(0, 0, 0, 0)
self.viewLayout.setSpacing(0)
self.addGroupWidget(widget)
@@ -2791,6 +2810,16 @@ class MemberManager(QWidget):
Config.member_dict[self.name]["Path"] / "config.json"
)
logger.success(
f"{self.name} 配置导入成功", module="脚本管理"
)
MainInfoBar.push_info_bar(
"success",
"操作成功",
f"{self.name} 配置导入成功",
3000,
)
def export_to_file(self):
"""导出配置到文件"""
@@ -2800,10 +2829,225 @@ class MemberManager(QWidget):
if file_path:
temp = self.config.toDict()
# 移除配置中可能存在的隐私信息
temp["Script"]["Name"] = Path(file_path).stem
for path in ["ScriptPath", "ConfigPath", "LogPath"]:
if Path(temp["Script"][path]).is_relative_to(
Path(temp["Script"]["RootPath"])
):
temp["Script"][path] = str(
Path(r"C:/脚本根目录")
/ Path(temp["Script"][path]).relative_to(
Path(temp["Script"]["RootPath"])
)
)
temp["Script"]["RootPath"] = str(Path(r"C:/脚本根目录"))
with open(file_path, "w", encoding="utf-8") as file:
json.dump(temp, file, ensure_ascii=False, indent=4)
logger.success(
f"{self.name} 配置导出成功", module="脚本管理"
)
MainInfoBar.push_info_bar(
"success",
"操作成功",
f"{self.name} 配置导出成功",
3000,
)
def import_from_web(self):
"""从「AUTO_MAA 配置分享中心」导入配置"""
# 从远程服务器获取配置列表
network = Network.add_task(
mode="get",
url="http://221.236.27.82:10023/api/list/config/general",
)
network.loop.exec()
network_result = Network.get_result(network)
if network_result["status_code"] == 200:
config_info: List[Dict[str, str]] = network_result[
"response_json"
]
else:
logger.warning(
f"获取配置列表时出错：{network_result['error_message']}",
module="脚本管理",
)
MainInfoBar.push_info_bar(
"warning",
"获取配置列表时出错",
f"网络错误：{network_result['status_code']}",
5000,
)
return None
choice = NoticeMessageBox(
self.window(),
"配置分享中心",
{
_[
"configName"
]: f"""
# {_['configName']}
- **作者**: {_['author']}
- **发布时间**：{_['createTime']}
- **描述**：{_['description']}
"""
for _ in config_info
},
)
if choice.exec() and choice.currentIndex != 0:
# 从远程服务器获取具体配置
network = Network.add_task(
mode="get",
url=config_info[choice.currentIndex - 1]["downloadUrl"],
)
network.loop.exec()
network_result = Network.get_result(network)
if network_result["status_code"] == 200:
config_data = network_result["response_json"]
else:
logger.warning(
f"获取配置列表时出错：{network_result['error_message']}",
module="脚本管理",
)
MainInfoBar.push_info_bar(
"warning",
"获取配置列表时出错",
f"网络错误：{network_result['status_code']}",
5000,
)
return None
with (
Config.member_dict[self.name]["Path"] / "config.json"
).open("w", encoding="utf-8") as file:
json.dump(
config_data, file, ensure_ascii=False, indent=4
)
self.config.load(
Config.member_dict[self.name]["Path"] / "config.json"
)
logger.success(
f"{self.name} 配置导入成功", module="脚本管理"
)
MainInfoBar.push_info_bar(
"success",
"操作成功",
f"{self.name} 配置导入成功",
3000,
)
def upload_to_web(self):
"""上传配置到「AUTO_MAA 配置分享中心」"""
choice = LineEditMessageBox(
self.window(), "请输入你的用户名", "用户名", "明文"
)
choice.input.setMinimumWidth(200)
if choice.exec() and choice.input.text() != "":
author = choice.input.text()
choice = LineEditMessageBox(
self.window(), "请输入配置名称", "配置名称", "明文"
)
choice.input.setMinimumWidth(200)
if choice.exec() and choice.input.text() != "":
config_name = choice.input.text()
choice = LineEditMessageBox(
self.window(),
"请描述一下您要分享的配置",
"配置描述",
"明文",
)
choice.input.setMinimumWidth(300)
if choice.exec() and choice.input.text() != "":
description = choice.input.text()
temp = self.config.toDict()
# 移除配置中可能存在的隐私信息
temp["Script"]["Name"] = config_name
for path in ["ScriptPath", "ConfigPath", "LogPath"]:
if Path(temp["Script"][path]).is_relative_to(
Path(temp["Script"]["RootPath"])
):
temp["Script"][path] = str(
Path(r"C:/脚本根目录")
/ Path(
temp["Script"][path]
).relative_to(
Path(temp["Script"]["RootPath"])
)
)
temp["Script"]["RootPath"] = str(
Path(r"C:/脚本根目录")
)
files = {
"file": (
f"{config_name}&&{author}&&{description}&&{int(datetime.now().timestamp() * 1000)}.json",
json.dumps(temp, ensure_ascii=False),
"application/json",
)
}
data = {
"username": author,
"description": description,
}
# 配置上传至远程服务器
network = Network.add_task(
"upload_file",
"http://221.236.27.82:10023/api/upload/share",
files=files,
data=data,
)
network.loop.exec()
network_result = Network.get_result(network)
if network_result["status_code"] == 200:
response = network_result["response_json"]
else:
logger.warning(
f"上传配置时出错：{network_result['error_message']}",
module="脚本管理",
)
MainInfoBar.push_info_bar(
"warning",
"上传配置时出错",
f"网络错误：{network_result['status_code']}",
5000,
)
return None
logger.success(
f"{self.name} 配置上传成功", module="脚本管理"
)
MainInfoBar.push_info_bar(
"success",
"上传配置成功",
(
response["message"]
if "message" in response
else response["text"]
),
5000,
)
class BranchManager(HeaderCardWidget):
"""分支管理父页面"""

4
app/ui/setting.py
View File

@@ -473,7 +473,7 @@ class Setting(QWidget):
# 从远程服务器获取代理信息
network = Network.add_task(
mode="get",
url="https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/download_info.json",
url="http://221.236.27.82:10197/d/AUTO_MAA/Server/download_info.json",
)
network.loop.exec()
network_result = Network.get_result(network)
@@ -565,7 +565,7 @@ class Setting(QWidget):
# 从远程服务器获取最新公告
network = Network.add_task(
mode="get",
url="https://gitee.com/DLmaster_361/AUTO_MAA/raw/server/notice.json",
url="http://221.236.27.82:10197/d/AUTO_MAA/Server/notice.json",
)
network.loop.exec()
network_result = Network.get_result(network)

6
main.py
View File

@@ -26,6 +26,10 @@ v4.4
"""
import os
import sys
# Nuitka环境检测和修复
def setup_nuitka_compatibility():
"""设置Nuitka打包环境的兼容性"""
@@ -84,8 +88,6 @@ def no_print(*args, **kwargs):
builtins.print = no_print
import os
import sys
import ctypes
import traceback
from PySide6.QtWidgets import QApplication

6
resources/version.json
View File

@@ -2,9 +2,15 @@
"main_version": "4.4.1.2",
"version_info": {
"4.4.1.2": {
"新增功能": [
"AUTO_MAA 配置分享中心上线"
],
"修复BUG": [
"日志读取添加兜底机制",
"修复 QTimer.singleShot 参数问题"
],
"程序优化": [
"小文件配置信息转移至AUTO_MAA自建服务"
]
},
"4.4.1.1": {