From 46ee99c5f243cc2a2aff7308e198016ff4ad89d6 Mon Sep 17 00:00:00 2001 From: DLmaster361 Date: Tue, 12 Aug 2025 01:38:32 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=85=B3=E5=8D=A1=E5=8F=B7=E6=B7=BB?= =?UTF-8?q?=E5=8A=A0=E6=9C=AC=E5=9C=B0=E7=BC=93=E5=AD=98=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/info.py | 13 +- app/core/config.py | 347 ++++++++++++++++++++++++++++----------------- main.py | 1 + 3 files changed, 220 insertions(+), 141 deletions(-) diff --git a/app/api/info.py b/app/api/info.py index 0a27b7e..00378eb 100644 --- a/app/api/info.py +++ b/app/api/info.py @@ -34,14 +34,10 @@ router = APIRouter(prefix="/api/info", tags=["信息获取"]) async def get_stage_info() -> InfoOut: try: - if_get_maa_stage, data = await Config.get_stage() + data = await Config.get_stage_info() except Exception as e: return InfoOut(code=500, status="error", message=str(e), data={}) - return InfoOut( - status="success" if if_get_maa_stage else "warning", - message="获取关卡号信息成功" if if_get_maa_stage else "未能获取活动关卡号信息", - data=data, - ) + return InfoOut(data=data) @router.post("/notice", summary="获取通知信息", response_model=InfoOut, status_code=200) @@ -75,10 +71,11 @@ async def add_overview() -> InfoOut: return InfoOut( status="success" if if_get_maa_stage else "warning", - message="获取活动关卡信息成功" if if_get_maa_stage else "未能获取活动关卡信息", + message=( + "获取活动关卡信息成功" if if_get_maa_stage else "未能获取活动关卡信息" + ), data=data, ) except Exception as e: return InfoOut(code=500, status="error", message=str(e), data={}) - diff --git a/app/core/config.py b/app/core/config.py index f4efb67..e4b7d46 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -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 diff --git a/main.py b/main.py index b8a80fa..aba7611 100644 --- a/main.py +++ b/main.py @@ -58,6 +58,7 @@ def main(): from app.core import Config, MainTimer await Config.init_config() + await Config.get_stage() main_timer = asyncio.create_task(MainTimer.second_task()) yield