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

feat(core): 重构日志记录，载入更多日志记录项

Browse Source
This commit is contained in:
DLmaster361
2025-07-18 18:12:47 +08:00
parent 8427bd9f6b
commit 9b492b5e0d
26 changed files with 2217 additions and 800 deletions
Download Patch File Download Diff File Expand all files Collapse all files

460
app/core/config.py
View File

@@ -25,7 +25,6 @@ v4.4
作者：DLmaster_361
"""
from loguru import logger
from PySide6.QtCore import Signal
import argparse
import sqlite3
@@ -53,6 +52,7 @@ from qfluentwidgets import (
from urllib.parse import urlparse
from typing import Union, Dict, List
from .logger import logger
from .network import Network
@@ -707,7 +707,7 @@ class GeneralSubConfig(LQConfig):
class AppConfig(GlobalConfig):
VERSION = "4.4.0.0"
VERSION = "4.4.1.1"
stage_refreshed = Signal()
PASSWORD_refreshed = Signal()
@@ -717,8 +717,8 @@ class AppConfig(GlobalConfig):
def __init__(self) -> None:
super().__init__()
self.app_path = Path(sys.argv[0]).resolve().parent # 获取软件根目录
self.app_path_sys = Path(sys.argv[0]).resolve() # 获取软件自身的路径
self.app_path = Path(sys.argv[0]).resolve().parent
self.app_path_sys = Path(sys.argv[0]).resolve()
self.log_path = self.app_path / "debug/AUTO_MAA.log"
self.database_path = self.app_path / "data/data.db"
@@ -781,7 +781,7 @@ class AppConfig(GlobalConfig):
self.init_logger()
self.check_data()
logger.info("程序初始化完成")
logger.info("程序初始化完成", module="配置管理")
def init_logger(self) -> None:
"""初始化日志记录器"""
@@ -789,7 +789,7 @@ class AppConfig(GlobalConfig):
logger.add(
sink=self.log_path,
level="DEBUG",
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <level>{message}</level>",
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[module]}</cyan> | <level>{message}</level>",
enqueue=True,
backtrace=True,
diagnose=True,
@@ -797,102 +797,28 @@ class AppConfig(GlobalConfig):
retention="1 month",
compression="zip",
)
logger.info("")
logger.info("===================================")
logger.info("AUTO_MAA 主程序")
logger.info(f"版本号： v{self.VERSION}")
logger.info(f"根目录： {self.app_path}")
logger.add(
sink=sys.stderr,
level="DEBUG",
format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | <cyan>{extra[module]}</cyan> | <level>{message}</level>",
enqueue=True,
backtrace=True,
diagnose=True,
colorize=True,
)
logger.info("", module="配置管理")
logger.info("===================================", module="配置管理")
logger.info("AUTO_MAA 主程序", module="配置管理")
logger.info(f"版本号： v{self.VERSION}", module="配置管理")
logger.info(f"根目录： {self.app_path}", module="配置管理")
logger.info(
f"运行模式： {'图形化界面' if self.args.mode == 'gui' else '命令行界面'}"
f"运行模式： {'图形化界面' if self.args.mode == 'gui' else '命令行界面'}",
module="配置管理",
)
logger.info("===================================")
logger.info("===================================", module="配置管理")
logger.info("日志记录器初始化完成")
def get_stage(self) -> None:
"""从MAA服务器获取活动关卡信息"""
network = Network.add_task(
mode="get",
url="https://api.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
)
network.loop.exec()
network_result = Network.get_result(network)
if network_result["status_code"] == 200:
stage_infos: List[Dict[str, Union[str, Dict[str, Union[str, int]]]]] = (
network_result["response_json"]["Official"]["sideStoryStage"]
)
else:
logger.warning(
f"无法从MAA服务器获取活动关卡信息:{network_result['error_message']}"
)
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_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]},
]
for day in range(0, 8):
today_stage_dict = {"value": [], "text": []}
for stage_info in 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"])
self.stage_dict[calendar.day_name[day - 1] if day > 0 else "ALL"] = {
"value": today_stage_dict["value"] + ss_stage_dict["value"],
"text": today_stage_dict["text"] + ss_stage_dict["text"],
}
self.stage_refreshed.emit()
def server_date(self) -> date:
"""获取当前的服务器日期"""
dt = datetime.now()
if dt.time() < datetime.min.time().replace(hour=4):
dt = dt - timedelta(days=1)
return dt.date()
logger.info("日志记录器初始化完成", module="配置管理")
def check_data(self) -> None:
"""检查用户数据文件并处理数据文件版本更新"""
@@ -914,11 +840,11 @@ class AppConfig(GlobalConfig):
version = cur.fetchall()
if version[0][0] != "v1.7":
logger.info("数据文件版本更新开始")
logger.info("数据文件版本更新开始", module="配置管理")
if_streaming = False
# v1.4-->v1.5
if version[0][0] == "v1.4" or if_streaming:
logger.info("数据文件版本更新：v1.4-->v1.5")
logger.info("数据文件版本更新：v1.4-->v1.5", module="配置管理")
if_streaming = True
member_dict: Dict[str, Dict[str, Union[str, Path]]] = {}
@@ -1046,7 +972,7 @@ class AppConfig(GlobalConfig):
# v1.5-->v1.6
if version[0][0] == "v1.5" or if_streaming:
logger.info("数据文件版本更新：v1.5-->v1.6")
logger.info("数据文件版本更新：v1.5-->v1.6", module="配置管理")
if_streaming = True
cur.execute("DELETE FROM version WHERE v = ?", ("v1.5",))
cur.execute("INSERT INTO version VALUES(?)", ("v1.6",))
@@ -1078,7 +1004,7 @@ class AppConfig(GlobalConfig):
pass
# v1.6-->v1.7
if version[0][0] == "v1.6" or if_streaming:
logger.info("数据文件版本更新：v1.6-->v1.7")
logger.info("数据文件版本更新：v1.6-->v1.7", module="配置管理")
if_streaming = True
if (self.app_path / "config/MaaConfig").exists():
@@ -1148,11 +1074,106 @@ class AppConfig(GlobalConfig):
cur.close()
db.close()
logger.info("数据文件版本更新完成")
logger.success("数据文件版本更新完成", module="配置管理")
def get_stage(self) -> None:
"""从MAA服务器更新活动关卡信息"""
logger.info("开始获取活动关卡信息", module="配置管理")
network = Network.add_task(
mode="get",
url="https://api.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
)
network.loop.exec()
network_result = Network.get_result(network)
if network_result["status_code"] == 200:
stage_infos: List[Dict[str, Union[str, Dict[str, Union[str, int]]]]] = (
network_result["response_json"]["Official"]["sideStoryStage"]
)
else:
logger.warning(
f"无法从MAA服务器获取活动关卡信息:{network_result['error_message']}",
module="配置管理",
)
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_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]},
]
for day in range(0, 8):
today_stage_dict = {"value": [], "text": []}
for stage_info in 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"])
self.stage_dict[calendar.day_name[day - 1] if day > 0 else "ALL"] = {
"value": today_stage_dict["value"] + ss_stage_dict["value"],
"text": today_stage_dict["text"] + ss_stage_dict["text"],
}
self.stage_refreshed.emit()
logger.success("活动关卡信息更新完成", module="配置管理")
def server_date(self) -> date:
"""
获取当前的服务器日期
:return: 当前的服务器日期
:rtype: date
"""
dt = datetime.now()
if dt.time() < datetime.min.time().replace(hour=4):
dt = dt - timedelta(days=1)
return dt.date()
def search_member(self) -> None:
"""搜索所有脚本实例"""
"""更新脚本实例配置信息"""
logger.info("开始搜索并读入脚本实例配置", module="配置管理")
self.member_dict: Dict[
str,
Dict[
@@ -1198,7 +1219,20 @@ class AppConfig(GlobalConfig):
sorted(self.member_dict.items(), key=lambda x: int(x[0][3:]))
)
logger.success(
f"脚本实例配置搜索完成，共找到 {len(self.member_dict)} 个实例",
module="配置管理",
)
def search_maa_user(self, name: str) -> None:
"""
更新指定 MAA 脚本实例的用户信息
:param name: 脚本实例名称
:type name: str
"""
logger.info(f"开始搜索并读入 MAA 脚本实例 {name} 的用户信息", module="配置管理")
user_dict: Dict[str, Dict[str, Union[Path, MaaUserConfig]]] = {}
for user_dir in (Config.member_dict[name]["Path"] / "UserData").iterdir():
@@ -1214,7 +1248,22 @@ class AppConfig(GlobalConfig):
sorted(user_dict.items(), key=lambda x: int(x[0][3:]))
)
logger.success(
f"MAA 脚本实例 {name} 的用户信息搜索完成，共找到 {len(user_dict)} 个用户",
module="配置管理",
)
def search_general_sub(self, name: str) -> None:
"""
更新指定通用脚本实例的子配置信息
:param name: 脚本实例名称
:type name: str
"""
logger.info(
f"开始搜索并读入通用脚本实例 {name} 的子配置信息", module="配置管理"
)
user_dict: Dict[str, Dict[str, Union[Path, GeneralSubConfig]]] = {}
for sub_dir in (Config.member_dict[name]["Path"] / "SubData").iterdir():
@@ -1230,8 +1279,15 @@ class AppConfig(GlobalConfig):
sorted(user_dict.items(), key=lambda x: int(x[0][3:]))
)
logger.success(
f"通用脚本实例 {name} 的子配置信息搜索完成，共找到 {len(user_dict)} 个子配置",
module="配置管理",
)
def search_plan(self) -> None:
"""搜索所有计划表"""
"""更新计划表配置信息"""
logger.info("开始搜索并读入计划表配置", module="配置管理")
self.plan_dict: Dict[str, Dict[str, Union[str, Path, MaaPlanConfig]]] = {}
if (self.app_path / "config/MaaPlanConfig").exists():
@@ -1252,8 +1308,15 @@ class AppConfig(GlobalConfig):
sorted(self.plan_dict.items(), key=lambda x: int(x[0][3:]))
)
logger.success(
f"计划表配置搜索完成，共找到 {len(self.plan_dict)} 个计划表",
module="配置管理",
)
def search_queue(self):
"""搜索所有调度队列实例"""
"""更新调度队列实例配置信息"""
logger.info("开始搜索并读入调度队列配置", module="配置管理")
self.queue_dict: Dict[str, Dict[str, Union[Path, QueueConfig]]] = {}
@@ -1273,8 +1336,20 @@ class AppConfig(GlobalConfig):
sorted(self.queue_dict.items(), key=lambda x: int(x[0][5:]))
)
logger.success(
f"调度队列配置搜索完成，共找到 {len(self.queue_dict)} 个调度队列",
module="配置管理",
)
def change_queue(self, old: str, new: str) -> None:
"""修改调度队列配置文件的队列参数"""
"""
修改调度队列配置文件的队列参数
:param old: 旧脚本名
:param new: 新脚本名
"""
logger.info(f"开始修改调度队列参数：{old} -> {new}", module="配置管理")
for queue in self.queue_dict.values():
@@ -1299,8 +1374,17 @@ class AppConfig(GlobalConfig):
if queue["Config"].get(queue["Config"].queue_Member_10) == old:
queue["Config"].set(queue["Config"].queue_Member_10, new)
logger.success(f"调度队列参数修改完成：{old} -> {new}", module="配置管理")
def change_plan(self, old: str, new: str) -> None:
"""修改脚本管理所有下属用户的计划表配置参数"""
"""
修改脚本管理所有下属用户的计划表配置参数
:param old: 旧计划表名
:param new: 新计划表名
"""
logger.info(f"开始修改计划表参数：{old} -> {new}", module="配置管理")
for member in self.member_dict.values():
@@ -1309,10 +1393,21 @@ class AppConfig(GlobalConfig):
if user["Config"].get(user["Config"].Info_StageMode) == old:
user["Config"].set(user["Config"].Info_StageMode, new)
logger.success(f"计划表参数修改完成：{old} -> {new}", module="配置管理")
def change_maa_user_info(
self, name: str, user_data: Dict[str, Dict[str, Union[str, Path, dict]]]
) -> None:
"""代理完成后保存改动的用户信息"""
"""
保存代理完成后发生改动的用户信息
:param name: 脚本实例名称
:type name: str
:param user_data: 用户信息字典，包含用户名称和对应的配置信息
:type user_data: Dict[str, Dict[str, Union[str, Path, dict]]]
"""
logger.info(f"开始保存 MAA 脚本实例 {name} 的用户信息变动", module="配置管理")
for user, info in user_data.items():
@@ -1345,10 +1440,21 @@ class AppConfig(GlobalConfig):
self.sub_info_changed.emit()
logger.success(f"MAA 脚本实例 {name} 的用户信息变动保存完成", module="配置管理")
def change_general_sub_info(
self, name: str, sub_data: Dict[str, Dict[str, Union[str, Path, dict]]]
) -> None:
"""代理完成后保存改动的配置信息"""
"""
保存代理完成后发生改动的配置信息
:param name: 脚本实例名称
:type name: str
:param sub_data: 子配置信息字典，包含子配置名称和对应的配置信息
:type sub_data: Dict[str, Dict[str, Union[str, Path, dict]]]
"""
logger.info(f"开始保存通用脚本实例 {name} 的子配置信息变动", module="配置管理")
for sub, info in sub_data.items():
@@ -1366,27 +1472,62 @@ class AppConfig(GlobalConfig):
self.sub_info_changed.emit()
logger.success(
f"通用脚本实例 {name} 的子配置信息变动保存完成", module="配置管理"
)
def set_power_sign(self, sign: str) -> None:
"""设置当前电源状态"""
"""
设置当前电源状态
:param sign: 电源状态标志
"""
self.power_sign = sign
self.power_sign_changed.emit()
logger.info(f"电源状态已更改为: {sign}", module="配置管理")
def save_history(self, key: str, content: dict) -> None:
"""保存历史记录"""
"""
保存历史记录
:param key: 调度队列的键
:type key: str
:param content: 包含时间和历史记录内容的字典
:type content: dict
"""
if key in self.queue_dict:
logger.info(f"保存调度队列 {key} 的历史记录", module="配置管理")
self.queue_dict[key]["Config"].set(
self.queue_dict[key]["Config"].Data_LastProxyTime, content["Time"]
)
self.queue_dict[key]["Config"].set(
self.queue_dict[key]["Config"].Data_LastProxyHistory, content["History"]
)
logger.success(f"调度队列 {key} 的历史记录已保存", module="配置管理")
else:
logger.warning(f"保存历史记录时未找到调度队列: {key}")
def save_maa_log(self, log_path: Path, logs: list, maa_result: str) -> bool:
"""保存MAA日志并生成对应统计数据"""
"""
保存MAA日志并生成对应统计数据
:param log_path: 日志文件保存路径
:type log_path: Path
:param logs: 日志内容列表
:type logs: list
:param maa_result: MAA 结果
:type maa_result: str
:return: 是否包含6★招募
:rtype: bool
"""
logger.info(
f"开始处理 MAA 日志，日志长度: {len(logs)}，日志标记：{maa_result}",
module="配置管理",
)
data: Dict[str, Union[str, Dict[str, Union[int, dict]]]] = {
"recruit_statistics": defaultdict(int),
@@ -1431,15 +1572,17 @@ class AppConfig(GlobalConfig):
# 查找所有Fight任务的开始和结束位置
fight_tasks = []
for i, line in enumerate(logs):
if "开始任务: Fight" in line:
if "开始任务: Fight" in line or "开始任务: 刷理智" in line:
# 查找对应的任务结束位置
end_index = -1
for j in range(i + 1, len(logs)):
if "完成任务: Fight" in logs[j]:
if "完成任务: Fight" in logs[j] or "完成任务: 刷理智" in logs[j]:
end_index = j
break
# 如果遇到新的Fight任务开始，则当前任务没有正常结束
if j < len(logs) and "开始任务: Fight" in logs[j]:
if j < len(logs) and (
"开始任务: Fight" in logs[j] or "开始任务: 刷理智" in logs[j]
):
break
# 如果找到了结束位置，记录这个任务的范围
@@ -1505,12 +1648,23 @@ class AppConfig(GlobalConfig):
with log_path.with_suffix(".json").open("w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
logger.info(f"处理完成：{log_path}")
logger.success(f"MAA 日志统计完成，日志路径：{log_path}", module="配置管理")
return if_six_star
def save_general_log(self, log_path: Path, logs: list, general_result: str) -> None:
"""保存通用日志并生成对应统计数据"""
"""
保存通用日志并生成对应统计数据
:param log_path: 日志文件保存路径
:param logs: 日志内容列表
:param general_result: 待保存的日志结果信息
"""
logger.info(
f"开始处理通用日志，日志长度: {len(logs)}，日志标记：{general_result}",
module="配置管理",
)
data: Dict[str, str] = {"general_result": general_result}
@@ -1521,10 +1675,23 @@ class AppConfig(GlobalConfig):
with log_path.with_suffix(".json").open("w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=4)
logger.info(f"处理完成：{log_path}")
logger.success(
f"通用日志统计完成，日志路径：{log_path.with_suffix('.log')}",
module="配置管理",
)
def merge_statistic_info(self, statistic_path_list: List[Path]) -> dict:
"""合并指定数据统计信息文件"""
"""
合并指定数据统计信息文件
:param statistic_path_list: 需要合并的统计信息文件路径列表
:return: 合并后的统计信息字典
"""
logger.info(
f"开始合并统计信息文件，共计 {len(statistic_path_list)} 个文件",
module="配置管理",
)
data = {"index": {}}
@@ -1591,12 +1758,28 @@ class AppConfig(GlobalConfig):
data["index"] = [data["index"][_] for _ in sorted(data["index"])]
logger.success(
f"统计信息合并完成，共计 {len(data['index'])} 条记录", module="配置管理"
)
return {k: v for k, v in data.items() if v}
def search_history(
self, mode: str, start_date: datetime, end_date: datetime
) -> dict:
"""搜索所有历史记录"""
"""
搜索指定范围内的历史记录
:param mode: 合并模式（按日合并、按周合并、按月合并）
:param start_date: 开始日期
:param end_date: 结束日期
:return: 搜索到的历史记录字典
"""
logger.info(
f"开始搜索历史记录，合并模式：{mode}，日期范围：{start_date} 至 {end_date}",
module="配置管理",
)
history_dict = {}
@@ -1638,10 +1821,43 @@ class AppConfig(GlobalConfig):
except ValueError:
logger.warning(f"非日期格式的目录: {date_folder}")
logger.success(
f"历史记录搜索完成，共计 {len(history_dict)} 条记录", module="配置管理"
)
return {
k: v
for k, v in sorted(history_dict.items(), key=lambda x: x[0], reverse=True)
}
def clean_old_history(self):
"""删除超过用户设定天数的历史记录文件（基于目录日期）"""
if self.get(self.function_HistoryRetentionTime) == 0:
logger.info("历史记录永久保留，跳过历史记录清理", module="配置管理")
return
logger.info("开始清理超过设定天数的历史记录", module="配置管理")
deleted_count = 0
for date_folder in (self.app_path / "history").iterdir():
if not date_folder.is_dir():
continue # 只处理日期文件夹
try:
# 只检查 `YYYY-MM-DD` 格式的文件夹
folder_date = datetime.strptime(date_folder.name, "%Y-%m-%d")
if datetime.now() - folder_date > timedelta(
days=self.get(self.function_HistoryRetentionTime)
):
shutil.rmtree(date_folder, ignore_errors=True)
deleted_count += 1
logger.info(f"已删除超期日志目录: {date_folder}", module="配置管理")
except ValueError:
logger.warning(f"非日期格式的目录: {date_folder}", module="配置管理")
logger.success(f"清理完成: {deleted_count} 个日期目录", module="配置管理")
Config = AppConfig()