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 22:12:34 +08:00
parent ff78f8eb27
commit 175efdefde
4 changed files with 125 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
app/api/info.py
View File

@@ -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})

180
app/core/config.py
View File

@@ -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(
{

2
app/core/task_manager.py
View File

@@ -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(
{

19
app/models/schema.py
View File

@@ -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表示永久保存"