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

refactor(ui): 重构历史记录保存与载入逻辑

Browse Source
This commit is contained in:
DLmaster361
2025-07-15 16:08:11 +08:00
parent 12cf10f97a
commit 403f69df8b
4 changed files with 135 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
app/core/config.py
View File

@@ -707,7 +707,7 @@ class GeneralSubConfig(LQConfig):
class AppConfig(GlobalConfig):
VERSION = "4.4.0.6"
VERSION = "4.4.0.0"
stage_refreshed = Signal()
PASSWORD_refreshed = Signal()
@@ -1386,7 +1386,7 @@ class AppConfig(GlobalConfig):
logger.warning(f"保存历史记录时未找到调度队列: {key}")
def save_maa_log(self, log_path: Path, logs: list, maa_result: str) -> bool:
"""保存MAA日志并生成初步统计数据"""
"""保存MAA日志并生成对应统计数据"""
data: Dict[str, Union[str, Dict[str, Union[int, dict]]]] = {
"recruit_statistics": defaultdict(int),
@@ -1507,117 +1507,77 @@ class AppConfig(GlobalConfig):
logger.info(f"处理完成：{log_path}")
self.merge_maa_logs("所有项", log_path.parent)
return if_six_star
def merge_maa_logs(self, mode: str, logs_path: Union[Path, List[Path]]) -> dict:
def merge_statistic_info(self, statistic_path_list: List[Path]) -> dict:
"""合并指定数据统计信息文件"""
data = {
"recruit_statistics": defaultdict(int),
"drop_statistics": defaultdict(dict),
"maa_result": defaultdict(str),
}
data = {"index": {}}
if mode == "所有项":
logs_path_list = list(logs_path.glob("*.json"))
elif mode == "指定项":
logs_path_list = logs_path
for json_file in logs_path_list:
for json_file in statistic_path_list:
with json_file.open("r", encoding="utf-8") as f:
single_data: Dict[str, Union[str, Dict[str, Union[int, dict]]]] = (
json.load(f)
)
# 合并公招统计
for star_level, count in single_data["recruit_statistics"].items():
data["recruit_statistics"][star_level] += count
for key in single_data.keys():
# 合并掉落统计
for stage, drops in single_data["drop_statistics"].items():
if stage not in data["drop_statistics"]:
data["drop_statistics"][stage] = {} # 初始化关卡
if key not in data:
data[key] = {}
for item, count in drops.items():
# 合并公招统计
if key == "recruit_statistics":
if item in data["drop_statistics"][stage]:
data["drop_statistics"][stage][item] += count
else:
data["drop_statistics"][stage][item] = count
for star_level, count in single_data[key].items():
if star_level not in data[key]:
data[key][star_level] = 0
data[key][star_level] += count
# 合并MAA结果
data["maa_result"][json_file.stem.replace("-", ":")] = single_data[
"maa_result"
]
# 合并掉落统计
if key == "drop_statistics":
# 生成汇总 JSON 文件
if mode == "所有项":
for stage, drops in single_data[key].items():
if stage not in data[key]:
data[key][stage] = {} # 初始化关卡
with logs_path.with_suffix(".json").open("w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
for item, count in drops.items():
logger.info(f"统计完成：{logs_path.with_suffix('.json')}")
if item not in data[key][stage]:
data[key][stage][item] = 0
data[key][stage][item] += count
return data
# 录入MAA结果
if key == "maa_result":
def load_maa_logs(
self, mode: str, json_path: Path
) -> Dict[str, Union[str, list, Dict[str, list]]]:
"""加载MAA日志统计信息"""
actual_date = datetime.strptime(
f"{json_file.parent.parent.name} {json_file.stem}",
"%Y-%m-%d %H-%M-%S",
) + timedelta(
days=(
1
if datetime.strptime(json_file.stem, "%H-%M-%S").time()
< datetime.min.time().replace(hour=4)
else 0
)
)
if mode == "总览":
if single_data[key] != "Success!":
if "error_info" not in data:
data["error_info"] = {}
data["error_info"][actual_date.strftime("%d日 %H:%M:%S")] = (
single_data[key]
)
with json_path.open("r", encoding="utf-8") as f:
info: Dict[str, Dict[str, Union[int, dict]]] = json.load(f)
data["index"][actual_date] = [
actual_date.strftime("%d日 %H:%M:%S"),
("完成" if single_data["maa_result"] == "Success!" else "异常"),
json_file,
]
data = {}
# 4点前的记录放在当日最后
sorted_maa_result = sorted(
info["maa_result"].items(),
key=lambda x: (
(
1
if datetime.strptime(x[0], "%H:%M:%S").time()
< datetime.min.time().replace(hour=4)
else 0
),
datetime.strptime(x[0], "%H:%M:%S"),
),
)
data["条目索引"] = [
[k, "完成" if v == "Success!" else "异常"] for k, v in sorted_maa_result
]
data["条目索引"].insert(0, ["数据总览", "运行"])
data["统计数据"] = {"公招统计": list(info["recruit_statistics"].items())}
data["index"] = [data["index"][_] for _ in sorted(data["index"])]
for game_id, drops in info["drop_statistics"].items():
data["统计数据"][f"掉落统计：{game_id}"] = list(drops.items())
data["统计数据"]["报错汇总"] = [
[k, v] for k, v in info["maa_result"].items() if v != "Success!"
]
elif mode == "单项":
with json_path.open("r", encoding="utf-8") as f:
info: Dict[str, Union[str, Dict[str, Union[int, dict]]]] = json.load(f)
data = {}
data["统计数据"] = {"公招统计": list(info["recruit_statistics"].items())}
for game_id, drops in info["drop_statistics"].items():
data["统计数据"][f"掉落统计：{game_id}"] = list(drops.items())
with json_path.with_suffix(".log").open("r", encoding="utf-8") as f:
log = f.read()
data["日志信息"] = log
return data
return {k: v for k, v in data.items() if v}
def search_history(
self, mode: str, start_date: datetime, end_date: datetime
@@ -1638,43 +1598,28 @@ class AppConfig(GlobalConfig):
continue # 只统计在范围内的日期
if mode == "按日合并":
history_dict[date.strftime("%Y年 %m月 %d日")] = list(
date_folder.glob("*.json")
)
date_name = date.strftime("%Y年 %m月 %d日")
elif mode == "按周合并":
year, week, _ = date.isocalendar()
if f"{year}年 第{week}周" not in history_dict:
history_dict[f"{year}年 第{week}周"] = {}
for user in date_folder.glob("*.json"):
if user.stem not in history_dict[f"{year}年 第{week}周"]:
history_dict[f"{year}年 第{week}周"][user.stem] = list(
user.with_suffix("").glob("*.json")
)
else:
history_dict[f"{year}年 第{week}周"][user.stem] += list(
user.with_suffix("").glob("*.json")
)
date_name = f"{year}年 第{week}周"
elif mode == "按月合并":
date_name = date.strftime("%Y年 %m月")
if date.strftime("%Y年 %m月") not in history_dict:
history_dict[date.strftime("%Y年 %m月")] = {}
if date_name not in history_dict:
history_dict[date_name] = {}
for user in date_folder.glob("*.json"):
for user_folder in date_folder.iterdir():
if not user_folder.is_dir():
continue # 只处理用户文件夹
if user.stem not in history_dict[date.strftime("%Y年 %m月")]:
history_dict[date.strftime("%Y年 %m月")][user.stem] = list(
user.with_suffix("").glob("*.json")
)
else:
history_dict[date.strftime("%Y年 %m月")][user.stem] += list(
user.with_suffix("").glob("*.json")
)
if user_folder.stem not in history_dict[date_name]:
history_dict[date_name][user_folder.stem] = list(
user_folder.with_suffix("").glob("*.json")
)
else:
history_dict[date_name][user_folder.stem] += list(
user_folder.with_suffix("").glob("*.json")
)
except ValueError:
logger.warning(f"非日期格式的目录: {date_folder}")

2
app/models/MAA.py
View File

@@ -701,7 +701,7 @@ class MaaManager(QObject):
logger.info(f"{self.name} | 更新动作结束")
# 发送统计信息
statistics = Config.merge_maa_logs("指定项", user_logs_list)
statistics = Config.merge_statistic_info(user_logs_list)
statistics["user_index"] = user[2]
statistics["user_info"] = user[0]
statistics["start_time"] = user_start_time.strftime("%Y-%m-%d %H:%M:%S")

160
app/ui/history.py
View File

@@ -48,7 +48,7 @@ import subprocess
from datetime import datetime, timedelta
from functools import partial
from pathlib import Path
from typing import Union, List, Dict
from typing import List, Dict
from app.core import Config, SoundPlayer
@@ -100,9 +100,9 @@ class History(QWidget):
datetime(end_date.year(), end_date.month(), end_date.day()),
)
for date, user in history_dict.items():
for date, user_dict in history_dict.items():
self.history_card_list.append(self.HistoryCard(mode, date, user, self))
self.history_card_list.append(self.HistoryCard(date, user_dict, self))
self.content_layout.addWidget(self.history_card_list[-1])
self.content_layout.addStretch(1)
@@ -172,14 +172,9 @@ class History(QWidget):
self.search.clicked.emit()
class HistoryCard(QuickExpandGroupCard):
"""历史记录卡片"""
def __init__(
self,
mode: str,
date: str,
user: Union[List[Path], Dict[str, List[Path]]],
parent=None,
):
def __init__(self, date: str, user_dict: Dict[str, List[Path]], parent=None):
super().__init__(
FluentIcon.HISTORY, date, f"{date}的历史运行记录与统计信息", parent
)
@@ -192,64 +187,28 @@ class History(QWidget):
self.user_history_card_list = []
if mode == "按日合并":
for user_path in user:
self.user_history_card_list.append(
self.UserHistoryCard(mode, user_path.stem, user_path, self)
)
Layout.addWidget(self.user_history_card_list[-1])
elif mode in ["按周合并", "按月合并"]:
for user, info in user.items():
self.user_history_card_list.append(
self.UserHistoryCard(mode, user, info, self)
)
Layout.addWidget(self.user_history_card_list[-1])
for user, info in user_dict.items():
self.user_history_card_list.append(
self.UserHistoryCard(user, info, self)
)
Layout.addWidget(self.user_history_card_list[-1])
class UserHistoryCard(HeaderCardWidget):
"""用户历史记录卡片"""
def __init__(
self,
mode: str,
name: str,
user_history: Union[Path, List[Path]],
parent=None,
):
def __init__(self, name: str, user_history: List[Path], parent=None):
super().__init__(parent)
self.setTitle(name)
if mode == "按日合并":
self.user_history = user_history
self.user_history_path = user_history
self.main_history = Config.load_maa_logs("总览", user_history)
self.index_card = self.IndexCard(
self.main_history["条目索引"], self
)
self.index_card.index_changed.connect(self.update_info)
self.viewLayout.addWidget(self.index_card)
elif mode in ["按周合并", "按月合并"]:
history = Config.merge_maa_logs("指定项", user_history)
self.main_history = {}
self.main_history["统计数据"] = {
"公招统计": list(history["recruit_statistics"].items())
}
for game_id, drops in history["drop_statistics"].items():
self.main_history["统计数据"][f"掉落统计：{game_id}"] = list(
drops.items()
)
self.index_card = self.IndexCard(self.user_history, self)
self.index_card.index_changed.connect(self.update_info)
self.statistics_card = QHBoxLayout()
self.log_card = self.LogCard(self)
self.viewLayout.addWidget(self.index_card)
self.viewLayout.addLayout(self.statistics_card)
self.viewLayout.addWidget(self.log_card)
self.viewLayout.setContentsMargins(0, 0, 0, 0)
@@ -259,19 +218,45 @@ class History(QWidget):
self.update_info("数据总览")
def get_statistics(self, mode: str) -> dict:
"""生成GUI相应结构化统计数据"""
history_info = Config.merge_statistic_info(
self.user_history if mode == "数据总览" else [Path(mode)]
)
statistics_info = {}
if "recruit_statistics" in history_info:
statistics_info["公招统计"] = list(
history_info["recruit_statistics"].items()
)
if "drop_statistics" in history_info:
for game_id, drops in history_info["drop_statistics"].items():
statistics_info[f"掉落统计：{game_id}"] = list(drops.items())
if mode == "数据总览" and "error_info" in history_info:
statistics_info["报错汇总"] = list(
history_info["error_info"].items()
)
return statistics_info
def update_info(self, index: str) -> None:
"""更新信息"""
# 移除已有统计信息UI组件
while self.statistics_card.count() > 0:
item = self.statistics_card.takeAt(0)
if item.spacerItem():
self.statistics_card.removeItem(item.spacerItem())
elif item.widget():
item.widget().deleteLater()
if index == "数据总览":
while self.statistics_card.count() > 0:
item = self.statistics_card.takeAt(0)
if item.spacerItem():
self.statistics_card.removeItem(item.spacerItem())
elif item.widget():
item.widget().deleteLater()
for name, item_list in self.main_history["统计数据"].items():
for name, item_list in self.get_statistics("数据总览").items():
statistics_card = self.StatisticsCard(name, item_list, self)
self.statistics_card.addWidget(statistics_card)
@@ -280,44 +265,24 @@ class History(QWidget):
else:
single_history = Config.load_maa_logs(
"单项",
self.user_history_path.with_suffix("")
/ f"{index.replace(":","-")}.json",
)
while self.statistics_card.count() > 0:
item = self.statistics_card.takeAt(0)
if item.spacerItem():
self.statistics_card.removeItem(item.spacerItem())
elif item.widget():
item.widget().deleteLater()
for name, item_list in single_history["统计数据"].items():
single_history = self.get_statistics(index)
log_path = Path(index).with_suffix(".log")
for name, item_list in single_history.items():
statistics_card = self.StatisticsCard(name, item_list, self)
self.statistics_card.addWidget(statistics_card)
self.log_card.text.setText(single_history["日志信息"])
with log_path.open("r", encoding="utf-8") as f:
log = f.read()
self.log_card.text.setText(log)
self.log_card.open_file.clicked.disconnect()
self.log_card.open_file.clicked.connect(
lambda: os.startfile(
self.user_history_path.with_suffix("")
/ f"{index.replace(":","-")}.log"
)
lambda: os.startfile(log_path)
)
self.log_card.open_dir.clicked.disconnect()
self.log_card.open_dir.clicked.connect(
lambda: subprocess.Popen(
[
"explorer",
"/select,",
str(
self.user_history_path.with_suffix("")
/ f"{index.replace(":","-")}.log"
),
]
)
lambda: subprocess.Popen(["explorer", "/select,", log_path])
)
self.log_card.show()
@@ -329,7 +294,7 @@ class History(QWidget):
index_changed = Signal(str)
def __init__(self, index_list: list, parent=None):
def __init__(self, history_list: List[Path], parent=None):
super().__init__(parent)
self.setTitle("记录条目")
@@ -339,11 +304,14 @@ class History(QWidget):
self.index_cards: List[StatefulItemCard] = []
index_list = Config.merge_statistic_info(history_list)["index"]
index_list.insert(0, ["数据总览", "运行", "数据总览"])
for index in index_list:
self.index_cards.append(StatefulItemCard(index))
self.index_cards.append(StatefulItemCard(index[:2]))
self.index_cards[-1].clicked.connect(
partial(self.index_changed.emit, index[0])
partial(self.index_changed.emit, str(index[2]))
)
self.Layout.addWidget(self.index_cards[-1])

7
resources/version.json
View File

@@ -1,9 +1,12 @@
{
"main_version": "4.4.0.6",
"main_version": "4.4.0.0",
"version_info": {
"4.4.0.6": {
"4.4.0.0": {
"修复BUG": [
"信任系统证书，并添加网络代理地址配置项 #50"
],
"程序优化": [
"重构历史记录保存与载入逻辑"
]
},
"4.4.0.5": {