mirror of
https://github.com/PaiGramTeam/MibooGram.git
synced 2024-11-24 17:19:31 +00:00
🐛 Fix asset service download icon
This commit is contained in:
parent
09ea3e2c37
commit
4151b80c0b
GPG Key ID:
4CBB3F4FA8C85659
@ -1,4 +1,5 @@
|
||||
"""用于下载和管理角色、武器、材料等的图标"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
@ -19,11 +20,9 @@ from typing_extensions import Self
|
||||
|
||||
from core.base_service import BaseService
|
||||
from core.config import config
|
||||
from metadata.genshin import AVATAR_DATA, HONEY_DATA, MATERIAL_DATA, NAMECARD_DATA, WEAPON_DATA
|
||||
from metadata.scripts.honey import update_honey_metadata
|
||||
from metadata.genshin import AVATAR_DATA, MATERIAL_DATA, NAMECARD_DATA, WEAPON_DATA
|
||||
from metadata.scripts.metadatas import update_metadata_from_ambr, update_metadata_from_github
|
||||
from metadata.shortname import roleToId, weaponToId
|
||||
from modules.wiki.base import HONEY_HOST
|
||||
from utils.const import AMBR_HOST, ENKA_HOST, PROJECT_ROOT
|
||||
from utils.log import logger
|
||||
from utils.typedefs import StrOrInt, StrOrURL
|
||||
@ -75,11 +74,6 @@ class _AssetsService(ABC):
|
||||
def game_name(self) -> str:
|
||||
"""游戏数据中的名称"""
|
||||
|
||||
@cached_property
|
||||
def honey_id(self) -> str:
|
||||
"""当前资源在 Honey Impact 所对应的 ID"""
|
||||
return HONEY_DATA[self.type].get(str(self.id), [""])[0]
|
||||
|
||||
@property
|
||||
def path(self) -> Path:
|
||||
"""当前资源的文件夹"""
|
||||
@ -162,12 +156,6 @@ class _AssetsService(ABC):
|
||||
"""从 enke.network 上获取目标链接"""
|
||||
yield None
|
||||
|
||||
async def _get_from_honey(self, item: str) -> AsyncIterator[str | None]:
|
||||
"""从 honey 上获取目标链接"""
|
||||
if (honey_name := self.honey_name_map.get(item, None)) is not None:
|
||||
yield HONEY_HOST.join(f"img/{honey_name}.png")
|
||||
yield HONEY_HOST.join(f"img/{honey_name}.webp")
|
||||
|
||||
async def _download_url_generator(self, item: str) -> AsyncIterator[str]:
|
||||
# 获取当前 `AssetsService` 的所有爬虫
|
||||
for func in map(lambda x: getattr(self, x), sorted(filter(lambda x: x.startswith("_get_from_"), dir(self)))):
|
||||
@ -221,11 +209,6 @@ class _AssetsService(ABC):
|
||||
def game_name_map(self) -> dict[str, str]:
|
||||
"""游戏中的图标名"""
|
||||
|
||||
@abstractmethod
|
||||
@cached_property
|
||||
def honey_name_map(self) -> dict[str, str]:
|
||||
"""来自honey的图标名"""
|
||||
|
||||
|
||||
class _AvatarAssets(_AssetsService):
|
||||
enka: EnkaCharacterAsset | None
|
||||
@ -253,10 +236,6 @@ class _AvatarAssets(_AssetsService):
|
||||
icon = avatar["icon"]
|
||||
return re.findall(r"UI_AvatarIcon_(.*)", icon)[0]
|
||||
|
||||
@cached_property
|
||||
def honey_id(self) -> str:
|
||||
return HONEY_DATA["avatar"].get(str(self.id), "")[0]
|
||||
|
||||
@cached_property
|
||||
def enka(self) -> Optional[EnkaCharacterAsset]:
|
||||
api = getattr(self, "_enka_api", None)
|
||||
@ -289,15 +268,6 @@ class _AvatarAssets(_AssetsService):
|
||||
if (item_id := self.game_name_map.get(item)) is not None:
|
||||
yield str(ENKA_HOST.join(f"ui/{item_id}.png"))
|
||||
|
||||
@cached_property
|
||||
def honey_name_map(self) -> dict[str, str]:
|
||||
return {
|
||||
"icon": f"{self.honey_id}_icon",
|
||||
"side": f"{self.honey_id}_side_icon",
|
||||
"gacha": f"{self.honey_id}_gacha_splash",
|
||||
"gacha_card": f"{self.honey_id}_gacha_card",
|
||||
}
|
||||
|
||||
@cached_property
|
||||
def game_name_map(self) -> dict[str, str]:
|
||||
return {
|
||||
@ -327,10 +297,6 @@ class _WeaponAssets(_AssetsService):
|
||||
"gacha": f"UI_Gacha_EquipIcon_{self.game_name}",
|
||||
}
|
||||
|
||||
@cached_property
|
||||
def honey_id(self) -> str:
|
||||
return f"i_n{self.id}"
|
||||
|
||||
def __call__(self, target: StrOrInt) -> Self:
|
||||
temp = target
|
||||
result = _WeaponAssets(self.client)
|
||||
@ -349,14 +315,6 @@ class _WeaponAssets(_AssetsService):
|
||||
if item in self.game_name_map:
|
||||
yield str(ENKA_HOST.join(f"ui/{self.game_name_map.get(item)}.png"))
|
||||
|
||||
@cached_property
|
||||
def honey_name_map(self) -> dict[str, str]:
|
||||
return {
|
||||
"icon": f"{self.honey_id}",
|
||||
"awaken": f"{self.honey_id}_awaken_icon",
|
||||
"gacha": f"{self.honey_id}_gacha_icon",
|
||||
}
|
||||
|
||||
|
||||
class _MaterialAssets(_AssetsService):
|
||||
@cached_property
|
||||
@ -367,10 +325,6 @@ class _MaterialAssets(_AssetsService):
|
||||
def game_name_map(self) -> dict[str, str]:
|
||||
return {"icon": f"UI_ItemIcon_{self.game_name}"}
|
||||
|
||||
@cached_property
|
||||
def honey_name_map(self) -> dict[str, str]:
|
||||
return {"icon": self.honey_id}
|
||||
|
||||
def __call__(self, target: StrOrInt) -> Self:
|
||||
temp = target
|
||||
result = _MaterialAssets(self.client)
|
||||
@ -388,10 +342,6 @@ class _MaterialAssets(_AssetsService):
|
||||
if item == "icon":
|
||||
yield str(AMBR_HOST.join(f"assets/UI/{self.game_name_map.get(item)}.png"))
|
||||
|
||||
async def _get_from_honey(self, item: str) -> AsyncIterator[str | None]:
|
||||
yield HONEY_HOST.join(f"/img/{self.honey_name_map.get(item)}.png")
|
||||
yield HONEY_HOST.join(f"/img/{self.honey_name_map.get(item)}.webp")
|
||||
|
||||
|
||||
class _ArtifactAssets(_AssetsService):
|
||||
flower: ICON_TYPE
|
||||
@ -409,10 +359,6 @@ class _ArtifactAssets(_AssetsService):
|
||||
circlet: ICON_TYPE
|
||||
"""理之冠"""
|
||||
|
||||
@cached_property
|
||||
def honey_id(self) -> str:
|
||||
return HONEY_DATA["artifact"][str(self.id)][0]
|
||||
|
||||
@cached_property
|
||||
def game_name(self) -> str:
|
||||
return f"UI_RelicIcon_{self.id}"
|
||||
@ -436,18 +382,6 @@ class _ArtifactAssets(_AssetsService):
|
||||
"circlet": f"UI_RelicIcon_{self.id}_3",
|
||||
}
|
||||
|
||||
@cached_property
|
||||
def honey_name_map(self) -> dict[str, str]:
|
||||
first_id = int(re.findall(r"\d+", HONEY_DATA["artifact"][str(self.id)][-1])[0])
|
||||
return {
|
||||
"icon": f"i_n{first_id + 30}",
|
||||
"flower": f"i_n{first_id + 30}",
|
||||
"plume": f"i_n{first_id + 10}",
|
||||
"sands": f"i_n{first_id + 40}",
|
||||
"goblet": f"i_n{first_id}",
|
||||
"circlet": f"i_n{first_id + 20}",
|
||||
}
|
||||
|
||||
|
||||
class _NamecardAssets(_AssetsService):
|
||||
enka: EnkaCharacterAsset | None
|
||||
@ -458,10 +392,6 @@ class _NamecardAssets(_AssetsService):
|
||||
profile: ICON_TYPE
|
||||
"""个人资料名片背景"""
|
||||
|
||||
@cached_property
|
||||
def honey_id(self) -> str:
|
||||
return HONEY_DATA["namecard"][str(self.id)][0]
|
||||
|
||||
@cached_property
|
||||
def game_name(self) -> str:
|
||||
return NAMECARD_DATA[str(self.id)]["icon"]
|
||||
@ -504,14 +434,6 @@ class _NamecardAssets(_AssetsService):
|
||||
"profile": NAMECARD_DATA[str(self.id)]["profile"],
|
||||
}
|
||||
|
||||
@cached_property
|
||||
def honey_name_map(self) -> dict[str, str]:
|
||||
return {
|
||||
"icon": self.honey_id,
|
||||
"navbar": f"{self.honey_id}_back",
|
||||
"profile": f"{self.honey_id}_profile",
|
||||
}
|
||||
|
||||
|
||||
class AssetsService(BaseService.Dependence):
|
||||
"""asset服务
|
||||
@ -548,7 +470,6 @@ class AssetsService(BaseService.Dependence):
|
||||
# todo 这3个任务同时异步下载
|
||||
await update_metadata_from_github(False)
|
||||
await update_metadata_from_ambr(False)
|
||||
await update_honey_metadata(False)
|
||||
logger.info("刷新元数据成功")
|
||||
|
||||
|
||||
|
||||
@ -1,197 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
from aiofiles import open as async_open
|
||||
from httpx import AsyncClient, HTTPError, Response
|
||||
|
||||
from modules.wiki.base import HONEY_HOST
|
||||
from utils.const import PROJECT_ROOT
|
||||
from utils.log import logger
|
||||
from utils.typedefs import StrOrInt
|
||||
|
||||
try:
|
||||
import ujson as jsonlib
|
||||
except ImportError:
|
||||
import json as jsonlib
|
||||
|
||||
__all__ = [
|
||||
"get_avatar_data",
|
||||
"get_artifact_data",
|
||||
"get_material_data",
|
||||
"get_namecard_data",
|
||||
"get_weapon_data",
|
||||
"update_honey_metadata",
|
||||
]
|
||||
|
||||
DATA_TYPE = Dict[StrOrInt, List[str]]
|
||||
FULL_DATA_TYPE = Dict[str, DATA_TYPE]
|
||||
|
||||
client = AsyncClient()
|
||||
|
||||
|
||||
async def request(url: str, retry: int = 5) -> Optional[Response]:
|
||||
for time in range(retry):
|
||||
try:
|
||||
return await client.get(url)
|
||||
except HTTPError:
|
||||
if time != retry - 1:
|
||||
await asyncio.sleep(1)
|
||||
continue
|
||||
return None
|
||||
except Exception as e:
|
||||
raise e
|
||||
|
||||
|
||||
async def get_avatar_data() -> DATA_TYPE:
|
||||
result = {}
|
||||
url = "https://genshin.honeyhunterworld.com/fam_chars/?lang=CHS"
|
||||
response = await request(url)
|
||||
chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0]
|
||||
json_data = jsonlib.loads(chaos_data) # 转为 json
|
||||
for data in json_data:
|
||||
cid = int("10000" + re.findall(r"\d+", data[1])[0])
|
||||
honey_id = re.findall(r"/(.*?)/", data[1])[0]
|
||||
name = re.findall(r">(.*)<", data[1])[0]
|
||||
rarity = int(re.findall(r">(\d)<", data[2])[0])
|
||||
result[cid] = [honey_id, name, rarity]
|
||||
return result
|
||||
|
||||
|
||||
async def get_weapon_data() -> DATA_TYPE:
|
||||
from modules.wiki.other import WeaponType
|
||||
|
||||
result = {}
|
||||
urls = [HONEY_HOST.join(f"fam_{i.lower()}/?lang=CHS") for i in WeaponType.__members__]
|
||||
for url in urls:
|
||||
response = await request(url)
|
||||
chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0]
|
||||
json_data = jsonlib.loads(chaos_data) # 转为 json
|
||||
for data in json_data:
|
||||
name = re.findall(r">(.*)<", data[1])[0]
|
||||
if name in ["「一心传」名刀", "石英大剑", "琥珀玥", "黑檀弓"]: # 跳过特殊的武器
|
||||
continue
|
||||
wid = int(re.findall(r"\d+", data[1])[0])
|
||||
honey_id = re.findall(r"/(.*?)/", data[1])[0]
|
||||
rarity = int(re.findall(r">(\d)<", data[2])[0])
|
||||
result[wid] = [honey_id, name, rarity]
|
||||
return result
|
||||
|
||||
|
||||
async def get_material_data() -> DATA_TYPE:
|
||||
result = {}
|
||||
|
||||
weapon = [HONEY_HOST.join(f"fam_wep_{i}/?lang=CHS") for i in ["primary", "secondary", "common"]]
|
||||
talent = [HONEY_HOST.join(f"fam_talent_{i}/?lang=CHS") for i in ["book", "boss", "common", "reward"]]
|
||||
namecard = [HONEY_HOST.join("fam_nameplate/?lang=CHS")]
|
||||
urls = weapon + talent + namecard
|
||||
|
||||
response = await request("https://api.ambr.top/v2/chs/material")
|
||||
ambr_data = jsonlib.loads(response.text)["data"]["items"]
|
||||
|
||||
for url in urls:
|
||||
response = await request(url)
|
||||
chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0]
|
||||
json_data = jsonlib.loads(chaos_data) # 转为 json
|
||||
for data in json_data:
|
||||
honey_id = re.findall(r"/(.*?)/", data[1])[0]
|
||||
name = re.findall(r">(.*)<", data[1])[0]
|
||||
rarity = int(re.findall(r">(\d)<", data[2])[0])
|
||||
mid = None
|
||||
for mid, item in ambr_data.items():
|
||||
if name == item["name"]:
|
||||
break
|
||||
mid = int(mid) or int(re.findall(r"\d+", data[1])[0])
|
||||
result[mid] = [honey_id, name, rarity]
|
||||
return result
|
||||
|
||||
|
||||
async def get_artifact_data() -> DATA_TYPE:
|
||||
async def get_first_id(_link) -> str:
|
||||
_response = await request(_link)
|
||||
_chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", _response.text)[0]
|
||||
_json_data = jsonlib.loads(_chaos_data)
|
||||
return re.findall(r"/(.*?)/", _json_data[-1][1])[0]
|
||||
|
||||
result = {}
|
||||
url = "https://genshin.honeyhunterworld.com/fam_art_set/?lang=CHS"
|
||||
|
||||
response = await request("https://api.ambr.top/v2/chs/reliquary")
|
||||
ambr_data = jsonlib.loads(response.text)["data"]["items"]
|
||||
|
||||
response = await request(url)
|
||||
chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0]
|
||||
json_data = jsonlib.loads(chaos_data) # 转为 json
|
||||
for data in json_data:
|
||||
honey_id = re.findall(r"/(.*?)/", data[1])[0]
|
||||
name = re.findall(r"alt=\"(.*?)\"", data[0])[0]
|
||||
link = HONEY_HOST.join(re.findall(r'href="(.*?)"', data[0])[0])
|
||||
first_id = await get_first_id(link)
|
||||
aid = None
|
||||
for aid, item in ambr_data.items():
|
||||
if name == item["name"]:
|
||||
break
|
||||
aid = aid or re.findall(r"\d+", data[1])[0]
|
||||
result[aid] = [honey_id, name, first_id]
|
||||
|
||||
return result
|
||||
|
||||
|
||||
async def get_namecard_data() -> DATA_TYPE:
|
||||
from metadata.genshin import NAMECARD_DATA
|
||||
|
||||
if not NAMECARD_DATA:
|
||||
# noinspection PyProtectedMember
|
||||
from metadata.genshin import Data
|
||||
from metadata.scripts.metadatas import update_metadata_from_github
|
||||
|
||||
await update_metadata_from_github()
|
||||
# noinspection PyPep8Naming
|
||||
NAMECARD_DATA = Data("namecard")
|
||||
url = HONEY_HOST.join("fam_nameplate/?lang=CHS")
|
||||
result = {}
|
||||
|
||||
response = await request(url)
|
||||
chaos_data = re.findall(r"sortable_data\.push\((.*?)\);\s*sortable_cur_page", response.text)[0]
|
||||
json_data = jsonlib.loads(chaos_data)
|
||||
for data in json_data:
|
||||
honey_id = re.findall(r"/(.*?)/", data[1])[0]
|
||||
name = re.findall(r"alt=\"(.*?)\"", data[0])[0]
|
||||
try:
|
||||
nid = [key for key, value in NAMECARD_DATA.items() if value["name"] == name][0]
|
||||
except IndexError: # 暂不支持 beta 的名片
|
||||
continue
|
||||
rarity = int(re.findall(r">(\d)<", data[2])[0])
|
||||
result[nid] = [honey_id, name, rarity]
|
||||
|
||||
return result
|
||||
|
||||
|
||||
async def update_honey_metadata(overwrite: bool = True) -> FULL_DATA_TYPE | None:
|
||||
path = PROJECT_ROOT.joinpath("metadata/data/honey.json")
|
||||
if not overwrite and path.exists():
|
||||
return
|
||||
avatar_data = await get_avatar_data()
|
||||
logger.success("Avatar data is done.")
|
||||
weapon_data = await get_weapon_data()
|
||||
logger.success("Weapon data is done.")
|
||||
material_data = await get_material_data()
|
||||
logger.success("Material data is done.")
|
||||
artifact_data = await get_artifact_data()
|
||||
logger.success("Artifact data is done.")
|
||||
namecard_data = await get_namecard_data()
|
||||
logger.success("Namecard data is done.")
|
||||
|
||||
result = {
|
||||
"avatar": avatar_data,
|
||||
"weapon": weapon_data,
|
||||
"material": material_data,
|
||||
"artifact": artifact_data,
|
||||
"namecard": namecard_data,
|
||||
}
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
async with async_open(path, mode="w", encoding="utf-8") as file:
|
||||
await file.write(jsonlib.dumps(result, ensure_ascii=False, indent=4))
|
||||
return result
|
||||
@ -2,7 +2,6 @@ from telegram import Update
|
||||
from telegram.ext import CallbackContext
|
||||
|
||||
from core.plugin import Plugin, handler
|
||||
from metadata.scripts.honey import update_honey_metadata
|
||||
from metadata.scripts.metadatas import update_metadata_from_ambr, update_metadata_from_github
|
||||
from metadata.scripts.paimon_moe import update_paimon_moe_zh
|
||||
from utils.log import logger
|
||||
@ -23,6 +22,4 @@ class MetadataPlugin(Plugin):
|
||||
await update_paimon_moe_zh()
|
||||
logger.info("正在从 ambr 上获取元数据")
|
||||
await update_metadata_from_ambr()
|
||||
logger.info("正在从 honey 上获取元数据")
|
||||
await update_honey_metadata()
|
||||
await msg.edit_text("正在刷新元数据，请耐心等待...\n完成！")
|
||||
|
||||
@ -187,7 +187,7 @@ class DailyMaterial(Plugin):
|
||||
if await aiofiles.os.path.exists(DATA_FILE_PATH):
|
||||
async with aiofiles.open(DATA_FILE_PATH, "rb") as cache:
|
||||
try:
|
||||
self.everyday_materials.parse_raw(await cache.read())
|
||||
self.everyday_materials = self.everyday_materials.parse_raw(await cache.read())
|
||||
except pydantic.ValidationError:
|
||||
await aiofiles.os.remove(DATA_FILE_PATH)
|
||||
asyncio.create_task(task_daily())
|
||||
@ -482,7 +482,9 @@ class DailyMaterial(Plugin):
|
||||
async with self.locks[0]: # 锁住第一把锁
|
||||
await self._refresh_everyday_materials()
|
||||
notice = await notice.edit_text(
|
||||
"每日素材表" + ("摘抄<b>完成！</b>" if self.everyday_materials else "坏掉了！等会它再长好了之后我再抄。。。") + "\n正搬运每日素材的图标中。。。",
|
||||
"每日素材表"
|
||||
+ ("摘抄<b>完成！</b>" if self.everyday_materials else "坏掉了！等会它再长好了之后我再抄。。。")
|
||||
+ "\n正搬运每日素材的图标中。。。",
|
||||
parse_mode=ParseMode.HTML,
|
||||
)
|
||||
time = await self._download_icon(notice)
|
||||
@ -501,10 +503,12 @@ class DailyMaterial(Plugin):
|
||||
for attempts in range(1, retry + 1):
|
||||
try:
|
||||
response = await self.client.get("https://genshin.honeyhunterworld.com/?lang=CHS")
|
||||
response.raise_for_status()
|
||||
except (HTTPError, SSLZeroReturnError):
|
||||
await asyncio.sleep(1)
|
||||
if attempts == retry:
|
||||
logger.error("每日素材刷新失败, 请稍后重试")
|
||||
return
|
||||
else:
|
||||
logger.warning("每日素材刷新失败, 正在重试第 %d 次", attempts)
|
||||
continue
|
||||
|
||||
Loading…
Reference in New Issue
Block a user