From 175efdefdeae8afc3e51d2a3b3eb0ac9eaba0430 Mon Sep 17 00:00:00 2001 From: DLmaster361 Date: Tue, 12 Aug 2025 22:12:34 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=85=B3=E5=8D=A1=E5=8F=B7=E7=9B=B8?= =?UTF-8?q?=E5=85=B3=E6=8E=A5=E5=8F=A3=E9=80=82=E9=85=8D=E4=B8=8B=E6=8B=89?= =?UTF-8?q?=E6=A1=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/info.py | 33 ++++--- app/core/config.py | 180 +++++++++++++++++++-------------------- app/core/task_manager.py | 2 + app/models/schema.py | 19 ++++- 4 files changed, 125 insertions(+), 109 deletions(-) diff --git a/app/api/info.py b/app/api/info.py index a2fa90a..bdbebef 100644 --- a/app/api/info.py +++ b/app/api/info.py @@ -29,15 +29,21 @@ router = APIRouter(prefix="/api/info", tags=["信息获取"]) @router.post( - "/stage", summary="获取关卡号信息", response_model=InfoOut, status_code=200 + "/combox/stage", + summary="获取关卡号下拉框信息", + response_model=ComboBoxOut, + status_code=200, ) -async def get_stage_info() -> InfoOut: +async def get_stage_combox( + stage: GetStageIn = Body(..., description="关卡号类型") +) -> ComboBoxOut: try: - data = await Config.get_stage_info() + raw_data = await Config.get_stage_info(stage.type) + data = [ComboBoxItem(**item) for item in raw_data] if raw_data else [] except Exception as e: - return InfoOut(code=500, status="error", message=str(e), data={}) - return InfoOut(data=data) + return ComboBoxOut(code=500, status="error", message=str(e), data=[]) + return ComboBoxOut(data=data) @router.post( @@ -49,7 +55,8 @@ async def get_stage_info() -> InfoOut: async def get_script_combox() -> ComboBoxOut: try: - data: List[ComboBoxItem] = await Config.get_script_combox() + raw_data = await Config.get_script_combox() + data = [ComboBoxItem(**item) for item in raw_data] if raw_data else [] except Exception as e: return ComboBoxOut(code=500, status="error", message=str(e), data=[]) return ComboBoxOut(data=data) @@ -82,15 +89,7 @@ async def get_apps_info() -> InfoOut: ) async def add_overview() -> InfoOut: try: - if_get_maa_stage, data = await Config.get_official_activity_stages() - - return InfoOut( - status="success" if if_get_maa_stage else "warning", - message=( - "获取活动关卡信息成功" if if_get_maa_stage else "未能获取活动关卡信息" - ), - data=data, - ) - + data = await Config.get_stage_info("Info") except Exception as e: - return InfoOut(code=500, status="error", message=str(e), data={}) + return InfoOut(code=500, status="error", message=str(e), data={"ALL": []}) + return InfoOut(data={"ALL": data}) diff --git a/app/core/config.py b/app/core/config.py index 5c1b7a9..bf790cc 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -220,7 +220,7 @@ class QueueConfig(ConfigBase): def __init__(self) -> None: super().__init__() - self.Info_Name = ConfigItem("Info", "Name", "") + self.Info_Name = ConfigItem("Info", "Name", "新队列") self.Info_TimeEnabled = ConfigItem( "Info", "TimeEnabled", False, BoolValidator() ) @@ -403,7 +403,7 @@ class MaaConfig(ConfigBase): def __init__(self) -> None: super().__init__() - self.Info_Name = ConfigItem("Info", "Name", "") + self.Info_Name = ConfigItem("Info", "Name", "新 MAA 脚本") self.Info_Path = ConfigItem("Info", "Path", ".", FolderValidator()) self.Run_TaskTransitionMethod = ConfigItem( @@ -440,7 +440,7 @@ class MaaPlanConfig(ConfigBase): def __init__(self) -> None: super().__init__() - self.Info_Name = ConfigItem("Info", "Name", "") + self.Info_Name = ConfigItem("Info", "Name", "新 MAA 计划表") self.Info_Mode = ConfigItem( "Info", "Mode", "ALL", OptionsValidator(["ALL", "Weekly"]) ) @@ -511,12 +511,12 @@ class MaaPlanConfig(ConfigBase): class GeneralUserConfig(ConfigBase): - """通用子配置""" + """通用脚本用户配置""" def __init__(self) -> None: super().__init__() - self.Info_Name = ConfigItem("Info", "Name", "新配置") + self.Info_Name = ConfigItem("Info", "Name", "新用户") self.Info_Status = ConfigItem("Info", "Status", True, BoolValidator()) self.Info_RemainedDay = ConfigItem( "Info", "RemainedDay", -1, RangeValidator(-1, 1024) @@ -566,7 +566,7 @@ class GeneralConfig(ConfigBase): def __init__(self) -> None: super().__init__() - self.Info_Name = ConfigItem("Info", "Name", "") + self.Info_Name = ConfigItem("Info", "Name", "新通用脚本") self.Info_RootPath = ConfigItem("Info", "RootPath", ".", FileValidator()) self.Script_ScriptPath = ConfigItem( @@ -656,7 +656,7 @@ class AppConfig(GlobalConfig): 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.stage_info: Optional[Dict[str, List[Dict[str, str]]]] = None self.temp_task: List[asyncio.Task] = [] self.ScriptConfig = MultipleConfig([MaaConfig, GeneralConfig]) @@ -1058,18 +1058,39 @@ class AppConfig(GlobalConfig): "https": self.get("Update", "ProxyAddress"), } - async def get_stage_info(self) -> Dict[str, Dict[str, List[str]]]: + async def get_stage_info( + self, + type: Literal[ + "Today", + "ALL", + "Monday", + "Tuesday", + "Wednesday", + "Thursday", + "Friday", + "Saturday", + "Sunday", + "Info", + ], + ): """获取关卡信息""" + if type == "Today": + dt = self.server_date() + index = dt.strftime("%A") + else: + index = type + 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() + await self.get_stage() - async def get_stage(self) -> Dict[str, Dict[str, List[str]]]: + return self.stage_info.get(index, []) if self.stage_info is not None else [] + + async def get_stage(self) -> Optional[Dict[str, List[Dict[str, str]]]]: """更新活动关卡信息""" if self.last_stage_update is not None and ( @@ -1132,7 +1153,9 @@ class AppConfig(GlobalConfig): proxies=self.get_proxies(), ) if response.status_code == 200: - stage_infos = response.json()["Official"]["sideStoryStage"] + stage_infos = ( + response.json().get("Official", {}).get("sideStoryStage", []) + ) if_get_maa_stage = True else: logger.warning(f"无法从MAA服务器获取活动关卡信息:{response.text}") @@ -1143,86 +1166,12 @@ class AppConfig(GlobalConfig): if_get_maa_stage = False 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_dict = {} - - 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"]) - - stage_dict[calendar.day_name[day - 1] if day > 0 else "ALL"] = { - "value": ss_stage_dict["value"] + today_stage_dict["value"], - "text": ss_stage_dict["text"] + today_stage_dict["text"], - } - - 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, - ) -> Tuple[bool, Dict[str, List[Dict[str, str]]]]: - """ - 获取 Official 区服当前开放的活动关卡（仅返回 Display/Value/Drop）。 - 返回: - (if_success, {"ALL": [ {"Display": "...", "Value": "...", "Drop": "..."}, ... ]}) - """ - def normalize_drop(value: str) -> str: # 去前后空格与常见零宽字符 s = str(value).strip() s = re.sub(r"[\u200b\u200c\u200d\ufeff]", "", s) return s - try: - resp = requests.get(url, timeout=timeout, proxies=self.get_proxies()) - except Exception: - return False, {"ALL": []} - - if resp.status_code != 200: - return False, {"ALL": []} - - try: - payload = resp.json() - except Exception: - return False, {"ALL": []} - now_utc = datetime.now(timezone.utc) def parse_utc(dt_str: str) -> datetime: @@ -1230,9 +1179,9 @@ class AppConfig(GlobalConfig): tzinfo=timezone.utc ) - results: List[Dict[str, Any]] = [] + side_story_info: List[Dict[str, Any]] = [] - for s in payload.get("Official", {}).get("sideStoryStage", []): + for s in stage_infos: act = s.get("Activity", {}) or {} try: start_utc = parse_utc(act["UtcStartTime"]) @@ -1251,7 +1200,7 @@ class AppConfig(GlobalConfig): "DESC:" + drop_id ) # 非纯数字，直接用文本.加一个DESC前缀方便前端区分 - results.append( + side_story_info.append( { "Display": s.get("Display", ""), "Value": s.get("Value", ""), @@ -1261,13 +1210,62 @@ class AppConfig(GlobalConfig): } ) - return True, {"ALL": results} + side_story_stage = [] + + 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" + ) + ): + side_story_stage.append( + {"label": stage_info["Value"], "value": stage_info["Value"]} + ) + + self.stage_info = {} + + for day in range(0, 8): + + today_stage = [] + + for stage_info in STAGE_DAILY_INFO: + + if day in stage_info["days"] or day == 0: + today_stage.append( + {"label": stage_info["text"], "value": stage_info["value"]} + ) + + self.stage_info[calendar.day_name[day - 1] if day > 0 else "ALL"] = ( + side_story_stage + today_stage + ) + + self.stage_info["Info"] = side_story_info + + 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(self.stage_info, f, ensure_ascii=False, indent=4) + + return self.stage_info async def get_script_combox(self): """获取脚本下拉框信息""" logger.info("Getting script combo box information...") - data = [] + data = [{"label": "未选择", "value": None}] for uid, script in self.ScriptConfig.items(): data.append( { diff --git a/app/core/task_manager.py b/app/core/task_manager.py index 5ae8507..2e5e67e 100644 --- a/app/core/task_manager.py +++ b/app/core/task_manager.py @@ -157,6 +157,8 @@ class _TaskManager: task_list = [] for queue_item in queue.QueueItem.values(): + if queue_item.get("Info", "ScriptId") is None: + continue uid = uuid.UUID(queue_item.get("Info", "ScriptId")) task_list.append( { diff --git a/app/models/schema.py b/app/models/schema.py index bd89ca5..50d80af 100644 --- a/app/models/schema.py +++ b/app/models/schema.py @@ -36,13 +36,30 @@ class InfoOut(OutBase): class ComboBoxItem(BaseModel): label: str = Field(..., description="展示值") - value: str = Field(..., description="实际值") + value: Optional[str] = Field(..., description="实际值") class ComboBoxOut(OutBase): data: List[ComboBoxItem] = Field(..., description="下拉框选项") +class GetStageIn(BaseModel): + type: Literal[ + "Today", + "ALL", + "Monday", + "Tuesday", + "Wednesday", + "Thursday", + "Friday", + "Saturday", + "Sunday", + ] = Field( + ..., + description="选择的日期类型, Today为当天, ALL为包含当天未开放关卡在内的所有项", + ) + + class GlobalConfig_Function(BaseModel): HistoryRetentionTime: Optional[Literal[7, 15, 30, 60, 90, 180, 365, 0]] = Field( None, description="历史记录保留时间, 0表示永久保存"