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-12 01:38:32 +08:00
parent b543fedfaa
commit 46ee99c5f2
3 changed files with 220 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

347
app/core/config.py
View File

@@ -20,22 +20,105 @@
# Contact: DLmaster_361@163.com
import calendar
import re
import shutil
from collections import defaultdict
from datetime import datetime, timedelta, date, timezone
from pathlib import Path
from typing import Literal, Optional, Tuple
import asyncio
import calendar
import requests
import truststore
from pathlib import Path
from collections import defaultdict
from datetime import datetime, timedelta, date, timezone
from typing import Literal, Optional, Tuple
from app.models.ConfigBase import *
from app.utils import get_logger
logger = get_logger("配置管理")
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]},
]
MATERIALS_MAP: Dict[str, str] = {
"30165": "重相位对映体",
"30155": "烧结核凝晶",
"30145": "晶体电子单元",
"30135": "D32钢",
"30125": "双极纳米片",
"30115": "聚合剂",
"31094": "手性屈光体",
"31093": "类凝结核",
"31084": "环烃预制体",
"31083": "环烃聚质",
"31074": "固化纤维板",
"31073": "褐素纤维",
"31064": "转质盐聚块",
"31063": "转质盐组",
"31054": "切削原液",
"31053": "化合切削液",
"31044": "精炼溶剂",
"31043": "半自然溶剂",
"31034": "晶体电路",
"31033": "晶体元件",
"31024": "炽合金块",
"31023": "炽合金",
"31014": "聚合凝胶",
"31013": "凝胶",
"30074": "白马醇",
"30073": "扭转醇",
"30084": "三水锰矿",
"30083": "轻锰矿",
"30094": "五水研磨石",
"30093": "研磨石",
"30104": "RMA70-24",
"30103": "RMA70-12",
"30014": "提纯源岩",
"30013": "固源岩组",
"30012": "固源岩",
"30011": "源岩",
"30064": "改量装置",
"30063": "全新装置",
"30062": "装置",
"30061": "破损装置",
"30034": "聚酸酯块",
"30033": "聚酸酯组",
"30032": "聚酸酯",
"30031": "酯原料",
"30024": "糖聚块",
"30023": "糖组",
"30022": "糖",
"30021": "代糖",
"30044": "异铁块",
"30043": "异铁组",
"30042": "异铁",
"30041": "异铁碎片",
"30054": "酮阵列",
"30053": "酮凝集组",
"30052": "酮凝集",
"30051": "双酮",
}
class GlobalConfig(ConfigBase):
"""全局配置"""
@@ -545,37 +628,12 @@ class GeneralConfig(ConfigBase):
self.UserData = MultipleConfig([GeneralUserConfig])
class AppConfig(GlobalConfig):
VERSION = "5.0.0.1"
CLASS_BOOK = {"MAA": MaaConfig, "MaaPlan": MaaPlanConfig, "General": GeneralConfig}
CLASS_BOOK = {
"MAA": MaaConfig,
"MaaPlan": MaaPlanConfig,
"General": GeneralConfig,
}
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]},
]
class AppConfig(GlobalConfig):
VERSION = "5.0.0.1"
def __init__(self) -> None:
super().__init__(if_save_multi_config=False)
@@ -598,6 +656,9 @@ class AppConfig(GlobalConfig):
self.silence_dict: Dict[Path, datetime] = {}
self.power_sign = "NoAction"
self.if_ignore_silence: List[uuid.UUID] = []
self.last_stage_update = None
self.stage_info: Optional[Dict[str, Dict[str, List[str]]]] = None
self.temp_task: List[asyncio.Task] = []
self.ScriptConfig = MultipleConfig([MaaConfig, GeneralConfig])
self.PlanConfig = MultipleConfig([MaaPlanConfig])
@@ -621,13 +682,13 @@ class AppConfig(GlobalConfig):
logger.info("程序初始化完成")
async def add_script(
self, script: Literal["MAA", "General"]
self, script: Literal["MAA", "General"]
) -> tuple[uuid.UUID, ConfigBase]:
"""添加脚本配置"""
logger.info(f"添加脚本配置：{script}")
return await self.ScriptConfig.add(self.CLASS_BOOK[script])
return await self.ScriptConfig.add(CLASS_BOOK[script])
async def get_script(self, script_id: Optional[str]) -> tuple[list, dict]:
"""获取脚本配置"""
@@ -644,7 +705,7 @@ class AppConfig(GlobalConfig):
return list(index), data
async def update_script(
self, script_id: str, data: Dict[str, Dict[str, Any]]
self, script_id: str, data: Dict[str, Dict[str, Any]]
) -> None:
"""更新脚本配置"""
@@ -703,7 +764,7 @@ class AppConfig(GlobalConfig):
return uid, config
async def update_user(
self, script_id: str, user_id: str, data: Dict[str, Dict[str, Any]]
self, script_id: str, user_id: str, data: Dict[str, Dict[str, Any]]
) -> None:
"""更新用户配置"""
@@ -765,7 +826,7 @@ class AppConfig(GlobalConfig):
return list(index), data
async def update_queue(
self, queue_id: str, data: Dict[str, Dict[str, Any]]
self, queue_id: str, data: Dict[str, Dict[str, Any]]
) -> None:
"""更新调度队列配置"""
@@ -810,7 +871,7 @@ class AppConfig(GlobalConfig):
return uid, config
async def update_time_set(
self, queue_id: str, time_set_id: str, data: Dict[str, Dict[str, Any]]
self, queue_id: str, time_set_id: str, data: Dict[str, Dict[str, Any]]
) -> None:
"""更新时间设置配置"""
@@ -851,13 +912,13 @@ class AppConfig(GlobalConfig):
await self.QueueConfig.save()
async def add_plan(
self, script: Literal["MaaPlan"]
self, script: Literal["MaaPlan"]
) -> tuple[uuid.UUID, ConfigBase]:
"""添加计划表"""
logger.info(f"添加计划表：{script}")
return await self.PlanConfig.add(self.CLASS_BOOK[script])
return await self.PlanConfig.add(CLASS_BOOK[script])
async def get_plan(self, plan_id: Optional[str]) -> tuple[list, dict]:
"""获取计划表配置"""
@@ -917,7 +978,7 @@ class AppConfig(GlobalConfig):
return uid, config
async def update_queue_item(
self, queue_id: str, queue_item_id: str, data: Dict[str, Dict[str, Any]]
self, queue_id: str, queue_item_id: str, data: Dict[str, Dict[str, Any]]
) -> None:
"""更新队列项配置"""
@@ -996,11 +1057,68 @@ class AppConfig(GlobalConfig):
"https": self.get("Update", "ProxyAddress"),
}
async def get_stage(self) -> tuple[bool, Dict[str, Dict[str, list]]]:
"""从MAA服务器更新活动关卡信息"""
async def get_stage_info(self) -> Dict[str, Dict[str, List[str]]]:
"""获取关卡信息"""
if self.stage_info is not None:
task = asyncio.create_task(self.get_stage())
self.temp_task.append(task)
task.add_done_callback(lambda t: self.temp_task.remove(t))
return self.stage_info
else:
return await self.get_stage()
async def get_stage(self) -> Dict[str, Dict[str, List[str]]]:
"""更新活动关卡信息"""
if self.last_stage_update is not None and (
datetime.now() - self.last_stage_update
) < timedelta(hours=1):
logger.info("No need to update stage info, using cached data.")
return self.stage_info if self.stage_info is not None else {}
logger.info("开始获取活动关卡信息")
response = requests.get(
"https://api.maa.plus/MaaAssistantArknights/api/stageAndTasksUpdateTime.json",
timeout=10,
proxies=self.get_proxies(),
)
if response.status_code == 200:
remote_time_stamp = datetime.strptime(
str(response.json().get("timestamp", 20000101000000)), "%Y%m%d%H%M%S"
)
else:
logger.warning(f"无法从MAA服务器获取活动关卡时间戳:{response.text}")
remote_time_stamp = datetime.fromtimestamp(0)
if (Path.cwd() / "data/StageInfo/TimeStamp.txt").exists() and (
Path.cwd() / "data/StageInfo/StageInfo.json"
).exists():
local_time_stamp = datetime.strptime(
(Path.cwd() / "data/StageInfo/TimeStamp.txt")
.read_text(encoding="utf-8")
.strip(),
"%Y%m%d%H%M%S",
)
with (Path.cwd() / "data/StageInfo/StageInfo.json").open(
"r", encoding="utf-8"
) as f:
local_stage_info = json.load(f)
else:
local_time_stamp = datetime.fromtimestamp(0)
# 本地文件关卡信息无需更新，直接返回本地数据
if datetime.fromtimestamp(0) < remote_time_stamp <= local_time_stamp:
logger.info("使用本地关卡信息")
self.stage_info = local_stage_info
self.last_stage_update = datetime.now()
return local_stage_info
# 需要更新关卡信息
logger.info("从远端更新关卡信息")
response = requests.get(
"https://api.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
timeout=10,
@@ -1020,13 +1138,13 @@ class AppConfig(GlobalConfig):
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"
)
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"])
@@ -1037,7 +1155,7 @@ class AppConfig(GlobalConfig):
today_stage_dict = {"value": [], "text": []}
for stage_info in self.STAGE_DAILY_INFO:
for stage_info in STAGE_DAILY_INFO:
if day in stage_info["days"] or day == 0:
today_stage_dict["value"].append(stage_info["value"])
@@ -1048,76 +1166,33 @@ class AppConfig(GlobalConfig):
"text": ss_stage_dict["text"] + today_stage_dict["text"],
}
return if_get_maa_stage, stage_dict
self.stage_info = stage_dict
if if_get_maa_stage:
logger.success("成功获取远端活动关卡信息")
self.last_stage_update = datetime.now()
(Path.cwd() / "data/StageInfo").mkdir(parents=True, exist_ok=True)
(Path.cwd() / "data/StageInfo/TimeStamp.txt").write_text(
remote_time_stamp.strftime("%Y%m%d%H%M%S"), encoding="utf-8"
)
with (Path.cwd() / "data/StageInfo/StageInfo.json").open(
"w", encoding="utf-8"
) as f:
json.dump(stage_dict, f, ensure_ascii=False, indent=4)
return stage_dict
async def get_official_activity_stages(
self,
url: str = "https://api.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
timeout: int = 10,
self,
url: str = "https://api.maa.plus/MaaAssistantArknights/api/gui/StageActivity.json",
timeout: int = 10,
) -> Tuple[bool, Dict[str, List[Dict[str, str]]]]:
"""
获取 Official 区服当前开放的活动关卡（仅返回 Display/Value/Drop）。
返回:
(if_success, {"ALL": [ {"Display": "...", "Value": "...", "Drop": "..."}, ... ]})
"""
materials_map: Dict[str, str] = {
"30165": "重相位对映体",
"30155": "烧结核凝晶",
"30145": "晶体电子单元",
"30135": "D32钢",
"30125": "双极纳米片",
"30115": "聚合剂",
"31094": "手性屈光体",
"31093": "类凝结核",
"31084": "环烃预制体",
"31083": "环烃聚质",
"31074": "固化纤维板",
"31073": "褐素纤维",
"31064": "转质盐聚块",
"31063": "转质盐组",
"31054": "切削原液",
"31053": "化合切削液",
"31044": "精炼溶剂",
"31043": "半自然溶剂",
"31034": "晶体电路",
"31033": "晶体元件",
"31024": "炽合金块",
"31023": "炽合金",
"31014": "聚合凝胶",
"31013": "凝胶",
"30074": "白马醇",
"30073": "扭转醇",
"30084": "三水锰矿",
"30083": "轻锰矿",
"30094": "五水研磨石",
"30093": "研磨石",
"30104": "RMA70-24",
"30103": "RMA70-12",
"30014": "提纯源岩",
"30013": "固源岩组",
"30012": "固源岩",
"30011": "源岩",
"30064": "改量装置",
"30063": "全新装置",
"30062": "装置",
"30061": "破损装置",
"30034": "聚酸酯块",
"30033": "聚酸酯组",
"30032": "聚酸酯",
"30031": "酯原料",
"30024": "糖聚块",
"30023": "糖组",
"30022": "糖",
"30021": "代糖",
"30044": "异铁块",
"30043": "异铁组",
"30042": "异铁",
"30041": "异铁碎片",
"30054": "酮阵列",
"30053": "酮凝集组",
"30052": "酮凝集",
"30051": "双酮"
}
def normalize_drop(value: str) -> str:
# 去前后空格与常见零宽字符
@@ -1141,7 +1216,9 @@ class AppConfig(GlobalConfig):
now_utc = datetime.now(timezone.utc)
def parse_utc(dt_str: str) -> datetime:
return datetime.strptime(dt_str, "%Y/%m/%d %H:%M:%S").replace(tzinfo=timezone.utc)
return datetime.strptime(dt_str, "%Y/%m/%d %H:%M:%S").replace(
tzinfo=timezone.utc
)
results: List[Dict[str, Any]] = []
@@ -1158,17 +1235,21 @@ class AppConfig(GlobalConfig):
drop_id = normalize_drop(raw_drop)
if drop_id.isdigit():
drop_name = materials_map.get(drop_id, "未知材料")
drop_name = MATERIALS_MAP.get(drop_id, "未知材料")
else:
drop_name = "DESC:"+drop_id # 非纯数字，直接用文本.加一个DESC前缀方便前端区分
drop_name = (
"DESC:" + drop_id
) # 非纯数字，直接用文本.加一个DESC前缀方便前端区分
results.append({
"Display": s.get("Display", ""),
"Value": s.get("Value", ""),
"Drop": raw_drop,
"DropName": drop_name,
"Activity": s.get("Activity", {})
})
results.append(
{
"Display": s.get("Display", ""),
"Value": s.get("Value", ""),
"Drop": raw_drop,
"DropName": drop_name,
"Activity": s.get("Activity", {}),
}
)
return True, {"ALL": results}
@@ -1259,7 +1340,7 @@ class AppConfig(GlobalConfig):
break
# 如果遇到新的Fight任务开始，则当前任务没有正常结束
if j < len(logs) and (
"开始任务: Fight" in logs[j] or "开始任务: 刷理智" in logs[j]
"开始任务: Fight" in logs[j] or "开始任务: 刷理智" in logs[j]
):
break
@@ -1270,7 +1351,7 @@ class AppConfig(GlobalConfig):
# 处理每个Fight任务
for start_idx, end_idx in fight_tasks:
# 提取当前任务的日志
task_logs = logs[start_idx: end_idx + 1]
task_logs = logs[start_idx : end_idx + 1]
# 查找任务中的最后一次掉落统计
last_drop_stats = {}
@@ -1333,7 +1414,7 @@ class AppConfig(GlobalConfig):
return if_six_star
async def save_general_log(
self, log_path: Path, logs: list, general_result: str
self, log_path: Path, logs: list, general_result: str
) -> None:
"""
保存通用日志并生成对应统计数据
@@ -1411,7 +1492,7 @@ class AppConfig(GlobalConfig):
days=(
1
if datetime.strptime(json_file.stem, "%H-%M-%S").time()
< datetime.min.time().replace(hour=4)
< datetime.min.time().replace(hour=4)
else 0
)
)
@@ -1438,7 +1519,7 @@ class AppConfig(GlobalConfig):
return {k: v for k, v in data.items() if v}
def search_history(
self, mode: str, start_date: datetime, end_date: datetime
self, mode: str, start_date: datetime, end_date: datetime
) -> dict:
"""
搜索指定范围内的历史记录
@@ -1519,7 +1600,7 @@ class AppConfig(GlobalConfig):
# 只检查 `YYYY-MM-DD` 格式的文件夹
folder_date = datetime.strptime(date_folder.name, "%Y-%m-%d")
if datetime.now() - folder_date > timedelta(
days=self.get("Function", "HistoryRetentionTime")
days=self.get("Function", "HistoryRetentionTime")
):
shutil.rmtree(date_folder, ignore_errors=True)
deleted_count += 1